Introducción a RxJS y Observables

Observables y suscripciones

Los observables son el núcleo de RxJS y representan streams de datos asíncronos que pueden emitir múltiples valores a lo largo del tiempo. A diferencia de las promesas que resuelven una sola vez, un observable puede emitir zero, uno o múltiples valores antes de completarse.

¿Qué es un Observable?

Un observable es como un canal de comunicación que puede enviar datos de forma asíncrona. Imagina una tubería por la que fluyen datos: el observable es esa tubería, y nosotros podemos "escuchar" lo que sale de ella mediante una suscripción.

// Un observable es un stream que puede emitir valores
// Observable<T> donde T es el tipo de datos que emite

Los observables son lazy, lo que significa que no ejecutan código hasta que alguien se suscribe a ellos. Es como tener una emisora de radio que solo comienza a transmitir cuando hay alguien escuchando.

El patrón Observer

Los observables implementan el patrón Observer, donde tenemos tres tipos de notificaciones:

• next: Se ejecuta cada vez que el observable emite un nuevo valor
• error: Se ejecuta cuando ocurre un error en el stream
• complete: Se ejecuta cuando el observable termina de emitir valores

// Estructura del observer
const observer = {
  next: (value) => console.log('Nuevo valor:', value),
  error: (err) => console.error('Error:', err), 
  complete: () => console.log('Stream completado')
};

Suscripción básica

Para recibir los datos de un observable, debemos suscribirnos a él. La suscripción es el proceso de "escuchar" los valores que emite el observable.

import { Component } from '@angular/core';

@Component({
  selector: 'app-ejemplo',
  template: `
    <div>
      <p>Valor actual: {{ valorActual }}</p>
    </div>
  `
})
export class EjemploComponent {
  valorActual = '';

  constructor() {
    // Ejemplo conceptual de suscripción
    // (veremos la creación en la siguiente sección)
    const miObservable = /* crearemos esto después */;
    
    miObservable.subscribe({
      next: (valor) => {
        this.valorActual = valor;
        console.log('Recibido:', valor);
      },
      error: (error) => {
        console.error('Error en el stream:', error);
      },
      complete: () => {
        console.log('El observable se ha completado');
      }
    });
  }
}

Sintaxis alternativa de suscripción

También podemos usar una sintaxis más compacta cuando solo nos interesa el valor:

// Solo el valor (next)
miObservable.subscribe(valor => {
  console.log('Valor:', valor);
});

// Con manejo de errores
miObservable.subscribe({
  next: valor => console.log(valor),
  error: error => console.error(error)
});

Desuscripción y prevención de memory leaks

Una de las características más importantes de los observables es que debemos desuscribirnos manualmente para evitar memory leaks. Cuando nos suscribimos a un observable, obtenemos un objeto Subscription que podemos usar para cancelar la suscripción.

