Потоки NodeJS не ожидают асинхронности

Question

Я столкнулся с проблемой при тестировании потоков NodeJS.Кажется, я не могу заставить свой проект ждать вывода из потоков Duplex и Transform после запуска stream.pipeline, даже если он возвращает обещание.Возможно, я что-то упустил, но я считаю, что сценарий должен дождаться возврата функции, прежде чем продолжить.Самая важная часть проекта, которую я пытаюсь заставить работать:

// Message system is a duplex (read/write) stream
export class MessageSystem extends Duplex {
    constructor() {
        super({highWaterMark: 100, readableObjectMode: true, writableObjectMode: true});
    }
    public _read(size: number): void {
        var chunk = this.read();
        console.log(`Recieved ${chunk}`);
        this.push(chunk);
    }
    public _write(chunk: Message, encoding: string, 
        callback: (error?: Error | null | undefined, chunk?: Message) => any): void {
        if (chunk.data === null) {
            callback(new Error("Message.Data is null"));
        } else {
            callback();
        }
    }
}

export class SystemStream extends Transform {
    public type: MessageType = MessageType.Global;
    public data: Array<Message> = new Array<Message>();
    constructor() {
        super({highWaterMark: 100, readableObjectMode: true, writableObjectMode: true});
    }
    public _transform(chunk: Message, encoding: string, 
        callback: TransformCallback): void {
        if (chunk.single && (chunk.type === this.type || chunk.type === MessageType.Global)) {
            console.log(`Adding ${chunk}`);
            this.data.push(chunk);
            chunk = new Message(chunk.data, MessageType.Removed, true);
            callback(undefined, chunk); // TODO: Is this correct?
        } else if (chunk.type === this.type || chunk.type === MessageType.Global) { // Ours and global
            this.data.push(chunk);
            callback(undefined, chunk);
        } else { // Not ours
            callback(undefined, chunk);
        }
    }
}

export class EngineStream extends SystemStream {
    public type: MessageType = MessageType.Engine;
}

export class IOStream extends SystemStream {
    public type: MessageType = MessageType.IO;
}

let ms = new MessageSystem();
let es = new EngineStream();
let io = new IOStream();

let pipeline = promisify(Stream.pipeline);

async function start() {
    console.log("Running Message System");
    console.log("Writing new messages");
    ms.write(new Message("Hello"));
    ms.write(new Message("world!"));
    ms.write(new Message("Engine data", MessageType.Engine));
    ms.write(new Message("IO data", MessageType.IO));
    ms.write(new Message("Order matters in the pipe, even if Global", MessageType.Global, true));
    ms.end(new Message("Final message in the stream"));
    console.log("Piping data");
    await pipeline(
        ms,
        es,
        io
    );
}

Promise.all([start()]).then(() => {
    console.log(`Engine Messages to parse: ${es.data.toString()}`);
    console.log(`IO Messages to parse: ${io.data.toString()}`);
});

Вывод должен выглядеть примерно так:

Running message system
Writing new messages
Hello
world!
Engine Data
IO Data
Order Matters in the pipe, even if Global
Engine messages to parse: Engine Data
IO messages to parse: IO Data

Любая помощь будет принята с благодарностью.Спасибо!

Примечание. Я опубликовал это с другим аккаунтом, а не с моим реальным аккаунтом.Извиняюсь за дубликат.

Редактировать: у меня изначально была закрытая репозитория, но я сделал ее общедоступной, чтобы помочь уточнить ответ.Больше возможностей можно найти в ветке feature / inital_system .Он может быть запущен с npm start при извлечении.

Редактировать: я разместил здесь свои пользовательские потоки для многословия.Я думаю, что я нахожусь на лучшем треке, чем раньше, но теперь получаю «нулевой» объект, полученный по конвейеру.

Answer 1

После некоторой работы за последние пару дней я нашел свой ответ.Проблема заключалась в моей реализации дуплексного потока.С тех пор я изменил MessageSystem на поток Transform, чтобы им было легче управлять и работать с ним.

Вот продукт:

export class MessageSystem extends Transform {
    constructor() {
        super({highWaterMark: 100, readableObjectMode: true, writableObjectMode: true});
    }
    public _transform(chunk: Message, encoding: string,
        callback: TransformCallback): void {
            try {
                let output: string = chunk.toString();
                callback(undefined, output);
            } catch (err) {
                callback(err);
            }
        }
}

Спасибо @estus за быстроеответь и проверь.Опять же, я все время нахожу свой ответ в API!

Архивное хранилище моих находок можно найти в этом хранилище .

Answer 2

Поскольку документация гласит, stream.pipeline на основе обратного вызова не возвращает обещание.

У него есть собственная обещанная версия, к которой можно получить доступ с помощью util.promisify:

const pipeline = util.promisify(stream.pipeline);

...

await pipeline(...);