node.js - ネストされたストリームの処理

Question

2 つの csv 入力ストリームを結合して出力ファイルを生成しようとしています。csv 1 の各レコードについて、csv 2 の各レコードの出力を生成したいと考えています。

同様のソリューションのスタックオーバーフローを閲覧しているときにハイランドに出くわし、出くわしました：

Highland.js でのネストされたストリーム操作

これを自分の問題に合わせて調整しようとしましたが、これまでのところ次のとおりです。

    const debug = require('debug')('csvparse');
    const csv = require('fast-csv');
    const fs = require('fs');
    const args = process.argv;
    const h = require('highland');

    const typestream = h(fs.createReadStream(args[2]).pipe(csv({ headers: true, ignoreEmpty: true })));
    const postcodestream = h(fs.createReadStream(args[3]).pipe(csv({ headers: true, ignoreEmpty: true })));

    const pipeline = typestream.flatMap((type) => {
        debug(type);

        return postcodestream.flatMap((postcode) => {
            debug(postcode);

            return h([`${type.type}-${postcode.postcode}\n`]);
        });
    });

    pipeline.pipe(process.stdout);

次の例では、csv1 を入力します。

type,
STREET,
ROAD,

csv2:

postcode,
3456
3446
1234

Idは出力を期待しています

STREET-3456
STREET-3446
STREET-1234
ROAD-3456
ROAD-3446
ROAD-1234

しかし、私はちょうど得ています:

STREET-3456
STREET-3446
STREET-1234

デバッグ ステートメントから、ROAD を一度終了すると停止することがわかります。