feat: fromCallback

Author: reconbot

lib/from-callback-test.ts

@@ -0,0 +1,38 @@
+import { assert } from 'chai'
+import { fromCallback } from './from-callback'
+import { collect } from './collect'
+import EventEmitter from 'events'
+
+describe('fromCallback', () => {
+  it('buffers values', async () => {
+    const itr = fromCallback()
+    itr.yield(1)
+    itr.yield(2)
+    itr.yield(3)
+    itr.end()
+    const values = await collect(itr)
+    assert.deepEqual(values, [1,2,3])
+  })
+  it('works with event emitters', async () => {
+    const emitter = new EventEmitter()
+    const itr = fromCallback()
+    emitter.on('data', itr.yield)
+    emitter.on('close', itr.end)
+    emitter.emit('data', 1)
+    emitter.emit('data', 2)
+    emitter.emit('data', 3)
+    emitter.emit('close')
+    const values = await collect(itr)
+    assert.deepEqual(values, [1,2,3])
+  })
+  it('ignores values after end', async () => {
+    const itr = fromCallback()
+    itr.yield(1)
+    itr.yield(2)
+    itr.yield(3)
+    itr.end()
+    itr.yield(5)
+    const values = await collect(itr)
+    assert.deepEqual(values, [1,2,3])
+  })
+})

lib/from-callback.ts

@@ -0,0 +1,59 @@
+import { deferGenerator } from 'inside-out-async'
+
+export interface CallbackIterable<T> extends AsyncIterable<T> {
+  yield(data: T): void
+  end(): void
+}
+
+/**
+ * Returns an iterable with methods to help turn event emitters or callbacks into async iterables.
+
+This leverages the [`inside-out-async`](https://www.npmjs.com/package/inside-out-async#deferGenerator) package which can be used directly if you want something similar for generators. (It is bundled so it's not a dependency.)
+
+It adds two methods to the returned iterable.
+
+- `itr.yield(data: T): void` queues data to be read
+- `itr.end(): void` ends the iterable
+
+And will buffer *all* data given to `yield()` until it's read.
+
+```ts
+import { fromCallback } from 'streaming-iterables'
+
+const pokeLog = fromCallback()
+itr.yield('Charmander')
+itr.yield('Ash')
+itr.yield('Pokeball')
+itr.end()
+
+for await (const pokeData of pokeLog) {
+  console.log(pokeData) // Charmander, Ash, Pokeball
+}
+
+// To use it as a callback
+const emitter = new EventEmitter()
+const consoles = fromCallback()
+emitter.on('data', consoles.yield)
+emitter.on('close', consoles.end)
+
+emitter.emit('data', 'nintendo')
+emitter.emit('data', 'sony')
+emitter.emit('data', 'sega')
+emitter.emit('close')
+
+for await (const console of consoles) {
+  console.log(console) // 'nintendo', 'sony', 'sega'
+}
+
+```
+ */
+export function fromCallback<T>(): CallbackIterable<T> {
+  const { generator, queueValue, queueReturn } = deferGenerator<T, T, undefined>()
+
+  const cbIterable: CallbackIterable<T> = {
+    ...generator,
+    yield: queueValue,
+    end: () => queueReturn()
+  }
+  return cbIterable
+}

lib/index.ts

@@ -27,3 +27,4 @@ export { throttle } from './throttle'
 export { time, TimeConfig, CurriedTimeResult } from './time'
 export { transform } from './transform'
 export { writeToStream, WritableStreamish } from './write-to-stream'
+export { fromCallback } from './from-callback'