Unexpected concatAll() behavior

[brennarvo]

Hello! I am trying to understand how the concatAll() operator works.

From how I interpreted the documentation, concatAll() will not subscribe to the next observable emitted to the source until the previous observable emitted to the source has completed. However, while putting together a local demonstration to test my understanding of this operator, I found this to not be the case.

My demonstration features two widgets: a calendar widget and a chat widget.

First, the "application" starts by publishing a LOGIN event to tell the widgets to initialize. Next, the calendar widget processes the LOGIN event and publishes a CALENDAR_HAS_BIRTHDAYS event.

However, at this point, I was expecting concatAll() piped onto the source to prevent the CALENDAR_HAS_BIRTHDAYS from from reaching the Chat Widget. I figured that since the Observable associated with the LOGIN event has not completed, concatAll() would wait to emit the CALENDAR_HAS_BIRTHDAYS event to subscribers until it was complete.

Here is the actual code, and a link to the associated Stackblitz:

import { concatAll, Observable, Subject } from 'rxjs';

const source = new Subject<Observable<string>>();
const eventBus = source.pipe(concatAll());

const publish = (event) => {
  source.next(
    new Observable((subscriber) => {
      subscriber.next(event);
      subscriber.complete();
    })
  );
};

// Calendar Widget
eventBus.subscribe({
  next: (event) => {
    if (event === 'LOGIN') {
      console.log('📅 Calendar initializing...');
      publish('CALENDAR_HAS_BIRTHDAYS');
    }
  },
});

// Chat Widget
eventBus.subscribe({
  next: (event) => {
    if (event === 'LOGIN') {
      console.log('💬 Chat initializing...🔄');
    }

    if (event === 'CALENDAR_HAS_BIRTHDAYS') {
      console.log('💬 Chat opening birthday prompt');
    }
  },
});

// LOGIN event
publish('LOGIN');

If you check your logs you'll see the following:

📅 Calendar initializing...
💬 Chat opening birthday prompt
💬 Chat initializing...

However, what I was expecting was this:

📅 Calendar initializing...
💬 Chat initializing...
💬 Chat opening birthday prompt

I'm a little bit lost and unsure where my understanding of concatAll() is falling apart.

Here is how I am interpreting the execution of the code above:

1. publish is called with the LOGIN event
2. An Observable is added to the source which will emit LOGIN when subscribed to
3. concatAll() checks to see if there are any incomplete Observable values emitted to source
4. Because there are no incomplete observables on source, concatAll() subscribes to the observable in step 2
5. concatAll() "pulls" the emitted value out of the inner observable, i.e., LOGIN
6. concatAll() emits the LOGIN event to subscribers of eventBus
7. The calendar widget (📅) processes the LOGIN event
8. The calendar widget (📅) logs "initializing..."
9. The calendar widget (📅) calls publish with a CALENDAR_HAS_BIRTHDAYS event
10. An Observable is added to source which will emit CALENDAR_HAS_BIRTHDAYS when subscribed to
11. concatAll() checks to see if there are any incomplete Observable values emitted to _eventBus

However, I was expecting step 12 to be the following:

12. Because there are incomplete observables, concatAll() does not subscribe to the observable

But in fact, this is not the case. concatAll() does seem to be emitting the CALENDAR_HAS_BIRTHDAYS event to subscribers, hence the "💬 Chat opening birthday prompt" log.

Could somebody help me understand where I'm misinterpreting this operator, and why my demonstration code isn't working as expected?

[brennarvo]

@cartant I read a blog post of yours from 2017 titled "RxJS: Understanding the publish and share Operators" after it was suggested that I use the share() operator for achieving what I'm after, which does in fact seem to give me the desired result that prompted me to create this discussion post:

const source = new Subject<Observable<string>>();
const eventBus = source.pipe(concatAll(), share());

However, I'm still failing to grasp why share solves the problem here. In my code, there are two subscriptions, which does mean that concatAll is executing twice, but I'm not sure why that is problematic in the case of concatAll determining if the previous Observable emitted to source has called complete().

[AronHoxha]

Hi @brennarvo!
From my understanding of rxjs, the reason for this behavior occurs due to the synchronous execution of .next() in your case, as well as the fact that even though the eventBus observable is shared between the two subscriptions they don't share any specific state between them (so you could consider the concatAlls being pretty much different per each subscription)
So consider this modified (but effectively equivalent) code of your example,:

import {concatMap, Observable, Subject, tap} from "rxjs";

const source = new Subject<[Observable<string>, string]>();

const publish = (event) => {
    source.next(
        [new Observable((subscriber) => {
            subscriber.next(event);
            subscriber.complete();
        }), event]
    );
};

// Calendar Widget
source
    .pipe(
        tap(([, val]) => console.log('Cleaner widget ', val)),
        concatMap(([obs]) => obs)
    ).subscribe({
    next: (event) => {
        if (event === 'LOGIN') {
            console.log('📅 Calendar initializing...');
            publish('CALENDAR_HAS_BIRTHDAYS');
        }
    },
});

// Chat Widget
source
    .pipe(
        tap(([, val]) => console.log('Chat widget ', val)),
        concatMap(([obs]) => obs)
    ).subscribe({
    next: (event) => {
        if (event === 'LOGIN') {
            console.log('💬 Chat initializing...🔄');
        }

        if (event === 'CALENDAR_HAS_BIRTHDAYS') {
            console.log('💬 Chat opening birthday prompt');
        }
    },
});

// LOGIN event
publish('LOGIN');

In the logs you'll see:

Cleaner widget  LOGIN
📅 Calendar initializing...
Cleaner widget  CALENDAR_HAS_BIRTHDAYS
Chat widget  CALENDAR_HAS_BIRTHDAYS
💬 Chat opening birthday prompt
Chat widget  LOGIN
💬 Chat initializing...🔄

This shows that the observable getting pushed into the conact first for the Chat widget is the event CALENDAR_HAS_BIRTHDAYS because it's synchronous:

1. .next('LOGIN') is called
2. Cleaner widget observer is executed first as it was subscribed to first (so also the concatAll of this subscription)
3. '.next('CALENDAR_HAS_BIRTHDAYS') is called
4. Cleaner widget observer executed again (here the new observable is also added to the concat all of the clear widget)
5. Chat widget concatAll is pushed the observable with CALENDAR_HAS_BIRTHDAYS and the observer is run with it.
   6 ...

The fact that share fixes your case is pretty much considering that only one subscription is made to the observable having concatAll and here the order is as it was run, and values are pushed as such.

I hope this explanation made sense to some extent.