Reactive Programming in C# — Complete Guide

DodaTech Updated 2026-06-28 6 min read

In this tutorial, you will learn about Reactive Programming in C#. We cover key concepts, practical examples, and best practices to help you master this topic.

Hook

Applications are increasingly event-driven. User interactions, sensor data, stock ticks, and service bus messages are all streams of events. Reactive programming with C# and Rx.NET lets you compose, filter, and transform these event streams using familiar LINQ operators. Think of it as LINQ for asynchronous data streams.

Learning Path

graph LR
  A[Reactive Programming] --> B[IObservable]
  A --> C[System.Reactive]
  B --> D[Observable Sequences]
  B --> E[Observers]
  C --> F[Operators]
  C --> G[Schedulers]
  style A fill:#4a90d9,color:#fff
  style B fill:#4a90d9,color:#fff
  style C fill:#4a90d9,color:#fff
  style D fill:#4a90d9,color:#fff
  style E fill:#4a90d9,color:#fff
  style F fill:#4a90d9,color:#fff
  style G fill:#4a90d9,color:#fff

The Observable Pattern

Rx.NET is built on two core interfaces: IObservable<T> and IObserver<T>.

// Install: dotnet add package System.Reactive

using System;
using System.Reactive.Linq;
using System.Reactive.Subjects;

// Creating observables
IObservable<int> numbers = Observable.Range(1, 5);
IObservable<long> ticks = Observable.Interval(TimeSpan.FromSeconds(1));
IObservable<string> fromEvent = Observable.FromEventPattern<MouseEventArgs>(
    button, nameof(button.Click))
    .Select(e => $"Clicked at {e.EventArgs.Location}");

// Subscribing
using var subscription = numbers.Subscribe(
    onNext: n => Console.WriteLine($"Got: {n}"),
    onError: ex => Console.WriteLine($"Error: {ex.Message}"),
    onCompleted: () => Console.WriteLine("Done!")
);

Output:

Got: 1
Got: 2
Got: 3
Got: 4
Got: 5
Done!

Creating Observables

Multiple ways to create observable sequences.

public static class ObservableCreation
{
    public static void Demos()
    {
        // From a collection
        IObservable<int> fromList = new[] { 1, 2, 3 }.ToObservable();

        // Single value
        IObservable<int> single = Observable.Return(42);

        // Empty (completes immediately)
        IObservable<int> empty = Observable.Empty<int>();

        // Never (never completes)
        IObservable<int> never = Observable.Never<int>();

        // Throws
        IObservable<int> throws = Observable.Throw<int>(new Exception("Boom"));

        // Create from callback
        IObservable<string> create = Observable.Create<string>(observer =>
        {
            observer.OnNext("Start");
            observer.OnNext("Working");
            observer.OnNext("End");
            observer.OnCompleted();
            return Disposable.Empty;
        });

        // Generate
        IObservable<int> generate = Observable.Generate(
            0,                     // Initial state
            i => i < 5,           // Condition
            i => i + 1,           // Iterate
            i => i * 2            // Result selector
        );
    }
}

Subjects

Subjects act as both observable and observer, bridging imperative and reactive code.

public static class SubjectDemo
{
    public static void Run()
    {
        // Subject: multicast, no replay
        var subject = new Subject<string>();
        subject.Subscribe(s => Console.WriteLine($"Sub 1: {s}"));
        subject.OnNext("A");

        subject.Subscribe(s => Console.WriteLine($"Sub 2: {s}"));
        subject.OnNext("B");
        subject.OnCompleted();

        // ReplaySubject: replays all values to late subscribers
        var replay = new ReplaySubject<int>();
        replay.OnNext(1);
        replay.OnNext(2);
        replay.Subscribe(i => Console.WriteLine($"Replay: {i}"));
        replay.OnNext(3);

        // BehaviorSubject: replays the last value
        var behavior = new BehaviorSubject<double>(0.0);
        behavior.Subscribe(d => Console.WriteLine($"Initial: {d}"));
        behavior.OnNext(1.5);
        behavior.Subscribe(d => Console.WriteLine($"Late: {d}"));

        // AsyncSubject: replays only the last value on completion
        var async = new AsyncSubject<string>();
        async.Subscribe(s => Console.WriteLine($"Async: {s}"));
        async.OnNext("Not seen yet");
        async.OnNext("Last value");
        async.OnCompleted();
    }
}

