Merge pull request #410 from autonomys/duplicate-streams

feat: safe fork stream method for asynchronous pkg

Author: clostao

packages/utility/asynchronous/package.json

@@ -22,12 +22,16 @@
   "types": "./dist/index.d.ts",
   "devDependencies": {
     "@types/jest": "^29.5.14",
+    "@types/stream-fork": "^1",
     "eslint": "^9.25.1",
     "interface-store": "^6.0.2",
     "jest": "^29.7.0",
     "prettier": "^3.5.3",
     "ts-jest": "^29.3.1",
     "typescript": "^5.8.3"
   },
-  "gitHead": "ef4c21d683cad697f7015e52becd399a8ca2ed84"
+  "gitHead": "ef4c21d683cad697f7015e52becd399a8ca2ed84",
+  "dependencies": {
+    "stream-fork": "^1.0.5"
+  }
 }

packages/utility/asynchronous/src/index.ts

@@ -1,2 +1,3 @@
 export * from './asyncGenerators'
 export * from './concurrency'
+export * from './streams'

packages/utility/asynchronous/src/streams/index.ts

@@ -0,0 +1,12 @@
+import { PassThrough, Readable } from 'stream'
+import streamFork from 'stream-fork'
+
+export async function forkStream(stream: Readable): Promise<[Readable, Readable]> {
+  const passThrough1 = new PassThrough()
+  const passThrough2 = new PassThrough()
+  const writable = streamFork.fork([passThrough1, passThrough2])
+
+  stream.pipe(writable)
+
+  return [passThrough1, passThrough2]
+}

yarn.lock

@@ -34,10 +34,12 @@ __metadata:
   resolution: "@autonomys/asynchronous@workspace:packages/utility/asynchronous"
   dependencies:
     "@types/jest": "npm:^29.5.14"
+    "@types/stream-fork": "npm:^1"
     eslint: "npm:^9.25.1"
     interface-store: "npm:^6.0.2"
     jest: "npm:^29.7.0"
     prettier: "npm:^3.5.3"
+    stream-fork: "npm:^1.0.5"
     ts-jest: "npm:^29.3.1"
     typescript: "npm:^5.8.3"
   languageName: unknown
@@ -7768,6 +7770,15 @@ __metadata:
   languageName: node
   linkType: hard
 
+"@types/stream-fork@npm:^1":
+  version: 1.0.5
+  resolution: "@types/stream-fork@npm:1.0.5"
+  dependencies:
+    "@types/node": "npm:*"
+  checksum: 10c0/139d047cccfcbe6ac23be7a7d025b0ffbb885a0075b19248f2b7bed1becc6647180196c50152cafecb6cba52d7d1e4507d2beb2f632500bc426c06d86797fce1
+  languageName: node
+  linkType: hard
+
 "@types/unist@npm:^2.0.0":
   version: 2.0.11
   resolution: "@types/unist@npm:2.0.11"
@@ -20935,6 +20946,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"stream-fork@npm:^1.0.5":
+  version: 1.0.5
+  resolution: "stream-fork@npm:1.0.5"
+  checksum: 10c0/b13a88550276d80ddaccb17be3e4e47e69f275a86b8f74f763125b02d699293ef41b52b456ed0897c21b38f4e895b7d4606495dc9411ab006050047476933bdf
+  languageName: node
+  linkType: hard
+
 "stream-shift@npm:^1.0.0":
   version: 1.0.3
   resolution: "stream-shift@npm:1.0.3"