Sebastian Gomez

Feb 24, 2022

Updated: Jun 24, 2026

Implementando concat para concatenar observables

En este post aprenderemos cómo implementar una función llamada concat que nos permita transformar diferentes observables en uno solo al concatenarlos.

Anteriormente exploramos la implementación de los operadores filter, map y of en observables. En esta ocasión abordaremos cómo concatenar observables. Esto es especialmente útil cuando deseamos combinar datos de una petición HTTP (por ejemplo, a través de fromEvent) con datos almacenados en arreglos o variables.

Antes de construir nuestra propia versión, veamos cómo se vería esto usando RxJS real. Una aclaración importante: en el resto del post implementaremos un Observable desde cero a modo ilustrativo, así que el siguiente ejemplo solo sirve para mostrar la idea con la librería tal y como se usa hoy.

// En RxJS moderno (7.x) los operadores son funciones independientes que importamos desde 'rxjs'
import { fromEvent, concat, of } from 'rxjs';

// Crea un observable "sourceOne" a partir de un evento.
// Este evento podria emitir, por ejemplo, los valores 1, 2, 3.
const sourceOne = fromEvent(/** algo que nos dé 1, 2, 3, por ejemplo */);

// Crea otro observable "sourceTwo" que emite los valores 4, 5, 6 de forma síncrona
const sourceTwo = of(4, 5, 6);

// Crea un observable "source" que concatena "sourceOne" y "sourceTwo".
// "sourceTwo" comenzará a emitir solo después de que "sourceOne" haya completado.
const source = concat(sourceOne, sourceTwo);

// Suscribe un observador al observable "source".
// El observador manejará los valores emitidos por ambos observables concatenados.
source.subscribe(/** lo que queramos hacer aquí */);

Nota: La API estática del estilo Observable.fromEvent(...) y Observable.concat(...) pertenece a RxJS 5 y fue eliminada en RxJS 6 en adelante. La versión vigente es la 7.x, donde todo se importa como funciones independientes desde rxjs.fromEvent antes de copiar el snippet a producción.

Implementación de concat

Para implementar concat debemos crear un método estático que genere un nuevo observable a partir de los observables que reciba como parámetros. Aquí está un esqueleto básico para la clase Observable:

// Define una clase llamada Observable
class Observable {
  // Define un método estático "concat" que toma un número variable de observables como argumento
  static concat(...observables) {
    // Retorna una nueva instancia de Observable
    return new Observable(function subscribe(observer) {
      // Aquí se implementará la lógica de combinación de observables.
      // Esta función se llamará cuando se suscriba un observador
      // al observable creado a partir de "concat".
    });
  }
  //...
}

Luego necesitamos tomar el primero de los observables, iterar sobre todos sus elementos y, al finalizar, continuar con el siguiente hasta que no queden más. Usaremos recursividad para implementar esta lógica:

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 {
        // Toma el siguiente observable de los parámetros.
        let observable = myObservables.shift();

        sub = observable.subscribe({
          next(v) {
            observer.next(v);
          },
          error(err) {
            observer.error(err);
            sub.unsubscribe();
          },
          complete() {
            // Al completar el observable actual, continuamos con el siguiente.
            processObservable();
          },
        });
      }
    };

    processObservable();

    // Devolvemos un objeto con un método unsubscribe() real.
    // Así el consumidor puede cancelar la suscripción interna activa en cualquier momento.
    return {
      unsubscribe() {
        if (sub) {
          sub.unsubscribe();
        }
      },
    };
  });
}

Fíjate en el detalle del cierre. Una versión ingenua devolvería return { sub }, pero eso captura el valor de sub en ese instante y no la suscripción interna que esté activa cuando el consumidor decida cancelar. Como sub se reasigna en cada processObservable(), lo correcto es devolver un objeto con un método unsubscribe() que, al invocarse, cancele la suscripción interna vigente en ese momento.

Conclusión

En este post aprendimos a:

• Implementar la función concat para concatenar observables.
• Utilizar la recursividad para procesar varios observables.
• Combinar datos de diferentes fuentes, como peticiones HTTP y variables.
• Devolver una suscripción con un unsubscribe() que se pueda cancelar correctamente.

Ejercicios para practicar

1. Implementa la función concat en un proyecto propio y prueba a combinar diferentes observables.
2. Modifica la función concat para que, en caso de error en un observable, continúe con el siguiente en lugar de detener el proceso.
3. Experimenta con la concatenación de más de dos observables.

Resumen en 3 puntos

1. La función concat nos permite combinar diferentes observables en uno solo.
2. La recursividad es una herramienta útil para procesar una lista de observables.
3. Devolver un unsubscribe() real, y no return { sub }, es lo que hace que nuestro operador sea verdaderamente utilizable.

Espero que este post te haya sido útil y te ayude a entender mejor cómo implementar la función concat para concatenar observables. Si tienes alguna pregunta, inquietud o deseas compartir tus logros con esta implementación, no dudes en dejar un comentario en la sección de comentarios. Si te gustó este contenido, compártelo con tus amigos y colegas a través de las redes sociales. ¡Buena suerte en tus proyectos y hasta la próxima!