Реализовать оператор ожидания в RxJS - PullRequest
Новая
       44

Реализовать оператор ожидания в RxJS

1 голос
/ 22 февраля 2020

У меня есть следующий оператор: 

 for await (const blob of client.list()) {
    console.log('\t', blob.name);
}

client.list () возвращает асинхронный итератор c и ожидает использования для ожидания ... * для решить обещания. Я хотел бы включить код в существующий канал rx js, который создает экземпляр клиента.

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

Любая помощь будет признательна!

1 Ответ

1 голос
/ 23 февраля 2020

Я не могу найти существующего оператора rx js, но это не так уж сложно сделать самостоятельно. При интеграции других API с наблюдаемыми объектами вы можете взаимодействовать с API внутри функции, передаваемой наблюдаемому конструктору. Это обеспечивает большую гибкость при запуске next / error / complete.

Редактировать - я добавил второй вариант для этого, используя операторы rx js и избегая явного вызова next / error / complete. 

const {
  Observable,
  operators,
  from
} = rxjs;
const {take, expand, map, filter} = operators;

const asyncGen = async function*(x = -1) {
  while(x++ < 5) {
    yield x;
  }
};

const fromAsyncIter = iterable => new Observable(subscriber => {
  let unsubscribed = false;
  const iterate = async () => {
    try {
      for await (let n of iterable) {
        console.log('await', n);
        subscriber.next(n);
        if (unsubscribed) return;
      }
      subscriber.complete();
    } catch (e) {
      subscriber.error(e);
    }
  }
  iterate();
  return () => unsubscribed = true;
});

const fromAsyncIter2 = iterable =>
  from(iterable.next()).pipe(
    expand(() => iterable.next()),
    filter(x => !x.done),
    map(x => x.value)
  );

// const source = fromAsyncIter(asyncGen()).pipe(take(2));
const source = fromAsyncIter2(asyncGen()).pipe(take(2));

source.subscribe({
  next: x => console.log('next', x),
  error: e => console.error(e),
  complete: () => console.log('complete')
});
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.4/rxjs.umd.js"></script>
...