[pinpoint-apm#182] Support gRPC stream flowable

feelform · Mar 22, 2024 · 3ff5b78 · 3ff5b78
1 parent 03a919a
commit 3ff5b78
Showing 1 changed file with 15 additions and 10 deletions.
diff --git a/lib/client/bounded-buffer-readable-stream.js b/lib/client/bounded-buffer-readable-stream.js
@@ -11,12 +11,10 @@ const { Readable } = require('node:stream')
 class BoundedBufferReadableStream {
     constructor(constructorOptions) {
         this.buffer = []
-        this.stopped = false
 
         const options = constructorOptions || {}
         this.readable = new Readable(Object.assign({
             read() {
-                this.stopped = false
                 this.readStart()
             }
         }, options))
@@ -26,28 +24,35 @@ class BoundedBufferReadableStream {
         if (!data) {
             return
         }
-
         this.buffer.push(data)
-        this.readStart()
+
+        if (this.canStart()) {
+            this.readStart()
+        }
+    }
+
+    canStart() {
+        return !this.stopped
     }
 
     end() {
         this.readable.end()
     }
 
     readStart() {
-        if (this.stopped) {
-            return
-        }
+        this.stopped = false
 
         const length = this.buffer.length
         for (let index = 0; index < length; index++) {
-            if (this.readable.push(this.buffer.shift())) {
-                this.stopped = true
-                return
+            if (!this.readable.push(this.buffer.shift())) {
+                return this.readStop()
             }
         }
     }
+
+    readStop() {
+        this.stopped = true
+    }
 }
 
 module.exports = BoundedBufferReadableStream