Programação Reativa em C# (.NET): Rx.NET e IObservable

Programação Reativa em C# (.NET): Guia Completo com Rx.NET

Programação reativa é um paradigma de desenvolvimento focado em fluxos de dados assíncronos e propagação de mudanças. No ecossistema C# (.NET), a principal ferramenta para implementar essa abordagem é a biblioteca Reactive Extensions (Rx.NET). Ela permite trabalhar com sequências de dados que mudam ao longo do tempo, tão facilmente quanto com coleções comuns através do LINQ.

Neste artigo, vamos explorar os conceitos-chave da programação reativa, analisar os principais componentes do Rx.NET e apresentar exemplos práticos de código para o site educacional PythonLib.

Conceitos Fundamentais da Programação Reativa

Na base do Rx.NET estão duas interfaces:

• IObservable<T> — fonte de dados (sequência observável) que pode enviar elementos, erros ou um sinal de conclusão.
• IObserver<T> — assinante (observador) que recebe notificações da fonte: OnNext, OnError, OnCompleted.

A ideia central: em vez de consultar a fonte (modelo pull), o assinante reage a eventos (modelo push). Isso é especialmente útil para tratar eventos de UI, requisições de rede, fluxos de dados de sensores ou qualquer outra operação assíncrona.

Primeiros Passos com Rx.NET

Para usar Rx.NET em um projeto C#, instale o pacote NuGet:

dotnet add package System.Reactive

Exemplo 1: Criando e assinando um observable

Vamos criar uma sequência simples de números que gera valores com atraso:

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

class Program{    static async Task Main()    {        var observable = Observable            .Interval(TimeSpan.FromSeconds(1)) // gera 0, 1, 2... a cada segundo            .Take(5); // pega apenas os primeiros 5 números

        var subscription = observable.Subscribe(            onNext: number => Console.WriteLine($"Recebido: {number}"),            onError: error => Console.WriteLine($"Erro: {error.Message}"),            onCompleted: () => Console.WriteLine("Fluxo concluído")        );

        // Aguarda a conclusão do fluxo        await Task.Delay(6000);        subscription.Dispose(); // cancela a assinatura    }}

Exemplo 2: Filtragem e transformação com LINQ

Rx.NET suporta todos os operadores LINQ padrão, tornando o código conciso:

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

var processed = values    .Where(x => x % 2 == 0) // apenas pares    .Select(x => x * 10)    // multiplica por 10    .Timestamp();           // adiciona carimbo de tempo

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

Recursos Avançados do Rx.NET

Combinação de fluxos

Você pode combinar várias sequências observáveis:

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); // combina em um único fluxomerged.Take(6).Subscribe(Console.WriteLine);

Gerenciamento de multithreading

Rx.NET permite alternar facilmente o contexto de execução:

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);

Observable    .Interval(TimeSpan.FromSeconds(1))    .ObserveOn(uiScheduler) // garante execução na thread da UI    .Subscribe(x => UpdateUI(x)); // atualização da interface

Tratamento de erros

Rx.NET oferece mecanismos poderosos para recuperação de falhas:

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

var safeStr