Buffer emits when source closes

[nous-]

Hello,

Not clear if this is an issue, or I don't understand the rxjs contract. My use case is kicking off multiple observables in parallel say for a download, where some operations cannot continue processing until other operations have completed. I expected the buffer operator below to emit 2 after base completes a second later, however it emits immediately because the source to the buffer operator completed (see source). Per the buffer documentation:

Buffers the source Observable values until closingNotifier emits.

Base never emitted, so this was unexpected.

const base = timer(1000);
const dep = of(2).pipe(
   buffer(base),
   mergeAll(),
   tap(x => console.log(x)
);

forkJoin([
	base, dep, // third, forth..
]).subscribe(() => {
	//Done
})

[josepot]

Hi @nous- !

I understand why you think that this behavior is unexpected.

However, this is the behavior that I was expecting and I was happy to see that there is a test that takes this case into account.

Although, I think that the sentence of the docs should be improved, because it's not accurate to say that the resulting observable emits only when the notifier does... it could also emit if the source completes, even if the notifier has not emitted yet.

[nous-]

How might I get the behavior I'm looking for? Would I need to make my own operator, or is there an operator that won't report complete downstream? Something else?

[josepot]

TBH I'm not sure if I understand what you are after, but if what you want is to prevent an Observable from completing then you could do something like:

const observableThatDoesNotComplete$ = concat(
  observableThatCompletes$,
  NEVER
)

In your example that would be:

concat(of(2), NEVER).pipe(
   buffer(base),
   mergeAll(),
   tap(x => console.log(x)
)

Although, perhaps the operator that you are after is delayWhen, rather than buffer? It's hard to say with so little context.

[nous-]

In attempting to clarify myself I found the confusion. Essentially I wanted to kick off all downloads simultaneously, and begin processing the download result, however some downloads are dependent on the completion of others, so I need to wait for dependencies as well. Here revised:

const base = timer(500).pipe(share());
const downloadA = of('val1');

forkJoin([
   base,
   forkJoin([downloadA, base]).pipe(tap(x => console.log("Begin processing", x[0])))
   // ...additional varied dependencies
]).subscribe()

Can you see how I might clean this up further, or an alternative approach?