Observer Pattern in Depth: event, IObservable<T>, IChangeToken and Domain Events

Version note: The code in this post was built and run with C# 14 and .NET 10.

Guess First

When an order is placed in Bookstore, two more things must happen: a confirmation email goes to the customer, and stock is reserved. The developer added an event to OrderService and attached both jobs to it. That way the service doesn't have to know the email or stock code:

// Naive version: the event fires before the save, synchronously
public sealed class NaiveOrderService(FakeOrderStore store)
{
    public event Action<OrderPlaced>? OrderPlaced;

    public void Place(Order order)
    {
        OrderPlaced?.Invoke(new OrderPlaced(order.Id, order.Email));
        store.Save(order);
    }
}

FakeOrderStore store = new();
NaiveOrderService service = new(store);
service.OrderPlaced += e => Console.WriteLine($"Email sent to {e.Email} for order {e.OrderId}");
service.OrderPlaced += e => throw new InvalidOperationException("Stock service timeout");

try
{
    service.Place(new Order(1001, "ada@example.com"));
}
catch (InvalidOperationException ex)
{
    Console.WriteLine($"Error: {ex.Message}");
}
Console.WriteLine($"Orders in store: {store.Count}");

The stock service times out. Does the customer get the confirmation email? Is the order written to the database? When Ada calls the next day asking "where is my order?", what does the support team find?

Notified, but Not Saved

The real output:

Email sent to ada@example.com for order 1001
Error: Stock service timeout
Orders in store: 0

Ada got the confirmation email for order 1001, but there is no such order in the database. Three separate bugs come together: the event was raised before the save; subscribers were called synchronously, one after another, and when one threw it cut off the others and the rest of Place; and a failure in the stock service took down the unrelated order save as well. The event itself was a good idea. What was wrong was when it was raised and with what guarantees.

In the C# OOP (Object-Oriented Programming) series post "A Few Design Patterns in C#: Factory, Strategy and Observer" we saw how one subscriber's exception cuts off the others and wrapped each of them in its own try/catch using GetInvocationList(). This post builds on that foundation: the IObservable<T> contract of the BCL (Base Class Library), the observers inside ASP.NET Core, and domain events at the application level.

GoF's Definition: A One-to-Many Dependency

GoF (Gang of Four: the 1994 Design Patterns book by Gamma, Helm, Johnson and Vlissides) lists Observer among the behavioral patterns. Its intent is to define a one-to-many dependency between objects: when one object's state changes, all its dependents are notified and update themselves. The roles: Subject (registers observers and notifies them), Observer (the notification interface), ConcreteSubject and ConcreteObserver. The book discusses two design choices explicitly. In the push model the subject sends the details of the change with the notification; in the pull model it only says "I changed" and the observer asks for what it needs. The cost the book warns about belongs to the same family as our opening bug: because the subject doesn't know what its observers do, an innocent-looking change can start an unexpected cascade of updates.

In C#, event is this pattern built into the language. Raising events, how subscriptions are held and subscription leaks are covered in the C# Advanced series post "Events in C#: EventHandler, Safe Raising and Subscription Leaks". Here we look at where an event is not enough.

The IObservable<T> and IObserver<T> Contract

An event says one thing: "it happened". When you model a stream of values, two more questions arise: has the stream ended, or was it cut off by an error? The IObservable<T> and IObserver<T> interfaces in the BCL put these three notifications into a contract: OnNext (a new value), OnError (the stream ended with an error) and OnCompleted (the stream ended normally). The contract's grammar is clear in the Rx (Reactive Extensions) design guidelines: zero or more OnNext calls, followed by at most one OnError or OnCompleted; nothing after that. Subscribe returns an IDisposable, and the way to end the subscription is to dispose it. The "Observer design pattern" page on Microsoft Learn explains it with an example implementing exactly this contract. A feed that publishes book price changes:

public sealed record PriceChanged(string Isbn, decimal OldPrice, decimal NewPrice);

// Subject: implements the IObservable<T> contract by hand
public sealed class PriceFeed : IObservable<PriceChanged>
{
    private readonly List<IObserver<PriceChanged>> _observers = [];
    private bool _completed;

    public IDisposable Subscribe(IObserver<PriceChanged> observer)
    {
        if (_completed)
        {
            observer.OnCompleted();
            return EmptyDisposable.Instance;
        }
        _observers.Add(observer);
        return new Unsubscriber(_observers, observer);
    }

    public void Publish(PriceChanged change)
    {
        // Contract: no OnNext after OnCompleted
        if (_completed)
            return;
        foreach (IObserver<PriceChanged> observer in _observers.ToArray())
            observer.OnNext(change);
    }

