01_no-global-slow-queue.md

Non global slowing message queue

As in the title

Example

async function fakeProvider(onData) {
    onData('line 1');
    onData('line 2');
    onData('line 3');
    await sleep(5000);
    onData('line 4');
    onData('line 5');
    onData('line 6');
}

slowify(fakeProvider, data => {
    console.log(`Log: "${data}"`);
});

02_queue.js

const sleep = t => new Promise(r => setTimeout(r, t))

export default function slowify() {
    if (arguments.length === 2) {
        const [is, onData] = arguments;
        return _slowify(is, {}, onData);
    }
    if (arguments.length === 3) {
        const [is, options, onData] = arguments;
        return _slowify(is, options, onData);
    }

    throw 'Illegal number of arguments';
}

function _slowify(inputStream, options, onData) {
    const {
        shortWait = 1000,
        longWait = 5000,
        stopImmediatly = false
    } = options;

    const queue = [];

    let done = false;
    (async () => {
        await inputStream(data => {
            console.log(`Got "${data}" from inputStream`);
            queue.push(data);
        });
        done = true;
    })();

    (async () => {
        while (!done || (!stopImmediatly && queue.length > 0)) {
            if (queue.length > 0) {
                const data = queue.shift();
                console.log(`Sending "${data}" to outputStream`);
                onData(data);
                await sleep(shortWait);
            } else {
                console.log(`Longwait...`);
                await sleep(longWait);
            }
        }
    })();
}

03_queue-async-generator.js

const sleep = t => new Promise(r => setTimeout(r, t))

function slowify(generator, options = {}) {
    const {
        shortWait = 1000,
        longWait = 5000,
        stopImmediatly = false
    } = options;

    const queue = [];

    let done = false;
    (async () => {
        for await (const data of generator) {
            console.log(`Got "${data}" from generator`);
            queue.push(data);
        }
        done = true;
    })();

    return (async function* () {
        await sleep(0);
        while (!done || (!stopImmediatly && queue.length > 0)) {
            if (queue.length > 0) {
                const data = queue.shift();
                console.log(`Sending "${data}" to outputStream`);
                yield data;
                await sleep(shortWait);
                console.log('stopped waiting');
            } else {
                console.log(`Longwait...`);
                await sleep(longWait);
            }
        }
    })();
}

// Example

async function* fakeProvider() {
    yield 'line 1';
    yield 'line 2';
    yield 'line 3';
    await sleep(5000);
    yield 'line 4';
    yield 'line 5';
    yield 'line 6';
}

const slower = slowify(fakeProvider());

for await (const data of slower) {
    console.log(`Log: "${data}"`);
}