Implement Observer Pattern in C# with Events and Rx

This article provides a concise guide to implementing the Observer design pattern in C# using both native language events and Reactive Extensions (Rx). You will learn how the built-in event keyword provides a straightforward publish-subscribe mechanism, how System.Reactive (IObservable<T> and IObserver<T>) elevates event handling with powerful LINQ-style composition, and how to choose the right approach for your application.

Implementing the Observer Pattern Using C# Events

C# has built-in language support for the Observer pattern through delegates and events. The publisher (Subject) exposes an event, and subscribers (Observers) attach event handlers to receive notifications.

Implementation

using System;

// 1. Define custom event arguments
public class PriceChangedEventArgs : EventArgs
{
    public decimal NewPrice { get; }

    public PriceChangedEventArgs(decimal newPrice)
    {
        NewPrice = newPrice;
    }
}

// 2. The Subject (Publisher)
public class StockTicker
{
    public event EventHandler<PriceChangedEventArgs>? PriceChanged;

    public void UpdatePrice(decimal newPrice)
    {
        // Notify all registered observers safely
        PriceChanged?.Invoke(this, new PriceChangedEventArgs(newPrice));
    }
}

// 3. The Observers (Subscribers)
public class Program
{
    public static void Main()
    {
        var ticker = new StockTicker();

        // Subscribe using a lambda or method group
        ticker.PriceChanged += (sender, args) =>
        {
            Console.WriteLine($"Observer 1: Price changed to {args.NewPrice:C}");
        };

        ticker.PriceChanged += LogToConsole;

        // Trigger updates
        ticker.UpdatePrice(120.50m);
        ticker.UpdatePrice(122.00m);

        // Unsubscribe to avoid memory leaks
        ticker.PriceChanged -= LogToConsole;
    }

    private static void LogToConsole(object? sender, PriceChangedEventArgs e)
    {
        Console.WriteLine($"Observer 2: Logged price {e.NewPrice:C}");
    }
}

Events are straightforward, require no external dependencies, and integrate directly with the .NET runtime. However, managing unsubscribing manually is necessary to prevent memory leaks, and complex operations such as throttling or combining multiple event sources require custom logic.

Implementing the Observer Pattern Using Reactive Extensions (Rx)

Reactive Extensions formalizes the Observer pattern via the IObservable<T> and IObserver<T> interfaces. Rx treats events as collections over time, allowing you to filter, transform, and aggregate data streams using LINQ.

Install the required package via NuGet:

dotnet add package System.Reactive

Implementation

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

// 1. The Subject using Rx
public class StockTickerRx : IDisposable
{
    // A Subject acts as both an Observer and an Observable
    private readonly Subject<decimal> _priceSubject = new();

    public IObservable<decimal> PriceStream => _priceSubject.AsObservable();

    public void UpdatePrice(decimal newPrice)
    {
        _priceSubject.OnNext(newPrice);
    }

    public void Complete()
    {
        _priceSubject.OnCompleted();
    }

    public void Dispose()
    {
        _priceSubject.Dispose();
    }
}

public class Program
{
    public static void Main()
    {
        using var ticker = new StockTickerRx();

        // 2. Subscribe directly to the stream
        IDisposable subscription1 = ticker.PriceStream.Subscribe(
            price => Console.WriteLine($"Observer 1: Price updated to {price:C}"),
            ex => Console.WriteLine($"Error: {ex.Message}"),
            () => Console.WriteLine("Stream completed.")
        );

        // 3. Transform and filter streams using Rx LINQ operators
        IDisposable subscription2 = ticker.PriceStream
            .Where(price => price > 100m)
            .DistinctUntilChanged()
            .Subscribe(price => Console.WriteLine($"Filtered Observer: High price {price:C}"));

        ticker.UpdatePrice(95.00m);
        ticker.UpdatePrice(105.00m);
        ticker.UpdatePrice(105.00m); // Ignored by subscription2 due to DistinctUntilChanged
        ticker.UpdatePrice(110.00m);

        // Clean up subscriptions
        subscription1.Dispose();
        subscription2.Dispose();
    }
}

Rx handles lifetime management cleanly through IDisposable. Disposing of the subscription token automatically detaches the observer from the observable stream.

Choosing Between Events and Reactive Extensions

  • Use C# Events when:
    • Implementing standard UI event handling (such as WPF, WinForms, or MAUI).
    • Creating lightweight notification systems within a single assembly.
    • The application does not require asynchronous stream composition, debouncing, or complex filtering.
  • Use Reactive Extensions (Rx) when:
    • Working with asynchronous data streams that require querying, throttling, merging, or windowing.
    • You need built-in handling for completion (OnCompleted) and exceptions (OnError).
    • You want declarative subscription lifecycle management via IDisposable.