Sebastian Gomez

Apr 14, 2023

Updated: Jun 25, 2026

Implementing concat to concatenate observables

In this post we will learn how to implement a function called concat that lets us turn several observables into a single one by concatenating them.

Earlier we explored implementing the filter, map, and of operators on observables. This time we will tackle how to concatenate observables. This is especially useful when we want to combine data from an HTTP request (for example, through fromEvent) with data stored in arrays or variables.

Before building our own version, let's see how this would look using real RxJS. One important clarification: in the rest of the post we will implement an Observable from scratch for illustration, so the example below only serves to show the idea with the library as it is used today.

// In modern RxJS (7.x) operators are standalone functions we import from 'rxjs'
import { fromEvent, concat, of } from 'rxjs';

// Create an observable "sourceOne" from an event.
// This event could emit, for example, the values 1, 2, 3.
const sourceOne = fromEvent(/** something that gives us 1, 2, 3, for example */);

// Create another observable "sourceTwo" that emits the values 4, 5, 6 synchronously
const sourceTwo = of(4, 5, 6);

// Create an observable "source" that concatenates "sourceOne" and "sourceTwo".
// "sourceTwo" will start emitting only after "sourceOne" has completed.
const source = concat(sourceOne, sourceTwo);

// Subscribe an observer to the "source" observable.
// The observer will handle the values emitted by both concatenated observables.
source.subscribe(/** whatever we want to do here */);

Note: The static API of the Observable.fromEvent(...) and Observable.concat(...) style belongs to RxJS 5 and was removed in RxJS 6 onward. The current version is 7.x, where everything is imported as standalone functions from rxjs.fromEvent signature before copying the snippet into production.

Implementing concat

To implement concat we need to create a static method that builds a new observable from the observables it receives as parameters. Here is a basic skeleton for the Observable class:

// Define a class called Observable
class Observable {
  // Define a static method "concat" that takes a variable number of observables as arguments
  static concat(...observables) {
    // Return a new Observable instance
    return new Observable(function subscribe(observer) {
      // The observable combination logic will go here.
      // This function will be called when an observer subscribes
      // to the observable created from "concat".
    });
  }
  //...
}

Then we need to take the first of the observables, iterate over all of its values, and when it finishes, continue with the next one until none are left. We will use recursion to implement this logic:

static concat(...observables) {
  return new Observable(function subscribe(observer) {
    let myObservables = observables.slice();
    let sub = null;

    let processObservable = () => {
      if (myObservables.length === 0) {
        observer.complete();
      } else {
        // Take the next observable from the parameters.
        let observable = myObservables.shift();

        sub = observable.subscribe({
          next(v) {
            observer.next(v);
          },
          error(err) {
            observer.error(err);
            sub.unsubscribe();
          },
          complete() {
            // When the current observable completes, we continue with the next one.
            processObservable();
          },
        });
      }
    };

    processObservable();

    // We return an object with a real unsubscribe() method.
    // This way the consumer can cancel the active inner subscription at any time.
    return {
      unsubscribe() {
        if (sub) {
          sub.unsubscribe();
        }
      },
    };
  });
}

Notice the detail in the teardown. A naive version would return return { sub }, but that captures the value of sub at that instant and not the inner subscription that is active when the consumer decides to cancel. Since sub is reassigned on every processObservable() call, the correct approach is to return an object with an unsubscribe() method that, when invoked, cancels whichever inner subscription is active at that moment.

Conclusion

In this post we learned how to:

• Implement the concat function to concatenate observables.
• Use recursion to process several observables.
• Combine data from different sources, such as HTTP requests and variables.
• Return a subscription with an unsubscribe() that can be canceled correctly.

Exercises to practice

1. Implement the concat function in your own project and try combining different observables.
2. Modify the concat function so that, when one observable errors, it continues with the next one instead of stopping the whole process.
3. Experiment with concatenating more than two observables.

3-point summary

1. The concat function lets us combine different observables into a single one.
2. Recursion is a useful tool for processing a list of observables.
3. Returning a real unsubscribe(), rather than return { sub }, is what makes our operator genuinely usable.

I hope this post was useful to you and helps you better understand how to implement the concat function to concatenate observables. If you have any questions, concerns, or want to share what you built with this implementation, feel free to leave a comment in the comments section. If you liked this content, share it with your friends and colleagues on social media. Good luck with your projects, and until next time!