RxJS: Depending on previous values to recover an object and then join and emit those previous values with the recovered object

Question

I have a case in which I have a bunch of ids that I recover from route.params, like this:

const ids$ = this.route.params.pipe(
  map(params => {
    const modelId = +params['modelId'];
    const deliverId = +params['deliverId'];
    const compositionId = +params['compositionId'];

    return { modelId, deliverId, compositionId };
  })
);

After that, I have to recover the composition object from the server, so I do this:

const getComposition$ = ids$.pipe(
  switchMap(ids =>
    this.compositionsService.getComposition(ids.compositionId)
  )
);

And now I want to have an aggregated object with the two first ids and, instead of the last one (compositionId), the composition object, I mean this:

entities: {
  modelId: number;
  deliverId: number;
  composition: Composition;
};

So I do this:

const aggregated$ = forkJoin(ids$, getComposition$).pipe(
  map(arr => ({
    modelId: arr[0].modelId,
    deliverId: arr[0].deliverId,
    aggregated: arr[1]
  }))
);

And then, I subscribe to it:

const aggregated$.subscribe(aggregated => {
  console.log(aggregated);
});

But it never prints anything on the console. Interestingly enough, if I check if the call to the server is made, it actually is, but the last observable (aggregated$) never emits anything.

What am I doing wrong? Is it there a better way to achieve this?

Answer 1

As you said both ids$ and getComposition$ are in fact this.route.params which comes from Angular. The forkJoin operator needs all its source Observables to emit at least one item and all of them have to complete. And this is what's not happening when you use this.route.params because it's a Subject that is never completed.

So you can use for example take(1) to complete both sources ids$ and getComposition$:

forkJoin(ids$.pipe(take(1)), getComposition$.pipe(take(1))
  ...