Reactive Programming in C# (.NET)

Reactive Programming in C# (.NET): Observable, Rx.NET and Working with Data Streams

Reactive Programming is a paradigm focused on asynchronous processing of data streams and propagation of changes. In the C# (.NET) ecosystem, this concept is implemented through the Rx.NET (Reactive Extensions) library and the IObservable<T> / IObserver<T> interfaces. Unlike the classical imperative approach, reactive programming allows you to declaratively describe how the system should react to events without blocking the main execution thread.

On the PythonLib page, we will explore the key components of reactive programming in C#, including Observable sequences, LINQ operators for stream transformation, and practical usage examples.

Core Concepts of Reactive Programming in C#

Reactive programming is based on the Observer pattern, where:

Observable — the data source (event stream).
Observer — the subscriber that reacts to incoming data.
Subscription — the mechanism for managing the subscription (unsubscribing from the stream).

In .NET, these concepts are represented by the IObservable<T> and IObserver<T> interfaces, but in practice the Rx.NET library (NuGet package System.Reactive) is more commonly used, providing a rich set of operators for working with sequences.

Installing Rx.NET

To get started, add the NuGet package to your project:

Install-Package System.Reactive

Or via .NET CLI:

dotnet add package System.Reactive

Creating an Observable and Subscribing to a Data Stream

Let's look at a simple example: we'll create an Observable that generates numbers from 1 to 5 with a delay, and subscribe to it.

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

class Program{    static async Task Main()    {        // Create an Observable sequence        var observable = Observable.Interval(TimeSpan.FromSeconds(1))            .Select(x => x + 1) // Start from 1            .Take(5);           // Take only 5 elements

        // Subscribe to the stream        var subscription = observable.Subscribe(            onNext: value => Console.WriteLine($"Received: {value}"),            onError: error => Console.WriteLine($"Error: {error.Message}"),            onCompleted: () => Console.WriteLine("Stream completed")        );

        // Wait for completion (in real code use await or blocking)        await Task.Delay(6000);

        // Unsubscribe (release resources)        subscription.Dispose();    }}

LINQ Operators for Stream Transformation

One of the main advantages of Rx.NET is the ability to apply operators familiar from LINQ to data streams: Where, Select, Aggregate, and others. This allows you to declaratively describe transformations.

Example: Filtering and Transformation

using System;using System.Reactive.Linq;

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

var processed = numbers    .Where(n => n % 2 == 0)    // Only even numbers    .Select(n => n * n)        // Square them    .Do(n => Console.WriteLine($"Processed: {n}")); // Side effect

processed.Subscribe(    onNext: n => Console.WriteLine($"Result: {n}"),    onCompleted: () => Console.WriteLine("Done"));