Programación reactiva en C# (.NET): Rx.NET e IObservable

Programación reactiva en C# (.NET): Guía completa con Rx.NET

La programación reactiva es un paradigma de desarrollo orientado a flujos de datos asíncronos y la propagación de cambios. En el ecosistema de C# (.NET), la herramienta principal para implementar este enfoque es la biblioteca Reactive Extensions (Rx.NET). Permite trabajar con secuencias de datos que cambian en el tiempo, de la misma manera conveniente que con colecciones ordinarias a través de LINQ.

En este artículo, analizaremos los conceptos clave de la programación reactiva, revisaremos los componentes principales de Rx.NET y proporcionaremos ejemplos prácticos de código para el sitio educativo PythonLib.

Conceptos fundamentales de la programación reactiva

En el núcleo de Rx.NET se encuentran dos interfaces:

• IObservable<T> — fuente de datos (secuencia observable) que puede enviar elementos, errores o una señal de finalización.
• IObserver<T> — suscriptor (observer) que recibe notificaciones de la fuente: OnNext, OnError, OnCompleted.

La idea clave: en lugar de consultar la fuente (modelo pull), el suscriptor reacciona a los eventos (modelo push). Esto es especialmente útil para manejar eventos de UI, solicitudes de red, flujos de datos de sensores o cualquier otra operación asíncrona.

Primeros pasos con Rx.NET

Para usar Rx.NET en un proyecto de C#, instale el paquete NuGet:

dotnet add package System.Reactive

Ejemplo 1: Creación y suscripción a un observable

Creemos una secuencia simple de números que genera valores con un retardo:

using System;using System.Reactive.Linq;using System.Threading.Tasks;

class Program{    static async Task Main()    {        var observable = Observable            .Interval(TimeSpan.FromSeconds(1)) // genera 0, 1, 2... cada segundo            .Take(5); // tomamos solo los primeros 5 números

        var subscription = observable.Subscribe(            onNext: number => Console.WriteLine($"Recibido: {number}"),            onError: error => Console.WriteLine($"Error: {error.Message}"),            onCompleted: () => Console.WriteLine("Flujo completado")        );

        // Esperamos la finalización del flujo        await Task.Delay(6000);        subscription.Dispose(); // cancelamos la suscripción    }}

Ejemplo 2: Filtrado y transformación con LINQ

Rx.NET admite todos los operadores estándar de LINQ, lo que hace que el código sea conciso:

var values = Observable.Range(1, 10);

var processed = values    .Where(x => x % 2 == 0) // solo pares    .Select(x => x * 10)    // multiplicamos por 10    .Timestamp();           // agregamos una marca de tiempo

processed.Subscribe(x => Console.WriteLine($"[{x.Timestamp}] Valor: {x.Value}"));

Funcionalidades avanzadas de Rx.NET

Combinación de flujos

Puede combinar varias secuencias observables:

var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");var stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(x => $"B{x}");

var merged = stream1.Merge(stream2); // combinamos en un solo flujomerged.Take(6).Subscribe(Console.WriteLine);

Gestión de la concurrencia

Rx.NET permite cambiar fácilmente el contexto de ejecución:

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);

Observable    .Interval(TimeSpan.FromSeconds(1))    .ObserveOn(uiScheduler) // garantizamos la ejecución en el hilo de UI    .Subscribe(x => UpdateUI(x)); // actualización de la interfaz

Manejo de errores

Rx.NET proporciona potentes mecanismos de recuperación ante fallos:

var faultyStream = Observable.Throw<int>(new Exception("¡Error!"));

var safeStr