feat(stream-utils): splitStream method was added

Author: dangreen

packages/stream-utils/src/index.spec.ts

@@ -1,12 +1,14 @@
 import { spawn } from 'child_process'
+import { Readable } from 'stream'
 import {
   describe,
   it,
   expect
 } from 'vitest'
 import {
   toArray,
-  mergeReadables
+  mergeReadables,
+  splitStream
 } from './index.js'
 
 function stream(id: string, time: number) {
@@ -30,4 +32,24 @@ describe('stream-utils', () => {
       expect(result).toHaveLength(20)
     })
   })
+
+  describe('splitStream', () => {
+    it('should split strings stream by separator', async () => {
+      const stream = Readable.from([
+        '1 2',
+        ' 3',
+        ' 4 5 6'
+      ])
+      const result = await toArray(splitStream(stream, ' '))
+
+      expect(result).toEqual([
+        '1',
+        '2',
+        '3',
+        '4',
+        '5',
+        '6'
+      ])
+    })
+  })
 })

packages/stream-utils/src/index.ts

@@ -89,3 +89,30 @@ export function mergeReadables<
 
   return mergedStream
 }
+
+/**
+ * Split stream by separator.
+ * @param stream
+ * @param separator
+ * @yields String chunks.
+ */
+export async function* splitStream(stream: AsyncIterable<string | Buffer>, separator: string) {
+  let chunk: string | Buffer
+  let payload: string[]
+  let buffer = ''
+
+  for await (chunk of stream) {
+    buffer += chunk.toString()
+
+    if (buffer.includes(separator)) {
+      payload = buffer.split(separator)
+      buffer = payload.pop() || ''
+
+      yield* payload
+    }
+  }
+
+  if (buffer) {
+    yield buffer
+  }
+}