chore: remove double-pipe

Author: achingbrain

package-lock.json

@@ -81,8 +81,6 @@
     "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",
     "multiformats": "^13.0.1",

package.json

@@ -1,7 +1,5 @@
 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'
 import type { Stream } from '@libp2p/interface'
@@ -18,24 +16,17 @@ interface InboundStreamOpts {
 }
 
 export class OutboundStream {
-  private readonly pushable: Pushable<Uint8Array>
-  private readonly lpPushable: Pushable<Uint8ArrayList>
+  private readonly pushable: Pushable<Uint8Array | Uint8ArrayList>
   private readonly closeController: AbortController
   private readonly maxBufferSize: number
 
   constructor (private readonly rawStream: Stream, errCallback: (e: Error) => void, opts: OutboundStreamOpts) {
-    this.pushable = pushable({ objectMode: false })
-    this.lpPushable = pushable({ objectMode: false })
+    this.pushable = pushable()
     this.closeController = new AbortController()
     this.maxBufferSize = opts.maxBufferSize ?? Infinity
 
     pipe(
-      abortableSource(
-        merge(
-          this.lpPushable,
-          map(this.pushable, buf => encode.single(buf))
-        ), this.closeController.signal, { returnOnAbort: true }
-      ),
+      abortableSource(this.pushable, this.closeController.signal, { returnOnAbort: true }),
       this.rawStream
     ).catch(errCallback)
   }
@@ -51,24 +42,23 @@ export class OutboundStream {
       throw Error(`OutboundStream buffer full, size > ${this.maxBufferSize}`)
     }
 
-    this.pushable.push(data)
+    this.pushable.push(encode.single(data))
   }
 
   /**
    * Same to push() but this is prefixed data so no need to encode length prefixed again
    */
   pushPrefixed (data: Uint8ArrayList): void {
-    if (this.lpPushable.readableLength > this.maxBufferSize) {
+    if (this.pushable.readableLength > this.maxBufferSize) {
       throw Error(`OutboundStream buffer full, size > ${this.maxBufferSize}`)
     }
-    this.lpPushable.push(data)
+    this.pushable.push(data)
   }
 
   async close (): Promise<void> {
     this.closeController.abort()
     // similar to pushable.end() but clear the internal buffer
     await this.pushable.return()
-    await this.lpPushable.return()
     await this.rawStream.close()
   }
 }