+// RPCStream is a stream for forwarding requests to BinapiRPC's stream.
+type RPCStream struct {
+ rpc *rpc.Client
+ id uint32
+}
+
+func (s *RPCStream) SendMsg(msg api.Message) error {
+ req := RPCStreamReqResp{
+ ID: s.id,
+ Msg: msg,
+ }
+ resp := RPCStreamReqResp{}
+ err := s.rpc.Call("BinapiRPC.SendMessage", req, &resp)
+ if err != nil {
+ return fmt.Errorf("RPC SendMessage call failed: %v", err)
+ }
+ return nil
+}
+
+func (s *RPCStream) RecvMsg() (api.Message, error) {
+ req := RPCStreamReqResp{
+ ID: s.id,
+ }
+ resp := RPCStreamReqResp{}
+ err := s.rpc.Call("BinapiRPC.ReceiveMessage", req, &resp)
+ if err != nil {
+ return nil, fmt.Errorf("RPC ReceiveMessage call failed: %v", err)
+ }
+ return resp.Msg, nil
+}
+
+func (s *RPCStream) Close() error {
+ req := RPCStreamReqResp{
+ ID: s.id,
+ }
+ resp := RPCStreamReqResp{}
+ err := s.rpc.Call("BinapiRPC.CloseStream", req, &resp)
+ if err != nil {
+ return fmt.Errorf("RPC CloseStream call failed: %v", err)
+ }
+ return nil
+}
+
+func (b *BinapiClient) NewStream(_ context.Context, _ ...api.StreamOption) (api.Stream, error) {
+ stream := &RPCStream{
+ rpc: b.rpc,
+ }
+ req := RPCStreamReqResp{}
+ resp := RPCStreamReqResp{}
+ err := stream.rpc.Call("BinapiRPC.NewAPIStream", req, &resp)
+ if err != nil {
+ return nil, fmt.Errorf("RPC NewAPIStream call failed: %v", err)
+ }
+ stream.id = resp.ID
+ return stream, err
+}
+
+func (b *BinapiClient) Invoke(_ context.Context, request api.Message, reply api.Message) error {
+ return invokeInternal(b.rpc, request, reply, b.timeout)
+}
+