Fix gRPC stream pausing

Try to pause as soon as possible to prevent buffer overflow
proxima-one · May 1, 2023 · b1ecef4 · b1ecef4
1 parent 96f4b90
commit b1ecef4
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/src/streamdb/consumerClient.ts b/src/streamdb/consumerClient.ts
@@ -72,6 +72,9 @@ export class StreamDBConsumerClient {
 
     return PausableStream.create<StreamEvent>((observer, pauseState) => {
       grpcStreamResponse.on("data", d => {
+        if (pauseState.isPaused && !grpcStreamResponse.isPaused()) {
+          grpcStreamResponse.pause();
+        }
         for (const obj of d.stateTransition) {
           observer.next(stateTransitionProtoToStreamEvent(obj));
         }

diff --git a/src/streamdb/converters.ts b/src/streamdb/converters.ts
@@ -34,7 +34,9 @@ export function protoToOffset(offsetProto: proto_model.Offset): Offset {
   return new Offset(
     offsetProto.id,
     BigInt(offsetProto.height),
-    offsetProto.timestamp ? protoToTimestamp(offsetProto.timestamp) : Timestamp.zero
+    offsetProto.timestamp
+      ? protoToTimestamp(offsetProto.timestamp)
+      : Timestamp.zero
   );
 }
 export function offsetToProto(offset: Offset): proto_model.Offset {