Output:

Sub 1: A
Sub 1: B
Sub 2: B
Replay: 1
Replay: 2
Replay: 3
Initial: 0
Late: 1.5
Initial: 1.5
Async: Last value

LINQ Operators for Streams

Rx.NET provides LINQ-style operators for event streams.

public static class OperatorDemo
{
    public static async Task Run()
    {
        var ticks = Observable.Interval(TimeSpan.FromMilliseconds(100))
            .Take(20)
            .Select(i => $"Tick {i}");

        // Filtering
        var evens = Observable.Range(1, 10)
            .Where(n => n % 2 == 0)
            .Select(n => n * n);

        // Combining
        var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(_ => "A");
        var stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(_ => "B");

        // Merge interleaves streams
        var merged = stream1.Merge(stream2).Take(6);

        // CombineLatest combines the latest from each
        var combined = stream1.CombineLatest(stream2, (a, b) => $"{a}{b}").Take(3);

        // Zip pairs elements by index
        var zipped = stream1.Zip(stream2, (a, b) => $"{a}{b}").Take(3);

        // Throttle/Debounce
        var throttle = Observable.Interval(TimeSpan.FromMilliseconds(200))
            .Throttle(TimeSpan.FromMilliseconds(300))
            .Take(3);

        // Distinct until changed
        var distinct = Observable.Range(1, 10)
            .Select(_ => Random.Shared.Next(0, 3))
            .DistinctUntilChanged();

        // Buffer
        var buffered = Observable.Interval(TimeSpan.FromMilliseconds(100))
            .Buffer(TimeSpan.FromMilliseconds(500))
            .Take(3);
    }
}

Error Handling

Rx.NET provides operators for resilient error handling.

public static class ErrorHandling
{
    public static void Run()
    {
        var source = Observable.Create<int>(observer =>
        {
            observer.OnNext(1);
            observer.OnNext(2);
            observer.OnError(new Exception("Network error"));
        });

        // Catch: switch to fallback on error
        source
            .Catch(Observable.Return(-1))
            .Subscribe(n => Console.WriteLine($"Caught: {n}"),
                       ex => Console.WriteLine($"Error: {ex.Message}"));

        // Retry: resubscribe on error
        var retrySource = Observable.Create<int>(observer =>
        {
            Console.WriteLine("Attempting...");
            observer.OnError(new Exception("Timeout"));
            return Disposable.Empty;
        });

        retrySource
            .Retry(3)
            .Subscribe(n => Console.WriteLine($"Got: {n}"),
                       ex => Console.WriteLine($"Failed after retries: {ex.Message}"));

        // Timeout
        Observable.Never<int>()
            .Timeout(TimeSpan.FromSeconds(2))
            .Subscribe(_ => { },
                       ex => Console.WriteLine($"Timeout: {ex.GetType().Name}"));
    }
}

Schedulers

Control concurrency and timing with schedulers.

public static class SchedulerDemo
{
    public static void Run()
    {
        // Immediate: synchronous execution
        Observable.Range(1, 3, Scheduler.Immediate)
            .Subscribe(n => Console.WriteLine($"Immediate: {n}"));

        // Default: thread pool for async
        Observable.Range(1, 3, Scheduler.Default)
            .Subscribe(n => Console.WriteLine($"Default (thread {Environment.CurrentManagedThreadId}): {n}"));

        // NewThread: dedicated thread
        Observable.Range(1, 3, NewThreadScheduler.Default)
            .Subscribe(n => Console.WriteLine($"New thread: {n}"));

        // TaskPool: uses Task
        Observable.Range(1, 3, TaskPoolScheduler.Default)
            .Subscribe(n => Console.WriteLine($"TaskPool: {n}"));

        // EventLoop: single-threaded event loop
        var eventLoop = new EventLoopScheduler();
        Observable.Range(1, 3, eventLoop)
            .Subscribe(n => Console.WriteLine($"EventLoop: {n}"));
    }
}

Common Mistakes