    public void Complete()
    {
        _completed = true;
        foreach (IObserver<PriceChanged> observer in _observers.ToArray())
            observer.OnCompleted();
        _observers.Clear();
    }

    private sealed class Unsubscriber(List<IObserver<PriceChanged>> observers, IObserver<PriceChanged> observer) : IDisposable
    {
        public void Dispose() => observers.Remove(observer);
    }

    private sealed class EmptyDisposable : IDisposable
    {
        public static readonly EmptyDisposable Instance = new();
        public void Dispose() { }
    }
}

public sealed class PriceDropAlert(string name) : IObserver<PriceChanged>
{
    public void OnNext(PriceChanged value)
    {
        if (value.NewPrice < value.OldPrice)
            Console.WriteLine($"{name}: {value.Isbn} dropped to {value.NewPrice}");
    }

    public void OnError(Exception error) => Console.WriteLine($"{name}: feed failed: {error.Message}");
    public void OnCompleted() => Console.WriteLine($"{name}: feed completed");
}

PriceFeed feed = new();
IDisposable alert = feed.Subscribe(new PriceDropAlert("Ada"));
IDisposable audit = feed.Subscribe(new PriceDropAlert("Audit"));

feed.Publish(new PriceChanged("978-0441013593", 18m, 15m));
alert.Dispose();
feed.Publish(new PriceChanged("978-0201633610", 55m, 49m));
feed.Complete();
feed.Publish(new PriceChanged("978-0132350884", 42m, 30m));
audit.Dispose();

Ada: 978-0441013593 dropped to 15
Audit: 978-0441013593 dropped to 15
Audit: 978-0201633610 dropped to 49
Audit: feed completed

When Ada disposed her subscription she didn't receive the second price drop, and OnCompleted didn't reach her either. The Publish after Complete() was swallowed, as the contract requires. We use _observers.ToArray() in the loops because of the "Collection was modified" error we saw in this series' Iterator post: if an observer unsubscribes inside OnNext, the list changes while it is being walked. Every hand-written subject is full of details like this; that is why in practice you use a library's ready-made subject instead of implementing these interfaces yourself.

Rx.NET: Adding Operators to the Same Interface

The BCL gives you the interfaces but no operators: to filter, transform or group you either write every observer by hand or use Rx.NET (the System.Reactive package, 7.0.0 here). Rx does for IObservable<T> what LINQ (Language Integrated Query) does for IEnumerable<T>. And it works on our hand-written PriceFeed too, because all it expects is the standard interface:

PriceFeed feed = new();
using IDisposable subscription = feed
    .Where(p => p.NewPrice < p.OldPrice)
    .Select(p => $"{p.Isbn} -{Math.Round((p.OldPrice - p.NewPrice) * 100 / p.OldPrice)}%")
    .Buffer(2)
    .Subscribe(
        batch => Console.WriteLine($"Digest: {string.Join(", ", batch)}"),
        () => Console.WriteLine("Digest: feed closed"));

feed.Publish(new PriceChanged("978-0441013593", 18m, 15m));
feed.Publish(new PriceChanged("978-0547928227", 15m, 16m));
feed.Publish(new PriceChanged("978-0201633610", 55m, 44m));
feed.Publish(new PriceChanged("978-0132350884", 40m, 30m));
feed.Complete();

Digest: 978-0441013593 -17%, 978-0201633610 -20%
Digest: 978-0132350884 -25%
Digest: feed closed

The book whose price went up was filtered out by Where, and Buffer(2) packed the discounts in pairs. The interesting part is the last lines: when the stream completed, Buffer didn't lose the single discount it was holding; it sent the half-full batch and then passed on OnCompleted. The "finished" notification in the contract is what lets operators behave correctly; you can't do this with an event. If you work on the Angular side this code should look familiar: RxJS is the JavaScript version of the same model, and a pipe(filter(...), map(...), bufferCount(2)) chain is exactly the same idea. Know one Rx rule too: OnNext calls to the same observer never overlap, they arrive one after another; if you write your own subject, providing that guarantee is your job.

Observer in .NET: IChangeToken and IOptionsMonitor.OnChange

ASP.NET Core's configuration infrastructure uses Observer's pull model. The central type is IChangeToken: it answers "has something changed?" with HasChanged and registers observers with RegisterChangeCallback. The surprising detail: a change token is single-use.

CancellationTokenSource cts = new();
IChangeToken token = new CancellationChangeToken(cts.Token);
token.RegisterChangeCallback(_ => Console.WriteLine("callback: token changed"), null);
cts.Cancel();
Console.WriteLine($"HasChanged: {token.HasChanged}");
cts.Dispose();

