CODEWITHSUNDEEP
node.jsnode-streamshighland.js
fbiville
fbiville

Reputation: 8950

Zipping two readable streams does not seem to produce anything

I have the following function:

const _ = require('highland');

module.exports =
    (numbers /* Readable */, words /* Readable */, repeated_words /* Writable */) => {
        const numberStream = _(numbers);
        const wordStream = _(words);
        numberStream
            .zip(wordStream)
            .flatMap((numberWordPair) => {
                const result = [];
                for (let i = 0; i < numberWordPair[0]; i++) {
                    result.push(numberWordPair[1]);
                }
                return _(result);
            })
            .pipe(repeated_words);
    };

The stream arguments are automatically injected and I am 100% sure the stream injection works (other streaming functions work).

When I replace this somewhat complex transformation by something as straightforward as _(numbers).each(xs => {console.log(xs)}), I can see the data being logged.

However, here, I must be clearly missing something with Highland.js as there is nothing produced at all.

I'm using version 2.13.5 of Highland.js.

What am I missing?

Upvotes: 0

Views: 36

Answers (1)

fbiville
fbiville

Reputation: 8950

It appears it all works properly, therefore the mistake must lie somewhere else. As a proof, I was able to run this little program:

const {TextDecoder} = require('util');
const repeater = require('./lib/repeater');
const {PassThrough, Readable, Transform} = require('stream');

class ConstantSource extends Readable {

    constructor(options, constant) {
        super(options);
        this.constant = constant;
    }

    _read(size) {
        for (let i = 0; i < size; i++) {
            this.push(this.constant);
        }
    }
}

class StdinSource extends Readable {

    constructor(options) {
        super(options);
        this.decoder = new TextDecoder('utf8');
        process.stdin.on('data', (chunk) => {
            this.push(this.decoder.decode(chunk).trim());
        });
    }

    _read(size) {

    }
}

class StdoutSink extends Transform {

    constructor(options) {
        super(options);
        this.pipe(process.stdout);
    }

    _transform(chunk, _, callback) {
        callback(null, '\x1b[33m' + chunk + '\x1b[0m\n');
    }
}

const main = () => {
    if (process.argv.length < 3) {
        console.error('Please specify a number.');
        process.exit(1);
    }

    const options = {objectMode: true};
    const inputStream1 = new ConstantSource(options, process.argv[2]);
    const inputStream2 = new StdinSource(options);
    const outputStream = new StdoutSink(options);

    repeater(inputStream1, inputStream2, outputStream);
}

main();

Upvotes: 0

Related Questions