async await in observer function

Question

I want to create an observable from which to push values to my observers.

The problem is, that those values are obtained via calling a promise-returning method.

Is there a way to call await on the promise-returning function within the observable, like the following pseudo-code (which does not work, because i can't mark the function passed to the observable async)...

return new Observable((observer) => {

  let counter = 0;

  for(...)
  {
    try
    {
      const value = await promisefunction(counter++);
                    ^^^^^
      observer.next(value);
    }
    catch(error)
    {
      observer.complete();
    }
  }

});

Whats the error? await can be used only within async function — Julius

Brandon Brandon · Accepted Answer · 2019-11-21T19:19:38

Most observable operators that consume an observable can also consume a promise. So you can get what you want via concatMap and defer:

const observable = of(arrayOfObjects).pipe(
    concatMap((object, i) => defer(() => promiseFunction(object, i))
)

We use defer to defer actually executing promisefunction (and "starting" the promise) until the previous promise completes. So the above construct will execute the function sequentially one at a time, just like you were trying to do with your async for loop.

Edit: You can also do it your way like so:

const observable = new Observable(observer => {
  let canceled = false;
  let count = 0;

  async function engine() {
    try {
      while (!canceled) {
        const value = await promiseFunction(++count);
        observer.next(value);
      }
    }
    catch (e) {
      observer.error(e);
    }
  }

  engine();
  return () => canceled = true;
});

async await in observer function

1 Answers