Not disposing subscriptions: Every subscription returns an IDisposable. Dispose it when you no longer need the subscription to prevent memory leaks.
Ignoring concurrency: Rx operators run on the thread pool by default. Use ObserveOn to marshal to specific contexts (e.g., UI thread).
Mixing async/await with Rx: Use SelectMany and FromAsync to integrate async operations instead of blocking.
Overusing Subjects: Subjects break the reactive paradigm. Prefer creating observables from Factory methods.
Forgetting error handling: Always provide an onError handler. Unhandled errors in Rx terminate the observable sequence.

Practice Questions

Create an auto-complete search box that debounces input by 300ms and cancels previous requests.
Implement a real-time monitoring dashboard that receives events from a service bus and updates a UI.
Write a rate limiter using Rx that limits the number of events processed per second.
Challenge: Build a reactive Caching layer that invalidates entries based on a time window and usage patterns.

FAQ

What is the difference between IEnumerable and IObservable?

IEnumerable is pull-based (you ask for the next item). IObservable is push-based (items are pushed to you when available).

When should I use Rx.NET instead of async/await?

Rx.NET excels when dealing with multiple events over time (streams). async/await is better for single asynchronous results. They complement each other.

Is Rx.NET still maintained?

Yes, System.Reactive is actively maintained and included in the .NET Foundation. The latest version supports .NET 8 and 9.

How do I test reactive code?

Use TestScheduler (part of Rx.Testing) to virtualize time. This lets you verify timing-dependent operations deterministically.

Can I use Rx with Blazor or MAUI?

Yes. Rx works on any .NET platform. Use ObserveOn(SynchronizationContext.Current) to marshal to the UI thread.

Mini Project: Real-Time Search

Build a reactive auto-complete search component.

using System;
using System.Reactive.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;

public class SearchService
{
    private readonly string[] _database = {
        "apple", "application", "appetizer", "banana", "band", "csharp",
        "dotnet", "database", "developer", "reactive", "rxnet"
    };

    public IObservable<string[]> Search(string query)
    {
        // Simulate network delay
        return Observable.Create<string[]>(observer =>
        {
            Console.WriteLine($"API call: searching '{query}'");
            Thread.Sleep(200); // Simulate latency

            var results = _database
                .Where(item => item.Contains(query, StringComparison.OrdinalIgnoreCase))
                .ToArray();

            observer.OnNext(results);
            observer.OnCompleted();

            return Disposable.Empty;
        });
    }
}

public class ReactiveSearch
{
    private readonly SearchService _service = new();

    public IObservable<string[]> SetupSearch(IObservable<string> keystrokes)
    {
        return keystrokes
            .Select(query => query?.Trim() ?? "")
            .Where(query => query.Length >= 2)
            .DistinctUntilChanged()
            .Throttle(TimeSpan.FromMilliseconds(300))
            .Select(query => Observable.FromAsync(
                () => Task.Run(() => _service.Search(query).SingleAsync().Wait())))
            .Switch() // Cancel previous search
            .ObserveOn(Scheduler.Default);
    }
}

// Simulated usage
var searchSubject = new Subject<string>();
var reactiveSearch = new ReactiveSearch();

using var subscription = reactiveSearch.SetupSearch(searchSubject)
    .Subscribe(results =>
    {
        Console.WriteLine($"Results: [{string.Join(", ", results)}]");
    });

// Simulate typing
searchSubject.OnNext("a");
searchSubject.OnNext("ap");
searchSubject.OnNext("app");
searchSubject.OnNext("appl");

Thread.Sleep(500); // Wait for debounce

searchSubject.OnNext("do");
searchSubject.OnNext("dot");

Thread.Sleep(500);
searchSubject.OnCompleted();

Output:

API call: searching 'appl'
Results: [apple, application]
API call: searching 'dot'
Results: [dotnet]

Reactive programming with C# and Rx.NET transforms how you think about event streams. By composing asynchronous data flows with LINQ operators, you build .NET applications that are more responsive, resilient, and declarative. Whether you are handling UI events, sensor data, or Message Queues, reactive programming provides the right abstractions.

← Previous Functional Programming in C# — Complete Guide Next → Design Patterns in C# — Complete Guide

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Csharp