AsyncStreamIterator

Convert any readable stream to an async iterator and promise-like.

API

• AsyncStreamIterator(source: any, options: Options)
  • next(): Promise<IteratorResult<T>>
  • stop(): void Explicitly stops the iterator.
  • then(onfulfilled?: (data: any) => any, onrejected?: (err: any) => any): Promise<any>
• Options<T = Buffer>
  • events
    • data set customized data event name.
    • error set customized error event name.
    • end set customized end event name.
  • preprocessors
    • onData?: (msg: any) => T
    • onError?: (err: any) => Error
    • onEnd?: () => void

Examples

HTTP request

import * as http from "http";
import AsyncStreamIterator from "async-stream-iterator";

var server = http.createServer(async (req, res) => {
    let data = "";
    
    for await (let chunk of new AsyncStreamIterator(req)) {
        // Instead of converting the data here, you can pass the options
        // `preprocessors.onData` a function to do so.
        data += String(chunk);
    }

    res.end(data);
});

Socket message

import * as net from "net";
import AsyncStreamIterator from "async-stream-iterator";

var server = net.createServer(async (socket) => {
    for await (let chunk of new AsyncStreamIterator(socket)) {
        socket.write(chunk);
    }
});

Browser WebSocket

import AsyncStreamIterator from "async-stream-iterator";

var ws = new WebSocket("ws://localhost");
var iterator = new AsyncStreamIterator(ws, {
    events: {
        // When the event is bound to a property, it shuold be prefixed with '#'.
        data: "#onmessage",
        error: "#onerror",
        end: "#onclose"
    },
    preprocessors: {
        onData: (event) => event.data
    }
});

(async () => {
    for await (let data of iterator) {
        console.log(data);
    }
})();

Browser EventSource

import AsyncStreamIterator from "async-stream-iterator";

var es = new EventSource("ws://localhost");
var iterator = new AsyncStreamIterator(ws, {
    events: {
        data: "#onmessage",
        error: "#onerror"
    },
    preprocessors: {
        onData: (event) => event.data,
        onEnd: () => es.close()
    }
});

(async () => {
    for await (let data of iterator) {
        // TODO...

        if (/* goes well */) {
            // Explicitly close the iterator and `preprocessors.onEnd` is bound
            // to close the EventSource instance, the request will be closed as
            // well.
            iterator.close();
        }
    }
})();

Socket.io

import * as SocketIO from "socket.io";
import AsyncStreamIterator from "async-stream-iterator";

var ws = SocketIO(8000).on("connection", socket => {
    let task = (async () => {
        let iterator = new AsyncStreamIterator(socket, {
            events: {
                data: "some event"
            }
        });

        for await (let data of iterator) {
            // TODO...
        }

        // NOTE: AsyncStreamIterator only supports one argument passed to the
        // corresponding data event.
    })();
});