Convert an Observable to an async generator

user_4499525 0
Viewed 1520

I'm trying to use rxjs in conjunction with babeljs to create an async generator function that yields when next is called, throws when error is called, and finishes when complete is called. The problem I have with this is that I can't yield from a callback.

I can await a Promise to handle the return/throw requirement.

async function *getData( observable ) {
    await new Promise( ( resolve, reject ) => {
        observable.subscribe( {
            next( data ) {
                yield data; // can't yield here
            },
            error( err ) {
                reject( err );
            },
            complete() {
                resolve();
            }
        } );
    } );
}

( async function example() {
    for await( const data of getData( foo ) ) {
        console.log( 'data received' );
    }
    console.log( 'done' );
}() );

Is this possible?

ecmascript-next reactive babeljs rxjs
2 Answers

I think a problem with this solution is that the observable could generate several values in one batch (without deferring). This is my proposal:

const defer = () => new Promise (resolve =>
    setTimeout (resolve, 0));

async function* getData (observable)
{
    let values = [];
    let error = null;
    let done = false;
    observable.subscribe (
        data => values.push (data),
        err => error = err,
        () => done = true);
    for (;;)
    {
        if (values.length)
        {
            for (const value of values)
                yield value;
            values = [];
        }
        if (error)
            throw error;
        if (done)
            return;
        await defer ();
    }
}
user_56908090
Related