Rxjs how to get all values that are buffered during a concatMap

If I understand the problem right, I would proceed like this.

First we isolate the source stream. Consider that we use the share operator to make sure that the source$ stream is shared by the other Observables we are going to create later on starting from source$.

const source$ = interval(1000).pipe(share(), take(5));

Then I think that delay you put in the example represents some kind of processing performed by some kind of function, probably a call to an async service. If this is true, then we could have a process function which returns an Observable. A simulated version of this function could be this

function process(val) {
  return of(val).pipe(
    delay(3000),
    tap({
      next: (val) => console.log(val, "processed"),
    })
  );
}

now, if all these assumptions are true, then we can define a dequeueSignal$ Subject, as in your example, and then 2 different streams, a stream that takes just the first element notified by source$ and a stream that take all other elements, like this

const dequeueSignal$ = new Subject<any>();

const first$ = source$.pipe(
  first(),
  concatMap((val) => {
    return process(val).pipe(
      tap({
        complete: () => {
          dequeueSignal$.next(null);
        },
      })
    );
  })
);

const afterFirst$ = source$.pipe(skip(1)).pipe(
  buffer(dequeueSignal$),
  concatMap((val) => {
    return process(val).pipe(
      tap({
        complete: () => {
          dequeueSignal$.next(null);
        },
      })
    );
  })
);

dequeueSignal$ is used to trigger the release of the buffer stored with the buffer operator.

dequeueSignal$ is nexted in the first$ stream once and any time the afterFirst$ stream notifies.

A notification of dequeueSignal$ triggers the release of the buffered items.

Here a stackblitz that shows the code.

Probably a more elegant solution can be implemented as a variation of the mergeMap operator code, but it may look a bit more complex.

The requirement can be covered by a custom operator which leverages concatMap but modifies its behavior as desired.

The custom operator has to keep some state:

• a bufferedNotifications buffer where it stores elements it receives from upstream in case there is already a request on fly
• a processing boolean which, if true, signals that there is a request currently processed, any notification received while processing is added to bufferedNotifications

It is important to note that a custom operator is a function that returns a function of type (observable: Observable) => Observable.

Having said that, let's look at the code of the custom operator

// the custom operator requires a function as input
// this function represents the actual processing that happens in the stream
// a type T is also added for better type checking and support
function concatBufferedMap<T>(processor: (val: T[]) => Observable<T[]>) {
  // as said above, the custom operator returns a function which expects 
  // an Obaservable as input and returns another Obaservable
  return (sourceObservable: Observable<T>) =>
    // the Observable to be returned is constructed using the Observable constructor
    new Observable<T[]>((subscriber) => {
      // this function will be called each time this
      // Observable is subscribed to.

      // here we define the variables holding the state
      // the state is initialized any time the Observable is subscribed 
      let bufferedNotifications = [] as T[];
      let processing = false;

      // the Observable returned by the custom operator transforms the
      // source observable via a pipe and then subscribes to it
      const subscription = sourceObservable
        .pipe(
          // the first thing we do when a value is notified by the 
          // source observable (i.e. by upstream) is to push the value
          // into the buffer
          tap((val) => {
            bufferedNotifications.push(val);
          }),
          // then we leverage concatMap to make sure that we concatenate
          // the various processings
          concatMap(() => {
            // if we are processing a previous request 
            // or if there are no items in the buffer 
            // just complete without notifying anything
            // concatMap requires that the value returned by the function
            // passed in as parameter is an Observable, so in this case
            // we return EMPTY, which is an Observable that does nothing
            // and immediately completes
            if (processing || bufferedNotifications.length === 0) {
              return EMPTY;
            }
            // otherwise, if there is no processing on fly and there are 
            // values in the buffer, we return the Observable created by
            // the processor function and we set the state so that it reflects
            // that we have started processing something
            processing = true;
            // we create a copy of the buffer so that the param passed
            // to the processor function is not altered by other potential
            // values which can be added to the buffer during the execution
            // of processor logic
            const _buffer = [...bufferedNotifications];
            // the buffer is also reset
            bufferedNotifications = [];
            return processor(_buffer).pipe(
              tap({
                complete: () => {
                  console.log("DONE concatMap");
                  // when processor completes its processing, the processing
                  // state variable is set to false
                  processing = false;
                },
              })
            );
          })
        )
        // in the subscription logic, we notify the subscriber (i.e. 
        // downstream) of the values notified by the processor logic
        .subscribe({
          next(value) {
            subscriber.next(value);
          },
          error(err) {
            // We need to make sure we're propagating our errors through.
            subscriber.error(err);
          },
          complete() {
            console.log("subscriber complete");
            subscriber.complete();
          },
        });

      // Return the teardown logic. This will be invoked when
      // the result errors, completes, or is unsubscribed.
      return () => {
        subscription.unsubscribe();
      };
    });
}

This stackblitz shows how the code works and should produce the expected results.