IterableEmitter for ES6/TypeScript - async iterator for an event stream

iterable-emitter.ts

import { EventEmitter } from "events"

/**
 * Transforms a stream of events into an array-like iterable which produces promises.
 * Empty event value means the end of the stream.
 * 
 * Poor-man's async iterator.
 * 
 * Usage:
 * ```
 * async function() {
 *   let events = new IterableEmitter(emitter, 'event_name')
 *   for (let item of events) {
 *     item = await item
 *     console.log(item)
 *   }
 *   console.log('done')
 * }
 */
class IterableEmitter<T> implements Iterable<Promise<T>> {
    queue: T[] = [];
    done = false;
    debug = false
    protected resolve: ((T) => void) | null = null;
    constructor(
            public emitter: EventEmitter,
            public event_name: string,
            public unpack: (any) => T = val=>val) {
        // empty
    }
    /**
     * Current size of the buffer queue.
     */
    get length() {
        return this.queue.length
    }
    log(msg: string) {
        if (this.debug)
            console.log(msg)
    }
    listener: Function | null;
    start() {
        const context = this
        this.listener = function(item: T) {
            item = context.unpack(item)
            context.log(`emitter.on('${context.event_name}')`)
            if (!item) {
                context.done = true
                context.log('emitter done')
            } else {
                context.queue.push(item)
            }
            if (context.resolve) {
                context.resolve(context.queue.shift())
                context.resolve = null
            }
        }
        this.emitter.on(this.event_name, this.listener)
    }
    dispose() {
        this.emitter.removeListener(this.event_name, this.listener)
    }
    *[Symbol.iterator]() {
        this.log('iterator start')
        while (!this.done || this.queue.length) {
            yield new Promise<T>(resolve=>{
                // resolve from the buffer
                if (this.queue.length)
                    resolve(this.queue.shift())
                // wait for data
                else
                    this.resolve = resolve
            })
        }
        this.log('iterator end')
    }
}

export default IterableEmitter