feat(Observable): now implements Symbol.asyncIterator

[benlesh]

Adds async iterator support to Observable!

#6779
#6857

[demensky]

I suggest to play it safe and do it first as a regular method, and after a while do it with Symbol.asyncIterator. So there will be additional time to decide whether this was the optimal solution.

For example, the current implementation will trigger a microtask on each event, even if we already have events stored in the queue.

In streams, we usually work not with individual items, but with their chunks AsyncGenerator<T[], void, void>. Here, too, a similar approach suggests itself, returning an array of accumulated events.

for await (const events of source$) {
   console.log(events); // event accumulated between iterations
}

At the same time, nothing will interfere with us to work with each element separately. At the same time, it will not be unnecessary to provoke microtasks (and parasite Promise).

for await (const events of source$) {
  for (const event of events) {
     console.log(event);
  }
}

I hope that I was able to convey my thought correctly. Sorry if I wrote inaudibly somewhere. English is not my bearing language.

[benlesh]

@demensky The library https://github.com/benlesh/rxjs-for-await as existed for quite some time. Every single use of that library I've found uses eachValueFrom. There's probably an argument to be made for pulling the rest of that library in here... however, given that the behavior here is nearly identical to concatMap, and people don't often trip over that, the core team has decided that this is the proper route.

[ktrz]

Looks great! I just left one question regarding the tests

[ktrz]

Either a test name is misleading here or the unsubscription is not tested here. Are we expecting the subscription to be unsubscribed earlier if we break out of the loop?

Based on the test name I would imagine something like this:

    it('should unsubscribe if the for-await-of loop is broken', async () => {
      let state = 'idle';
      const source = new Observable<number>((subscriber) => {
        subscriber.next(1);
        setTimeout(() => subscriber.next(2), 100);
        setTimeout(() => subscriber.next(3), 200);
        setTimeout(() => subscriber.complete(), 500);

        return () => {
          state = 'unsubscribed'
        }
      });

      expect(state).to.equal('idle')

      const results: number[] = [];
      for await (const value of source) {
        results.push(value);
        break;
      }

      expect(state).to.equal('unsubscribed')
      expect(results).to.deep.equal([1]);
    });

[jayphelps]

Bump, as I see what they're saying. Same comment for the other unsubscribe test.

[benlesh]

I'll change this one around to test an active subscription count.

[ktrz]

TIL
I didn't know about this part of the iteration protocol. This is great!

[benlesh]

CORE TEAM: Approval.

[jayphelps]

Generally good, but some points that I at least wanted to run by you.

[jayphelps]

Bump, as I see what they're saying. Same comment for the other unsubscribe test.

[jayphelps]

Since chai doesn't have a way to assert the number of expected assertions (like jest does) I think it's generally a good idea to try to avoid assertions that rely on a condition that the test itself is trying to verify. e.g. if it stops throwing any error at all, this test would still pass. It would only fail if it does throw an error but that error doesn't have { message: 'wee' }.

Something like:

      let expectedError: any;
      try {
        for await (const value of source) {
          results.push(value);
        }
      } catch (err: any) {
        expectedError = err;
      }
      
      expect(expectedError.message).to.equal('wee');

[jayphelps]

IMO this code is "fine" but I think it would be much clearer for someone who isn't super-super familiar with how all this works to use an intermediate variable for the resolve, and maybe even an object with field names instead of an array tuple.

e.g.

const { resolve } = deferreds.shift()!;
resolve({ value, done: false });

[benlesh]

A solid suggestion. Implemented.

[jakovljevic-mladen]

LGTM, only minor types issues that could be considered fixing.