IConfigurationRoot config = new ConfigurationBuilder()
    .AddInMemoryCollection(new Dictionary<string, string?> { ["Shipping:FreeThreshold"] = "100" })
    .Build();

// Tokens are single-use: ChangeToken.OnChange gets a new one after every firing
using IDisposable reloads = ChangeToken.OnChange(
    () => config.GetReloadToken(),
    () => Console.WriteLine($"reload: FreeThreshold = {config["Shipping:FreeThreshold"]}"));

ServiceCollection services = new();
services.Configure<ShippingOptions>(config.GetSection("Shipping"));
using ServiceProvider provider = services.BuildServiceProvider();
IOptionsMonitor<ShippingOptions> monitor = provider.GetRequiredService<IOptionsMonitor<ShippingOptions>>();
using IDisposable? onChange = monitor.OnChange(o => Console.WriteLine($"options: FreeThreshold = {o.FreeThreshold}"));

Console.WriteLine($"current: {monitor.CurrentValue.FreeThreshold}");
config["Shipping:FreeThreshold"] = "150";
config.Reload();
config["Shipping:FreeThreshold"] = "200";
config.Reload();
Console.WriteLine($"current: {monitor.CurrentValue.FreeThreshold}");

callback: token changed
HasChanged: True
current: 100
options: FreeThreshold = 150
reload: FreeThreshold = 150
reload: FreeThreshold = 200
options: FreeThreshold = 200
current: 200

CancellationChangeToken changed once and stayed changed; it can't send a second notification. On every reload, configuration fires the old token and puts a new one in its place, so registering a callback on a single token only catches the first change. That is why ChangeToken.OnChange exists: it takes a function that produces a token (the producer) and an observer (the consumer), and after every firing gets a new token from the producer and subscribes again. IOptionsMonitor<T> uses the same mechanism internally and pushes the re-bound options object to the observer you register with OnChange. In a real application you don't call Reload() yourself; when a JSON (JavaScript Object Notation) file added with reloadOnChange: true changes, the file watcher calls it. Notice one more detail in the output: the two observers attached to the same reload token ran in one order on the first reload and in the opposite order on the second. Never rely on the order of observers relative to each other. The Options pattern itself is covered in the ASP.NET Core Web API (Application Programming Interface) series post "Configuration and the Options Pattern in ASP.NET Core: Provider Order, User Secrets and Validation".

The third member of this family comes from the desktop and MVVM (Model-View-ViewModel) world: INotifyPropertyChanged. When a view model property changes, it raises the PropertyChanged event with the property's name, and the UI (user interface) binding reads that property again. This is the pull model too: the notification carries only the name, and the observer reads the value.

Domain Events: After the Save, Independent of Each Other

Let's fix the opening bug at the application level. The idea: when something happens, the entity doesn't publish it directly; it records it. The service first makes the data durable, then dispatches the recorded events. The dispatcher runs each observer in isolation from the others.

public interface IDomainEvent;

public abstract class Entity
{
    private readonly List<IDomainEvent> _domainEvents = [];
    public IReadOnlyList<IDomainEvent> DomainEvents => _domainEvents;

    protected void Raise(IDomainEvent domainEvent) => _domainEvents.Add(domainEvent);
    public void ClearDomainEvents() => _domainEvents.Clear();
}

public sealed class Order(int id, string email) : Entity
{
    public int Id { get; } = id;
    public string Email { get; } = email;

    // The entity records what happened, it doesn't know who is listening
    public void Place() => Raise(new OrderPlaced(Id, Email));
}

public interface IDomainEventHandler<in TEvent> where TEvent : IDomainEvent
{
    Task Handle(TEvent domainEvent, CancellationToken ct);
}

public sealed class DomainEventDispatcher(IServiceProvider services)
{
    public async Task DispatchAsync(IEnumerable<IDomainEvent> events, CancellationToken ct = default)
    {
        foreach (IDomainEvent domainEvent in events)
        {
            Type handlerType = typeof(IDomainEventHandler<>).MakeGenericType(domainEvent.GetType());
            foreach (object? handler in services.GetServices(handlerType))
            {
                try
                {
                    await (Task)handlerType.GetMethod("Handle")!.Invoke(handler, [domainEvent, ct])!;
                }
                catch (Exception ex)
                {
                    // One observer's failure doesn't stop the others; log it and carry on
                    Console.WriteLine($"Handler {handler!.GetType().Name} failed: {ex.Message}");
                }
            }
        }
    }
}

