Reactive Programming in C# (.NET): Rx.NET and IObservable

Reactive Programming in C# (.NET): A Complete Guide with Rx.NET

Reactive programming is a development paradigm focused on asynchronous data streams and the propagation of changes. In the C# (.NET) ecosystem, the primary tool for implementing this approach is the Reactive Extensions (Rx.NET) library. It allows you to work with sequences of data that change over time as conveniently as with regular collections via LINQ.

In this article, we will break down the key concepts of reactive programming, examine the main components of Rx.NET, and provide practical code examples for the educational website PythonLib.

Key Concepts of Reactive Programming

At the core of Rx.NET are two interfaces:

• IObservable<T> — a data source (observable sequence) that can send items, errors, or a completion signal.
• IObserver<T> — a subscriber (observer) that receives notifications from the source: OnNext, OnError, OnCompleted.

The key idea: instead of polling the source (pull model), the subscriber reacts to events (push model). This is especially useful for handling UI events, network requests, data streams from sensors, or any other asynchronous operations.

Getting Started with Rx.NET

To use Rx.NET in a C# project, install the NuGet package:

dotnet add package System.Reactive

Example 1: Creating and Subscribing to an Observable

Let's create a simple sequence of numbers that generates values with a delay:

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

class Program{    static async Task Main()    {        var observable = Observable            .Interval(TimeSpan.FromSeconds(1)) // generates 0, 1, 2... every second            .Take(5); // take only the first 5 numbers

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

        // Wait for the stream to complete        await Task.Delay(6000);        subscription.Dispose(); // unsubscribe    }}

Example 2: Filtering and Transformation with LINQ

Rx.NET supports all standard LINQ operators, making the code concise:

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

var processed = values    .Where(x => x % 2 == 0) // only even numbers    .Select(x => x * 10)    // multiply by 10    .Timestamp();           // add a timestamp

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

Advanced Rx.NET Features

Merging Streams

You can combine multiple observable sequences:

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); // merge into one streammerged.Take(6).Subscribe(Console.WriteLine);

Multithreading Management

Rx.NET makes it easy to switch execution contexts:

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);

Observable    .Interval(TimeSpan.FromSeconds(1))    .ObserveOn(uiScheduler) // ensure execution on the UI thread    .Subscribe(x => UpdateUI(x)); // update the interface

Error Handling

Rx.NET provides powerful recovery mechanisms after failures:

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

var safeStr