polling goroutine permanant

dydxprotocol · May 24, 2024 · d69c744 · d69c744
1 parent 1066d2b
commit d69c744
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/protocol/streaming/grpc/grpc_streaming_manager.go b/protocol/streaming/grpc/grpc_streaming_manager.go
@@ -68,11 +68,13 @@ func NewGrpcStreamingManager(
 
 	// Worker goroutine to consistently read from channel and send out updates
 	go func() {
-		for internalResponse := range grpcStreamingManager.updateBuffer {
-			grpcStreamingManager.logger.Info("start polling a response", "len", len(grpcStreamingManager.updateBuffer))
-			grpcStreamingManager.sendUpdateResponse(internalResponse)
-			grpcStreamingManager.logger.Info("finish polling a response", "len", len(grpcStreamingManager.updateBuffer))
+		for {
+			for internalResponse := range grpcStreamingManager.updateBuffer {
+				grpcStreamingManager.logger.Info("start polling a response", "len", len(grpcStreamingManager.updateBuffer))
+				grpcStreamingManager.sendUpdateResponse(internalResponse)
+				grpcStreamingManager.logger.Info("finish polling a response", "len", len(grpcStreamingManager.updateBuffer))
 
+			}
 		}
 	}()
 
@@ -268,6 +270,7 @@ func (sm *GrpcStreamingManagerImpl) mustEnqueueOrderbookUpdate(internalResponse
 			sm.removeSubscription(k)
 		}
 		// Clear out the buffer
+		close(sm.updateBuffer)
 		sm.updateBuffer = make(chan bufferInternalResponse, sm.updateBufferWindowSize)
 	}
 	sm.EmitMetrics()