How do I append to an observable inside the observable itself

user_14600972 0
Viewed 134

My situation is as follows: I am performing sequential HTTP requests, where one HTTP request depends on the response of the previous. I would like to combine the response data of all these HTTP requests into one observable. I have implemented this before using an async generator. The code for this was relatively simple:

async function* AsyncGeneratorVersion() {
  let moreItems = true; // whether there is a next page
  let lastAssetId: string | undefined = undefined; // used for pagination
  while (moreItems) {
    // fetch current batch (this performs the HTTP request)
    const batch = await this.getBatch(/* arguments */, lastAssetId);
    moreItems = batch.more_items;
    lastAssetId = batch.last_assetid;
    yield* batch.getSteamItemsWithDescription();
  }
}

I am trying to move away from async generators, and towards RxJs Observables. My best (and working) attempt is as follows:

const observerVersion = new Observable<SteamItem>((subscriber) => {
  (async () => {
    let moreItems = true;
    let lastAssetId: string | undefined = undefined;
    while (moreItems) {
      // fetch current batch (this performs the HTTP request)
      const batch = await this.getBatch(/* arguments */, lastAssetId);
      moreItems = batch.more_items;
      lastAssetId = batch.last_assetid;
      const items = batch.getSteamItemsWithDescription();
      for (const item of items) subscriber.next(item);
    }
    subscriber.complete();
  })();
});

Now, I believe that there must be some way of improving this Observer variant - this code does not seem very reactive to me. I have tried several things using pipe, however unfortunately these were all unsuccessful. I found concatMap to come close to a solution. This allowed me to concat the next HTTP request as an observable (done with the this.getBatch method), however I could not find a good way to also not abandon the response of the current HTTP request.

How can this be achieved? In short I believe this problem could be described as appending data to an observable inside the observable itself. (But perhaps this is not a good way of handling this situation)

reactivex rxjs
2 Answers

TLDR;

Here's a working StackBlitz demo.

Explanation

Here would be my approach:

// Faking an actual request
const makeReq = (prevArg, response) =>
  new Promise((r) => {
    console.log(`Running promise with the prev arg as: ${prevArg}!`);
    setTimeout(r, 1000, { prevArg, response });
  });

// Preparing the sequential requests.
const args = [1, 2, 3, 4, 5];

from(args)
  .pipe(
    // Running the reuqests sequantially.
    mergeScan(
      (acc, crtVal) => {
        // `acc?.response` will refer to the previous response
        // and we're using it for the next request.
        return makeReq(acc?.response, crtVal);
      },
      // The seed(works the same as `reduce`).
      null,
      // Making sure that only one request is run at a time.
      1
    ),

    // Combining all the responses into one object
    // and emitting it after all the requests are done.
    reduce((acc, val, idx) => ({ ...acc, [`request${idx + 1}`]: val }), {})
  )
  .subscribe(console.warn);

Firstly, from(array) will emit each item from the array, synchronously and one by one.

Then, there is mergeScan. It is exactly the result of combining scan and merge. With scan, we can accumulate values(in this case we're using it to have access to the response of the previous request) and what merge does is to allow us to use observables.
To make things a bit easier to understand, think of the Array.prototype.reduce function. It looks something like this:

[].reduce((acc, value) => { return { ...acc }}, /* Seed value */{});

What merge does in mergeScan is to allow us to use the accumulator something like (acc, value) => new Observable(...) instead of return { ...acc }. The latter indicates a synchronous behavior, whereas with the former we can have asynchronous behavior.

Let's go a bit step by step:

  • when 1 is emitted, makeReq(undefined, 1) will be invoked
  • after the first makeReq(from above) resolves, makeReq(1, 2) will be invoked
  • after makeReq(1, 2) resolves, makeReq(2, 3) will be invoked and so on...
user_96326210

Somebody I consulted regarding this matter came up with this solution, I think it's quite elegant:

defer(() => this.getBatch(options)).pipe(
  expand(({ more_items, last_assetid }) =>
    more_items
      ? this.getBatch({ ...options, startAssetId: last_assetid })
      : EMPTY,
  ),
  concatMap((batch) => batch.getSteamItemsWithDescription()),
);

From my understanding the use of expand here is very similar to the use of mergeScan in @Andrei's answer

user_146009720
Related