RxJs: Merge all values after concapMap so they emit together in the subscribe "next" event?

Question

I have set up an observable to basically call an endpoint for each item in an array but I then need to merge together the items before publishing to to the subscribe "next". Right now, the subscribe "next" is executed N number of times depending on the total in the array, any ideas ? Here is my code

   from(items)
      .pipe(
        concatMap(item => {    
          return this.httpClient.get("https://randomuser.me/api/")
        }),
        tap(items => {
          debugger
        })
      )
      .subscribe(
        result => {
          debugger
        },
        error => {
          debugger
        }
      )

I tried placing a mergeAll before the TAP but it crashes.

Any ideas ?

Thanks in advance

Item is not being used under the HTTP request right now, but its only a test url. But eventually there is an id under item and i will call an endpoint using this id — Ian Gregson, May 08 '19 at 14:18
@martin use forkJoin ? If I place forkJoint after the concatMap pipe, it doesn't seem to work, do you have an example? — Ian Gregson, May 08 '19 at 14:19

Zazaeil · Accepted Answer · 2019-05-08T14:32:43.530

toArray() would accumulate an array of responses,
reduce() can be used in order to reproduce toArray() behavior in a more flexible way: you can apply more sophisticated merging logic (like ignore duplicate entries or whatsnot).

P.S. Not sure, but it might be the case that you've misdesigned something. Rather than use accumulation there you might want a higer order observable such that your consumer does not require array explicitly and underlying logic is more decoupled.

All right, the thing is that you can think that toArray() is quite special case of reduction. Let's start from the signature: reduce(accumulator: function, seed: any): Observable. So seed, the second parameter, is initial value you start from (and in case the stream you're trying to reduce is empty seed is the only value will will get as a result). In case of toArray() seed is just [] - an... empty... array! Then the first parameter, accumulator, it is a function that takes (result, current) - two paramers, where result represents the result so far and the current represents "not yet reduced" element of a stream. So the type of result is array and the type of current can be whatever the type of your stream is, in your case - http responses I guess. (result, current) => { result.Add(current); return result; } - that's the simplest possible implementation, which will give you toArray() behavior. However, you could do more, like decide not to add current if it is outdated or invalid, that's why I told you that reduce() is much more powerful.

P.P.S. If you'd like to know how deep the rabbit's hole is, than reduce (also known like foldl in Haskell and other functional languages) is basically the recursion pattern. Most of the stuff (but definitely not everything!) that defined via recursion, can be rewritten as a reduce with appropariate seed and accumulator parameters; for example, it is quite easy to implement filter throught reduction. See this question to find out more.

Thanks! Do you have an example with reduce ? I have tried toArray and that seems to work. — Ian Gregson, May 08 '19 at 14:22

RxJs: Merge all values after concapMap so they emit together in the subscribe "next" event?

1 Answers1