Как реализовать асинхронный генератор для потоковой передачи с узлом Readable Stream? - PullRequest
Новая
       7

Как реализовать асинхронный генератор для потоковой передачи с узлом Readable Stream?

0 голосов
/ 25 февраля 2019

Я хочу сделать что-то подобное 

const { Readable } = require("stream");

function generatorToStream(generator) {
  return new Readable({
    read() {
      (async () => {
        for await (const result of generator()) {
          if (result.done) {
            this.push(null);
          } else {
            this.push(result.value);
          }
        }
      })();
    }
  });
}

generatorToStream(async function*() {
  const msg1 = await new Promise(resolve =>
    setTimeout(() => resolve("ola amigao"), 2000)
  );
  yield msg1;
  const msg2 = await new Promise(resolve =>
    setTimeout(() => resolve("ola amigao"), 2000)
  );
  yield msg2;

  const msg3 = await new Promise(resolve =>
    setTimeout(() => resolve("ola amigao"), 2000)
  );
  yield msg3;
}).pipe(process.stdout);

, но оно не работает, событие завершения никогда не вызывалось, и я не получил никаких данных на своем терминале.

Любойрешение или советы по его реализации?

1 Ответ

0 голосов
/ 27 февраля 2019

Я являюсь автором Scramjet , функциональной среды обработки потоков, которая может быть простым решением для вас.

Если вы согласны с добавлением всего 3 зависимостейВ этом случае это не может быть проще: 

const {StringStream} = require("scramjet");

StringStream.from(async function* () {
    yield await something();
    ...
});

Если вы хотите реализовать это самостоятельно, взгляните на исходный код в строке DataStream 112 - это должно произойтибыть довольно простым в реализации.В общем, вам нужно реализовать что-то вроде этого: 

function generatorToStream(gen) {
    // obtain an iterator
    const iter = await gen();
    // create your output
    const out = new Passthrough();

    // this IIFE will do all the work
    (async () => {
        let done = false;
        for await (let chunk of iter) {
            // if write returns true, continue, otherwise wait until out is drained.
            if (!out.write(chunk)) await new Promise((res, rej) => this.once("drain", res);
        }
    })()
        // handle errors by pushing them to the stream for example
        .catch(e => out.emit('error', e));

    // return the output stream
    return out;
}

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

...