rx-from-async-iterator

A method to convert AsyncGeneratorObject to Rx.Observable

Install

npm i -S rx-from-async-iterator

Example

$3

``typescript
import { fromAsyncIterator } from 'rx-from-async-iterator';
import { fromEvent } from 'rxjs';
import { take } from 'rxjs/operators';

const click$ = fromEvent('click', document.body).pipe(take(2));

async function sleepOneSecond() {
return new Promise(r => setTimeout(r, 1000));
}

async function* subTask() {
await sleepOneSecond();
yield "d";
await sleepOneSecond();
yield "e";
await sleepOneSecond();
}

async function* taskAsync(error?: Error) {
yield "a";
await sleepOneSecond();
yield "b";
await sleepOneSecond();
yield "c";
yield* subTask();
yield "f";
yield click$;
}

fromAsyncIterator(taskAsync()).subscribe(console.log);
/// output as follows:
// -> 'a'
// -> 'b'
// -> 'c'
// -> 'd'
// -> 'e'
// -> 'f' /// and one second duration between each letter.
// -> MouseEvent
// -> MouseEvent
`

$3

Imagine we have an QRCode sign up feature to do, we need to poll an backend interface called /check/login

`typescript
async function* pollCheckLogin() {
let currentStatus = 'currentStatus';
while(currentStatus === 'pending') {
yield currentStatus = (await fetch('/check/login').then(resp => resp.json)).status;
}
if (currentStatus === 'timeout') {
yield 'timeout';
}
if (currentStatus === 'success') {
yield 'login success';
return;
}
throw new Error('unknown status');
}

fromAsyncIterator(pollCheckLogin()).subscribe((status) => {
switch(status) {
case 'timeout':
// handle timeout
break;
case 'login success':
// go to home page
break;
case 'pending':
// handle pending
break;
}
});
``

Try it