0

我有以下功能：

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);
    };

流参数是自动注入的，我 100% 确定流注入有效（其他流函数有效）。

当我用简单的东西替换这个有点复杂的转换时_(numbers).each(xs => {console.log(xs)})，我可以看到正在记录的数据。

但是，在这里，我必须清楚地错过Highland.js的某些内容，因为根本没有产生任何东西。

我正在使用 Highland.js 的 2.13.5 版本。

我错过了什么？

4

1 回答 1

0

看来一切正常，因此错误必须出在其他地方。作为证明，我能够运行这个小程序：

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();
于 2019-08-07T11:22:40.397 回答