fix: guarantee single write of length prefixed data

ChainSafe · Jan 30, 2024 · e5fdd3e · e5fdd3e
1 parent 57d12f2
commit e5fdd3e
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 17 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -81,6 +81,7 @@
     "abortable-iterator": "^5.0.1",
     "denque": "^2.1.0",
     "it-length-prefixed": "^9.0.4",
+    "it-map": "^3.0.5",
     "it-merge": "^3.0.3",
     "it-pipe": "^3.0.1",
     "it-pushable": "^3.2.3",

diff --git a/src/stream.ts b/src/stream.ts
@@ -1,5 +1,6 @@
 import { abortableSource } from 'abortable-iterator'
 import { encode, decode } from 'it-length-prefixed'
+import map from 'it-map'
 import merge from 'it-merge'
 import { pipe } from 'it-pipe'
 import { pushable, type Pushable } from 'it-pushable'
@@ -30,12 +31,9 @@ export class OutboundStream {
 
     pipe(
       abortableSource(
-        merge<Uint8Array | Uint8ArrayList>(
+        merge(
           this.lpPushable,
-          pipe(
-            this.pushable,
-            (source) => encode(source)
-          )
+          map(this.pushable, buf => encode.single(buf))
         ), this.closeController.signal, { returnOnAbort: true }
       ),
       this.rawStream