Programmation réactive en C# (.NET) : Rx.NET et IObservable

Programmation réactive en C# (.NET) : Guide complet avec Rx.NET

\n\n

La programmation réactive est un paradigme de développement axé sur les flux de données asynchrones et la propagation des changements. Dans l'écosystème C# (.NET), le principal outil pour implémenter cette approche est la bibliothèque Reactive Extensions (Rx.NET). Elle permet de travailler avec des séquences de données qui évoluent dans le temps, aussi facilement qu'avec des collections classiques via LINQ.

\n\n

Dans cet article, nous analyserons les concepts clés de la programmation réactive, examinerons les principaux composants de Rx.NET et fournirons des exemples de code pratiques pour le site éducatif PythonLib.

\n\n

Concepts fondamentaux de la programmation réactive

\n\n

Au cœur de Rx.NET se trouvent deux interfaces :

\n

\n
• IObservable<T> — la source de données (séquence observable) qui peut envoyer des éléments, des erreurs ou un signal de fin.
\n
• IObserver<T> — l'abonné (observateur) qui reçoit les notifications de la source : OnNext, OnError, OnCompleted.
\n

\n\n

L'idée clé : au lieu d'interroger la source (modèle pull), l'abonné réagit aux événements (modèle push). Cela est particulièrement utile pour la gestion des événements d'interface utilisateur, des requêtes réseau, des flux de données provenant de capteurs ou de toute autre opération asynchrone.

\n\n

Débuter avec Rx.NET

\n\n

Pour utiliser Rx.NET dans un projet C#, installez le package NuGet :

\n

dotnet add package System.Reactive

\n\n

Exemple 1 : Création et abonnement à un observable

\n

Créons une séquence simple de nombres qui génère des valeurs avec un délai :

\n

using System;\nusing System.Reactive.Linq;\nusing System.Threading.Tasks;\n\nclass Program\n{\n    static async Task Main()\n    {\n        var observable = Observable\n            .Interval(TimeSpan.FromSeconds(1)) // génère 0, 1, 2... chaque seconde\n            .Take(5); // ne prend que les 5 premiers nombres\n\n        var subscription = observable.Subscribe(\n            onNext: number => Console.WriteLine($"Reçu : {number}"),\n            onError: error => Console.WriteLine($"Erreur : {error.Message}"),\n            onCompleted: () => Console.WriteLine("Flux terminé")\n        );\n\n        // Attend la fin du flux\n        await Task.Delay(6000);\n        subscription.Dispose(); // se désabonne\n    }\n}

\n\n

Exemple 2 : Filtrage et transformation avec LINQ

\n

Rx.NET prend en charge tous les opérateurs LINQ standard, rendant le code concis :

\n

var values = Observable.Range(1, 10);\n\nvar processed = values\n    .Where(x => x % 2 == 0) // uniquement les pairs\n    .Select(x => x * 10)    // multiplie par 10\n    .Timestamp();           // ajoute un horodatage\n\nprocessed.Subscribe(x => Console.WriteLine($"[{x.Timestamp}] Valeur : {x.Value}"));

\n\n

Fonctionnalités avancées de Rx.NET

\n\n

Fusion de flux

\n

Vous pouvez combiner plusieurs séquences observables :

\n

var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");\nvar stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(x => $"B{x}");\n\nvar merged = stream1.Merge(stream2); // fusionne en un seul flux\nmerged.Take(6).Subscribe(Console.WriteLine);

\n\n

Gestion du multithreading

\n

Rx.NET permet de changer facilement le contexte d'exécution :

\n

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);\n\nObservable\n    .Interval(TimeSpan.FromSeconds(1))\n    .ObserveOn(uiScheduler) // garantit l'exécution dans le thread UI\n    .Subscribe(x => UpdateUI(x)); // mise à jour de l'interface

\n\n

Gestion des erreurs

\n

Rx.NET offre des mécanismes puissants de récupération après des pannes :

\n

var faultyStream = Observable.Throw<int>(new Exception("Erreur !"));\n\nvar safeStr