public sealed class OrderService(FakeOrderStore store, DomainEventDispatcher dispatcher)
{
    public async Task PlaceAsync(Order order, CancellationToken ct = default)
    {
        order.Place();
        await store.SaveAsync(order);
        // Events are dispatched only after the save succeeds
        await dispatcher.DispatchAsync(order.DomainEvents, ct);
        order.ClearDomainEvents();
    }
}

The handlers are three classes registered in the DI (dependency injection) container: SendConfirmationEmail, ReserveStock, which still times out, and the newly added UpdateBestsellers. OrderService knows none of them:

ServiceCollection services = new();
services.AddSingleton<FakeOrderStore>();
services.AddSingleton<DomainEventDispatcher>();
services.AddSingleton<OrderService>();
services.AddSingleton<IDomainEventHandler<OrderPlaced>, SendConfirmationEmail>();
services.AddSingleton<IDomainEventHandler<OrderPlaced>, ReserveStock>();
services.AddSingleton<IDomainEventHandler<OrderPlaced>, UpdateBestsellers>();
using ServiceProvider provider = services.BuildServiceProvider();

OrderService service = provider.GetRequiredService<OrderService>();
await service.PlaceAsync(new Order(1001, "ada@example.com"));
Console.WriteLine($"Orders in store: {provider.GetRequiredService<FakeOrderStore>().Count}");

Order 1001 saved
Email sent to ada@example.com for order 1001
Handler ReserveStock failed: Stock service timeout
Bestseller list updated for order 1001
Orders in store: 1

The order was saved first, the email went out after that, and the stock failure didn't block the bestseller list. Adding a new observer is a single registration line. But let's be honest: this solution has a gap too. If the process crashes right after the order is saved, the email never goes out, because the events live only in memory. Writing the record and the "message to send" in the same transaction and dispatching them from a separate process is the job of the Outbox pattern; that is a subject for the upcoming Software Architecture series. Where domain events sit in DDD (Domain-Driven Design), the aggregate boundary and when events are collected are covered in the upcoming DDD series.

Common Pitfalls

Slowness in synchronous observers. event and our dispatcher await observers one after another; if the email service stalls for two seconds, the order endpoint waits two seconds too. Inside the observer, drop slow work onto a queue and let a background job process it.

Exceptions flowing back into the subject. As in the opening, in an observer list without isolation the first exception cuts off both the other observers and the rest of the subject's work. Either protect each observer separately or choose this behavior deliberately (if you say "no stock reservation, no order", stock is not an observer, it is part of the operation).

Subscription leaks. A long-lived subject (a singleton feed, a static event) holds short-lived observers by reference, and the garbage collector can't collect them. Keep the IDisposable returned by Subscribe and dispose it, including the one returned by IOptionsMonitor.OnChange.

Publishing the event before the save. An "order placed" event should be published when the order really has been placed, that is, after the save succeeds. Otherwise you announce to the world something that never happened.

Observer chains and order. An observer's reaction can raise another event, which triggers another observer. That is GoF's "unexpected updates" warning: as the chain grows, you can't find by reading the code how far a change spreads. And as we saw, you can't rely on the order between observers either.

When Not to Use It

Don't build an Observer for a single listener and a relationship that won't change; a direct method call is more readable and easier to debug. Don't turn steps that are part of the operation, whose success decides the operation's success, into observers: if stock reservation is a condition of the order, it is not an event but an explicit call inside the service. Use IObservable<T> and Rx when you really have a stream of values spread over time (a price feed, sensor data, user input); for a one-off notification an event or a Task is enough. If you are consuming an async, ordered sequence, IAsyncEnumerable<T>, which pulls instead of pushes, is often the simpler model.

A Small Experiment

★☆☆ Add a Fail(Exception) method to PriceFeed: it should send OnError to every observer and ignore any Publish and Complete calls that follow. Does Buffer in the Rx chain send its half-full batch when the error arrives?

★★☆ Write a console app that adds appsettings.json with reloadOnChange: true, attach an observer to IOptionsMonitor.OnChange, and save the file in an editor. How many times did the callback run for a single save? How would you prevent the duplicates?

★★★ Change DomainEventDispatcher so that each handler runs in its own DI scope (IServiceScopeFactory). Then rework OrderService.PlaceAsync so that it writes events to a Channel<IDomainEvent> that a BackgroundService reads. Which guarantee did you gain, and which did you lose?

Observer was about an object telling others when its state changes. In the next post the state itself becomes behavior: the State pattern, which moves the switch (Status) blocks repeated in every method of an order into state objects.