import { Component, OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs';

@Component({
  selector: 'app-ejemplo',
  template: `<p>Componente con suscripción</p>`
})
export class EjemploComponent implements OnDestroy {
  private subscription!: Subscription;

  constructor() {
    // Guardamos la referencia a la suscripción
    this.subscription = miObservable.subscribe({
      next: (valor) => {
        console.log('Valor recibido:', valor);
      }
    });
  }

  ngOnDestroy() {
    // Muy importante: desuscribirse al destruir el componente
    if (this.subscription) {
      this.subscription.unsubscribe();
    }
  }
}

Múltiples suscripciones

Cuando tenemos varias suscripciones, podemos gestionarlas de forma más eficiente:

import { Component, OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs';

@Component({
  selector: 'app-multiples-suscripciones',
  template: `<p>Múltiples observables</p>`
})
export class MultipleSuscripcionesComponent implements OnDestroy {
  private suscripciones = new Subscription();

  constructor() {
    // Añadimos múltiples suscripciones a una principal
    this.suscripciones.add(
      observable1.subscribe(valor => console.log('Observable 1:', valor))
    );

    this.suscripciones.add(
      observable2.subscribe(valor => console.log('Observable 2:', valor))
    );
  }

  ngOnDestroy() {
    // Desuscribe todas las suscripciones de una vez
    this.suscripciones.unsubscribe();
  }
}

Estados del observable

Los observables pueden estar en diferentes estados:

• Activo: Emitiendo valores normalmente
• Error: Ha ocurrido un error y el stream se detiene
• Completado: Ha terminado de emitir valores exitosamente
• Cerrado: Ya no puede emitir más valores (por error o completado)

const subscription = miObservable.subscribe({
  next: (valor) => {
    console.log('Stream activo, valor:', valor);
  },
  error: (error) => {
    console.log('Stream terminado por error:', error);
    // El observable ya no emitirá más valores
  },
  complete: () => {
    console.log('Stream completado exitosamente');
    // El observable ya no emitirá más valores
  }
});

// Podemos verificar si la suscripción está cerrada
console.log('¿Está cerrada?', subscription.closed);

Importancia en Angular

Los observables son fundamentales en Angular porque se usan en:

• HttpClient: Todas las peticiones HTTP devuelven observables
• Eventos del router: Cambios de ruta y parámetros
• Formularios reactivos: Cambios en los valores y validación
• EventEmitter: Para comunicación entre componentes

// Ejemplo conceptual de uso en Angular
export class MiServicio {
  obtenerDatos() {
    // HttpClient devuelve un Observable
    return this.http.get<any[]>('/api/datos');
  }
}

export class MiComponente implements OnDestroy {
  private subscription!: Subscription;
  
  constructor(private miServicio: MiServicio) {
    this.subscription = this.miServicio.obtenerDatos().subscribe({
      next: (datos) => {
        console.log('Datos recibidos:', datos);
      },
      error: (error) => {
        console.error('Error al obtener datos:', error);
      }
    });
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

La gestión correcta de suscripciones es crucial para mantener el rendimiento de nuestra aplicación Angular y evitar comportamientos inesperados cuando los componentes se destruyen y recrean.

Creación básica

Una vez que comprendemos qué son los observables y cómo suscribirnos a ellos, el siguiente paso es aprender a crearlos desde cero. RxJS proporciona varios métodos para generar observables de forma sencilla, cada uno optimizado para diferentes tipos de datos y casos de uso.

Creación con of()

El método of() es la forma más simple de crear un observable que emite valores estáticos. Perfecto cuando queremos convertir valores conocidos en un stream.

import { of } from 'rxjs';
import { Component, OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs';

@Component({
  selector: 'app-ejemplo-of',
  template: `
    <div>
      <h3>Valores emitidos:</h3>
      @for (valor of valores; track valor) {
        <p>{{ valor }}</p>
      }
    </div>
  `
})
export class EjemploOfComponent implements OnDestroy {
  valores: string[] = [];
  private subscription!: Subscription;

  ngOnInit() {
    // Crear observable que emite varios valores
    const numerosObservable = of(1, 2, 3, 4, 5);
    
    this.subscription = numerosObservable.subscribe({
      next: (numero) => {
        this.valores.push(`Número: ${numero}`);
      },
      complete: () => {
        this.valores.push('¡Completado!');
      }
    });
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

El observable creado con of() emite todos los valores síncronamente y se completa inmediatamente:

// Diferentes tipos de datos
const textosObservable = of('Angular', 'RxJS', 'TypeScript');
const objetoObservable = of({ id: 1, nombre: 'Usuario' });
const booleanObservable = of(true, false, true);

// También podemos pasar un array completo como un solo valor
const arrayObservable = of([1, 2, 3, 4, 5]);

Creación con from()

El método from() convierte estructuras iterables (arrays, promesas, otros observables) en observables. Es especialmente útil para transformar datos existentes.

import { from } from 'rxjs';

@Component({
  selector: 'app-ejemplo-from',
  template: `
    <div>
      <h3>Usuarios:</h3>
      @for (usuario of usuarios; track usuario.id) {
        <div class="usuario-card">
          <h4>{{ usuario.nombre }}</h4>
          <p>Email: {{ usuario.email }}</p>
        </div>
      }
    </div>
  `
})
export class EjemploFromComponent implements OnDestroy {
  usuarios: any[] = [];
  private subscription!: Subscription;

  ngOnInit() {
    // Array de datos que queremos convertir en observable
    const datosUsuarios = [
      { id: 1, nombre: 'Ana García', email: 'ana@email.com' },
      { id: 2, nombre: 'Luis Rodríguez', email: 'luis@email.com' },
      { id: 3, nombre: 'María López', email: 'maria@email.com' }
    ];

    // Convertir array en observable
    const usuariosObservable = from(datosUsuarios);
    
    this.subscription = usuariosObservable.subscribe({
      next: (usuario) => {
        this.usuarios.push(usuario);
      }
    });
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

Una diferencia clave entre of() y from() con arrays:

// of() trata el array como UN SOLO valor
of([1, 2, 3]).subscribe(valor => {
  console.log(valor); // Output: [1, 2, 3] (una sola emisión)
});

// from() emite CADA elemento del array por separado
from([1, 2, 3]).subscribe(valor => {
  console.log(valor); // Output: 1, luego 2, luego 3 (tres emisiones)
});

Conversión de promesas con from()

from() también puede convertir promesas en observables, lo cual es útil para integrar código existente:

@Component({
  selector: 'app-ejemplo-promesa',
  template: `
    <div>
      @if (loading) {
        <p>Cargando datos...</p>
      }
      @if (datos) {
        <p>Datos obtenidos: {{ datos }}</p>
      }
    </div>
  `
})
export class EjemploPromesaComponent implements OnDestroy {
  datos: any = null;
  loading = true;
  private subscription!: Subscription;

  ngOnInit() {
    // Simular una promesa existente
    const promesaDatos = new Promise(resolve => {
      setTimeout(() => {
        resolve('Información importante desde la promesa');
      }, 2000);
    });

    // Convertir promesa en observable
    const datosObservable = from(promesaDatos);
    
    this.subscription = datosObservable.subscribe({
      next: (resultado) => {
        this.datos = resultado;
        this.loading = false;
      }
    });
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

Creación con interval()

El método interval() crea observables que emiten valores secuenciales en intervalos de tiempo regulares. Es perfecto para crear timers o actualizaciones periódicas.

import { interval } from 'rxjs';

@Component({
  selector: 'app-ejemplo-interval',
  template: `
    <div>
      <h3>Contador automático</h3>
      <p class="contador">{{ contador }}</p>
      <button (click)="detener()" [disabled]="detenido">
        Detener contador
      </button>
    </div>
  `
})
export class EjemploIntervalComponent implements OnDestroy {
  contador = 0;
  detenido = false;
  private subscription!: Subscription;

  ngOnInit() {
    // Crear observable que emite cada 1000ms (1 segundo)
    const intervalObservable = interval(1000);
    
    this.subscription = intervalObservable.subscribe({
      next: (numero) => {
        this.contador = numero;
      }
    });
  }

  detener() {
    this.subscription.unsubscribe();
    this.detenido = true;
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

Ejemplo práctico: Sistema de notificaciones

Combinemos estos métodos para crear un sistema simple de notificaciones:

import { of, from, interval } from 'rxjs';

interface Notificacion {
  id: number;
  mensaje: string;
  tipo: 'info' | 'success' | 'warning';
}

@Component({
  selector: 'app-notificaciones',
  template: `
    <div class="notificaciones-container">
      <h3>Sistema de Notificaciones</h3>
      
      <div class="controles">
        <button (click)="enviarNotificacionInmediata()">
          Notificación Inmediata
        </button>
        <button (click)="procesarCola()">
          Procesar Cola
        </button>
        <button (click)="iniciarNotificacionesPeriodicas()" [disabled]="notificacionesActivas">
          Iniciar Notificaciones Periódicas
        </button>
      </div>

      <div class="notificaciones">
        @for (notif of notificaciones; track notif.id) {
          <div class="notificacion" [class]="notif.tipo">
            {{ notif.mensaje }}
          </div>
        }
      </div>
    </div>
  `
})
export class NotificacionesComponent implements OnDestroy {
  notificaciones: Notificacion[] = [];
  notificacionesActivas = false;
  private subscriptions = new Subscription();
  private contadorId = 1;

  // Notificación inmediata con of()
  enviarNotificacionInmediata() {
    const notificacion = {
      id: this.contadorId++,
      mensaje: 'Notificación inmediata enviada',
      tipo: 'success' as const
    };

    const notifObservable = of(notificacion);
    
    this.subscriptions.add(
      notifObservable.subscribe({
        next: (notif) => {
          this.notificaciones.push(notif);
        }
      })
    );
  }

  // Procesar cola de notificaciones con from()
  procesarCola() {
    const colaNotificaciones = [
      { id: this.contadorId++, mensaje: 'Procesando elemento 1', tipo: 'info' as const },
      { id: this.contadorId++, mensaje: 'Procesando elemento 2', tipo: 'info' as const },
      { id: this.contadorId++, mensaje: 'Proceso completado', tipo: 'success' as const }
    ];

    const colaObservable = from(colaNotificaciones);
    
    this.subscriptions.add(
      colaObservable.subscribe({
        next: (notif) => {
          // Simular procesamiento con delay
          setTimeout(() => {
            this.notificaciones.push(notif);
          }, 500);
        }
      })
    );
  }

  // Notificaciones periódicas con interval()
  iniciarNotificacionesPeriodicas() {
    this.notificacionesActivas = true;
    const intervalObservable = interval(3000); // Cada 3 segundos
    
    this.subscriptions.add(
      intervalObservable.subscribe({
        next: (numero) => {
          const notif: Notificacion = {
            id: this.contadorId++,
            mensaje: `Notificación automática #${numero + 1}`,
            tipo: 'warning'
          };
          this.notificaciones.push(notif);
        }
      })
    );
  }

  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

Cuándo usar cada método

La elección del método de creación depende del tipo de datos que necesitemos manejar:

• of(): Cuando tenemos valores estáticos conocidos que queremos emitir inmediatamente
• from(): Para convertir arrays, promesas o estructuras existentes en observables
• interval(): Cuando necesitamos emitir valores de forma periódica o crear contadores

Estos métodos son los fundamentos para comenzar a trabajar con RxJS en Angular, y nos permiten crear streams de datos simples que posteriormente podremos transformar y combinar con operadores más avanzados.