from async generator

[ronag]

Would be nice if we somehow had a helper for the following pattern:

makeObservable(gen, options) {
  return  new rxjs.Observable((o) => {
    const ac = new AbortController()
    const subscription = rxjs
      .from(gen(options, { signal: ac.signal }))
      .subscribe(o)
    return () => {
      ac.abort()
      subscription.unsubscribe()
    }
  })
}

Or something like that... the point here is that the creating the generator should take a signal which should be connected to the lifetime of the observable.

[ronag]

@benjamingr wdyt?

[kwonoj]

Moved to discussion as gh issue is for the bug tracking purpose.

[kwonoj]

also fyi, I feel there's some overlap for official abortsignal support discussion (#5863) though it is not exactly same thing.

[benjamingr]

I would be cool (in Node.js) with making our streams Symbol.observable so that doing from works for them

[josepot]

I'm a bit confused now. Do you mean that it would be cool to provide an API that allows to consume Node.js streams as Asycn Iterables, through the [Symbol.asyncIterator] property on their instance? 🤔

I mean, don't get me wrong, I really wish that Symbol.observable was thing! and I would love to have Node.js streams work as InteropObservables... But I'm afraid that's not what you meant, right?

[benjamingr]

@josepot I mean it would be cool if Node.js streams directly interop'd (from Node's side) with RxJS observables. It's more for ronag. So basically:

I mean, don't get me wrong, I really wish that Symbol.observable was thing! and I would love to have Node.js streams work as InteropObservables... But I'm afraid that's not what you meant, right?

It is :) Node streams already have Symbol.asyncIterator but Node streams have two modes (push like RxJS and pull like async iterators) and it should be fine to support a "community agreed" well known symbol tell RxJS how to work with them automatically.

[josepot]

@ronag I've actually been thinking quite a lot about this lately b/c I'm currently working on a library that exposes InteropObservables. However, since InteropObservables are tricky to work with, I'm actually thinking about the possibility of exposing "aboratable" Async Iterables, instead.

But that's trickier than it looks, because RxJS from wouldn't be able to "abort" them when the consumer unsubscribes from the Observable that from returns... So I've spent a some time thinking what could be the best kind of convention (and operator) for "abortable" Async Iterables.

These are some ideas that I got:

• First and foremost, I think that there must be a convention for how the function that returns the cancelable AsyncIterator is going to receive the signal. For instance, a convention could be that it must be the first argument, lets see an example, so that it's easier to understand:

const delayOneSec = (signal: AbortSignal) => new Promise<void>((res, rej) => {
  function onAbort() {
    clearTimeout(token);
    rej(new AbortError());
  }

  signal.addEventListener('abort', onAbort);

  const token = setTimeout(() => {
    signal.removeEventListener('abort', onAbort);
    res();
  }, 1_000);
});

async function* asyncCounter(signal: AbortSignal, init = 1) {
  let count = init;
  try {
    while (true) {
      await delayOneSec(signal);
      yield count++;
    }
  } catch (e) {
    if (e instanceof Error && e.name === 'AbortError') return;
    throw e;
  } finally {
    // Maybe some extra cleanup goes here...
  }
}

• And then it would be possible to have an operator like this:

const fromCancellableAsyncIterable = <T, A extends Array<unknown>>(
  cancelableAsyncGenerator: (
    signal: AbortSignal,
    ...args: A
  ) => AsyncIterableIterator<T>,
  ...args: A
) => new Observable<T>((observer) => {
  const ac = new AbortController();
  const asyncIterator = cancelableAsyncGenerator(ac.signal, ...args);

  (async () => {
    for await (const value of asyncIterator) observer.next(value);
  })().catch((e) => {
    observer.error(e);
  });

  return () => {
    ac.abort();
  };
});

• Which could be used like this:

fromCancellableAsyncIterable(asyncCounter, 10)
  .pipe(take(3))
  .subscribe((x) => {
    console.log('count', x);
  });

You can find the full example here, in case that you want to fiddle with it.

That way the asyncCounter could be used from RxJS or from anywhere else that supports this convention for functions that return "cancellable" Async Iterables.

There are other possible conventions, of course, but IMO this one (passing AbortSignal as the first argument) is the one that makes more sense.

The real problem is that it's difficult to enforce a convention like this, when there is no way to identify Abort errors originating from controllers... So yeah, that's a bit of a bummer 😞. I mean, I guess that as part of the convention the .abort method should be called with either no arguments or with an AbortError or something like that...

Anyways, just some food for thought, I guess...

[voliva]

I'm not fully convinced rxjs should include any operator that accepts functions that have some specific shape, when there's no standard convention for that.

I guess that's focusing on async generators, but as far as I know, there's no standard way of passing an AbortSignal to them. It's completely up to the developer.

If that's the case, then you'll have projects that will put the AbortSignal on the first parameter. Others will put it on the last parameter. Others will put it in a config object (similar to DOM's fetch api). Then why should rxjs enforce one of them specifically?

Wouldn't it be better to leave this up to the consumer? If we didn't use rxjs, and want to expose a "push" version of one async generator, we would do something like:

const ac = new AbortController();

const iterable = myAsyncGenerator(myParams, { signal: ac.signal, otherOption: 'blah' })

const result = function (callback: (value: any) => void) {
  async function run() {
    let next;

    while(!(next = await iterable.next()).done) {
        callback(next.value);
    }
  }
  run();

  return () => ac.abort();
}

If we want to do this in rxjs, it would be greatly simplified (assuming from accepts async iterables):

const ac = new AbortController();

const result$ = from(
  myAsyncGenerator(myParams, { signal: ac.signal, otherOption: 'blah' })
).pipe(
  finalize(() => ac.abort())
)

I haven't tried, I'm probably missing something, but it should be something really close to this.

But if you look closely, rxjs is not dealing directly with the generator function. It's dealing with the async iterable that it returned, which will be aborted when the signal (that rxjs doesn't have access) goes off.

So as developers we need to abort that signal when our consumer unsubscribes from it. In my implementation it will also abort the signal once the asyncGenerator ends, but I think that should be alright?

If a project uses a specific convention of how to use AbortSignals, then maybe it's up to a developer of that project of writing a helper function that will suit their own use case. But I don't think the AbortController should be of any concern to the internals of rxjs for this particular case.

In a way it would be like adding a function in rxjs for:

const ac = new AbortController();

const myPromise = doSomethingAsync(ac.signal);

from(myPromise).pipe(
  finalize(() => ac.abort())
)

Like.... yes, it could. But how exactly? What specific convention should we use? It's completely up to the developer that wants to call the function to pass in the parameters it wants in the order and shape they want.

Edit - To make examples simpler I eagerly called the generator/promises. You'd probably want to wrap these into a defer(() => on rxjs, and in a closure for vanilla JS