Processing asynchronously pushed data

async-data.js

// Originally from: http://exploringjs.com/es6/ch_generators.html#_example-processing-asynchronously-pushed-data

'use strict'
const createReadStream = require('fs').createReadStream

function coroutine(generatorFunction) {
    return function (...args) {
        const generatorObject = generatorFunction(...args);
        generatorObject.next()
        return generatorObject
    };
}


function readFile(fileName, target) {
    const readStream = createReadStream(fileName, { 
        encoding: 'utf-8', 
        bufferSize: 1024 
    })

    readStream.on('data', buffer => {
        const str = buffer.toString('utf8')
        target.next(str)
    })
    readStream.on('end', () => target.return())
}

const splitLines = coroutine(function* (target) {
    let previous = ''
    try {
        while (true) {
            previous += yield
            let eolIndex
            while ((eolIndex = previous.indexOf('\n')) >= 0) {
                const line = previous.slice(0, eolIndex)
                target.next(line)
                previous = previous.slice(eolIndex + 1)
            }
        }
    } finally {
        if (previous.length > 0) {
            target.next(previous)
        }
        target.return()
    }
})

const numberLines = coroutine(function* (target) {
    try {
        for (let lineNo = 0; /* noop */; lineNo++) {
            const line = yield
            target.next(`${lineNo}: ${line}`)
        }
    } finally {
        target.return()
    }
})

const printLines = coroutine(function* () {
    while (true) {
        const line = yield
        console.log(line)
    }
})

// readFile(<file>, splitLines(numberLines(printLines())))