+type RPCStreamReqResp struct {
+ ID uint32
+ Msg api.Message
+}
+
+func (s *BinapiRPC) NewAPIStream(req RPCStreamReqResp, resp *RPCStreamReqResp) error {
+ if !s.serviceAvailable() {
+ log.Println(binapiErrorMsg)
+ return errors.New("server does not support RPC calls at this time, try again later")
+ }
+ log.Debugf("BinapiRPC.NewAPIStream - REQ: %#v", req)
+
+ stream, err := s.binapiConn.NewStream(context.Background())
+ if err != nil {
+ return err
+ }
+
+ if s.streams == nil {
+ s.streams = make(map[uint32]api.Stream)
+ }
+
+ s.streamsLock.Lock()
+ s.maxStreamID++
+ s.streams[s.maxStreamID] = stream
+ resp.ID = s.maxStreamID
+ s.streamsLock.Unlock()
+
+ return nil
+}
+
+func (s *BinapiRPC) SendMessage(req RPCStreamReqResp, resp *RPCStreamReqResp) error {
+ if !s.serviceAvailable() {
+ log.Println(binapiErrorMsg)
+ return errors.New("server does not support RPC calls at this time, try again later")
+ }
+ log.Debugf("BinapiRPC.SendMessage - REQ: %#v", req)
+
+ stream, err := s.getStream(req.ID)
+ if err != nil {
+ return err
+ }
+
+ return stream.SendMsg(req.Msg)
+}
+
+func (s *BinapiRPC) ReceiveMessage(req RPCStreamReqResp, resp *RPCStreamReqResp) error {
+ if !s.serviceAvailable() {
+ log.Println(binapiErrorMsg)
+ return errors.New("server does not support RPC calls at this time, try again later")
+ }
+ log.Debugf("BinapiRPC.ReceiveMessage - REQ: %#v", req)
+
+ stream, err := s.getStream(req.ID)
+ if err != nil {
+ return err
+ }
+
+ resp.Msg, err = stream.RecvMsg()
+ return err
+}
+
+func (s *BinapiRPC) CloseStream(req RPCStreamReqResp, resp *RPCStreamReqResp) error {
+ if !s.serviceAvailable() {
+ log.Println(binapiErrorMsg)
+ return errors.New("server does not support RPC calls at this time, try again later")
+ }
+ log.Debugf("BinapiRPC.CloseStream - REQ: %#v", req)
+
+ stream, err := s.getStream(req.ID)
+ if err != nil {
+ return err
+ }
+
+ s.streamsLock.Lock()
+ delete(s.streams, req.ID)
+ s.streamsLock.Unlock()
+
+ return stream.Close()
+}
+
+func (s *BinapiRPC) getStream(id uint32) (api.Stream, error) {
+ s.streamsLock.Lock()
+ stream := s.streams[id]
+ s.streamsLock.Unlock()
+
+ if stream == nil || reflect.ValueOf(stream).IsNil() {
+ s.streamsLock.Lock()
+ // delete the stream in case it is still in the map
+ delete(s.streams, id)
+ s.streamsLock.Unlock()
+ return nil, errors.New("BinapiRPC stream closed")
+ }
+ return stream, nil
+}
+