grpc: allocate a stream pool

SynapticDreams · Apr 12, 2022 · 8705f38 · 8705f38
1 parent fb5703b
commit 8705f38
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/pkg/grpc/server.go b/pkg/grpc/server.go
@@ -58,6 +58,7 @@ func (s *Server) Subscribe(request *pb.SubscribeRequest, server pb.MarketDataSer
 		}
 	}
 
+	streamPool := map[string]types.Stream{}
 	for sessionName, subs := range exchangeSubscriptions {
 		if session, ok := s.Environ.Session(sessionName); ok {
 			stream := session.Exchange.NewStream()
@@ -83,10 +84,15 @@ func (s *Server) Subscribe(request *pb.SubscribeRequest, server pb.MarketDataSer
 					log.WithError(err).Error("grpc stream send error")
 				}
 			})
-			go stream.Connect(server.Context())
+			streamPool[sessionName] = stream
 		}
 	}
 
+	for sessionName, stream := range streamPool {
+		log.Infof("connecting stream %s", sessionName)
+		go stream.Connect(server.Context())
+	}
+
 	return nil
 }