Observer Pattern

Observer solves the "notification problem" — when something changes, how do you tell everyone who cares without hardcoding who those "someones" are? Think of it like a YouTube channel. When a creator posts a new video, they don't personally call every viewer. Instead, viewers subscribe to the channel, and YouTube automatically notifies everyone on the list. The creator doesn't know (or care) who's subscribed — they just publish, and the platform handles delivery.

Without Observer, the code that changes data would have to manually call screen.Update(), logger.Log(), emailer.Send() — and every time you add a new consumer, you'd edit that code again. That's like the creator maintaining a personal phone list and calling each fan individually. Observer flips this: interested parties subscribe themselves, and the broadcaster just says "something changed" to everyone on the list. It only knows the interface, never the specific subscriber types. Adding a new consumer is just adding a new subscriber — zero changes to the publisher.

Imagine a restaurant kitchen. In the push model, the chef puts the finished plate on the counter and calls out "Table 4, your steak is ready!" — the food comes to you with all the details. In the pull model, the chef just rings a bell and says "something's ready" — and the waiter has to walk over, look at the counter, and figure out what it is and who it's for.

Push: Subject sends data directly — observer.Update(newPrice). Simpler for observers but they receive everything whether they want it or not. This is the default in most modern implementations because it eliminates timing bugs (the observer always gets the exact data that triggered the notification).

Pull: Subject sends a bare notification (observer.Update(this)), then the observer calls back to fetch only what it needs: var price = subject.GetPrice(). More flexible when the payload is large, but creates a back-reference and risks stale reads. In .NET: C# events are push (EventArgs carry the data). INotifyPropertyChanged is pull-ish — it sends the property name, and the observer calls the getter to read the value.

C# events ARE Observer — language-level syntactic sugar. The event keyword wraps a MulticastDelegateThe backing type for C# events. A linked list of method pointers. += adds to the chain, -= removes. The event keyword restricts external access so only the declaring class can invoke it — preventing callers from firing the event themselves or overwriting the entire delegate. (the observer list) and restricts external callers to only += and -=. += = Subscribe, -= = Unsubscribe, ?.Invoke() = Notify. The compiler enforces that outside code cannot overwrite or invoke the event directly.

INotifyPropertyChanged is a built-in .NET interface used in MVVM. It exposes one event: PropertyChangedEventHandler? PropertyChanged. When a ViewModel property changes, it raises this event with the property name string. WPF/MAUI data binding subscribes to PropertyChanged and updates the bound UI element — that binding IS the Observer. ViewModel = Subject, binding engine = Observer, UI element = display updated by the observer.

By default, an exception from one observer propagates up and stops all subsequent observers — they never run. This is almost always wrong. Fix: wrap each observer call individually:

public void Notify(T state)
{
    foreach (var observer in _observers.ToList())
    {
        try { observer.Update(state); }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Observer {Type} threw",
                observer.GetType().Name);
            // Continue — one bad observer won't stop the others
        }
    }
}

Think of it this way: Observer is like shouting across a room — you can see everyone you're talking to, and they hear you instantly. Pub/Sub is like posting a letter through the post office — you drop it off, the postal system figures out who gets it, and the recipients might be in a completely different city.

Observer: The Subject holds direct references to observers. They live in the same process, same memory space. The subject calls observer methods directly — like a function call. Tightly coupled physically (same process), but loosely coupled in terms of types (only knows the interface).

Pub/SubPublish/Subscribe uses a message broker (Azure Service Bus, Kafka, RabbitMQ) as an intermediary. Publishers and subscribers don't know each other at all — they both know only the broker and the message contract. Works across processes, services, and data centers. Provides durability (messages persist if subscriber is down) and fan-out (one publish → many subscribers across services).: Publishers and subscribers communicate through a message broker (like Azure Service Bus or Kafka). No direct references. The publisher doesn't know who's listening. The broker handles delivery, retries, and persistence. Works across processes, services, machines, and even data centers. Observer = in-process, fast, simple. Pub/Sub = distributed, durable, scalable.

Yes — this is a chain of observers / observer pipeline. A filtering subject subscribes to a raw data source, processes events, and re-publishes to its own subscribers. Rx.NET operators work this way — each Where/Select/Throttle operator is both an IObserver<T> (subscribing upstream) and an IObservable<T> (publishing downstream). The risk: circular chains (A observes B observes A). Always design observer graphs as DAGs — no cycles.

• Only one consumer: Just inject a callback delegate — events add unnecessary ceremony.
• Request-response flow: If you need a return value from the observer, use direct method calls.
• Sequential dependency between observers: If Observer B needs Observer A's output, use a pipeline not parallel notification.
• Publisher must confirm consumer success: Events are fire-and-forget. If confirmation is required, use awaitable method calls with result types.
• Debugging priority: Observer makes call stacks harder to trace. In systems where debuggability > flexibility, direct calls win.

The core problem is simple: if one thread is looping through the observer list to send notifications, and another thread adds or removes an observer from the same list at the same time, things break. You might get a crash, a skipped observer, or a notification sent to an observer that just unsubscribed. The solution is the "copy-on-write" pattern — instead of modifying the existing list, every subscription change creates a brand new list and swaps the reference atomically.

public class ThreadSafeSubject<T>
{
    private readonly object _lock = new();
    private List<IObserver<T>> _observers = new();

    public void Subscribe(IObserver<T> observer)
    {
        lock (_lock) // copy-on-write: atomic list replacement
            _observers = [.._observers, observer];
    }

    public void Unsubscribe(IObserver<T> observer)
    {
        lock (_lock)
        {
            var copy = new List<IObserver<T>>(_observers);
            copy.Remove(observer);
            _observers = copy;
        }
    }

    public void Notify(T value)
    {
        List<IObserver<T>> snapshot;
        lock (_lock) { snapshot = _observers; } // atomic read — no copy needed
        // Iterate outside lock — observers can safely subscribe/unsubscribe during this
        foreach (var o in snapshot) o.OnNext(value);
    }
}

The copy-on-writeNever modify the existing collection in place. Create a new copy with the change, then atomically swap the reference under a lock. Readers (Notify) always see a consistent snapshot. Writers (Subscribe/Unsubscribe) briefly lock only to swap the reference. Maximizes read concurrency — notifications never hold the lock. pattern ensures Notify() iterates a stable snapshot without holding the lock, preventing deadlocks if observers subscribe during notification.

The gold standard is IDisposable subscription tokens — Subscribe() returns an IDisposable, and the subscriber calls Dispose() in its own cleanup:

public IDisposable Subscribe(Action<T> handler)
{
    _handlers.Add(handler);
    return new Subscription(() => _handlers.Remove(handler));
}

private sealed class Subscription(Action cleanup) : IDisposable
{
    private int _disposed;
    public void Dispose()
    {
        if (Interlocked.Exchange(ref _disposed, 1) == 0)
            cleanup();
    }
}

// Subscriber owns its subscriptions — disposes all at once
public class Dashboard : IDisposable
{
    private readonly CompositeDisposable _subs = new();

    public Dashboard(IStockTicker ticker, INewsFeed news)
    {
        _subs.Add(ticker.Subscribe(OnPriceChanged));
        _subs.Add(news.Subscribe(OnNewsArrived));
    }

    public void Dispose() => _subs.Dispose();
}

• Composability: IObservable chains with LINQ-like operators (Where, Select, Buffer, Throttle). Events cannot be composed without Rx.
• Error propagation: OnError(Exception) carries exceptions through the pipeline. Events either crash or swallow.
• Completion: OnCompleted() signals "no more values." Events have no end signal.
• Lifetime management: Subscribe() returns IDisposable — built-in cleanup. Events need manual -=.
• First-class value: IObservable<T> can be stored in fields, passed as parameters, returned from methods. Events cannot.

Reach for Rx when you need: operator composition (throttle rapid input, buffer batches, merge streams), time-based operations (Debounce, Timeout, Sample), cross-thread scheduling (ObserveOn(scheduler)), or complex coordination (Zip two streams, CombineLatest). Plain events win for: simple one-to-many notification with no composition, existing WPF infrastructure, or when the Rx NuGet dependency isn't justified. Rule of thumb: if you've written more than one filter or one TimeSpan-based condition on events, reach for Rx.

Three approaches: filter in the subject (store predicates), filter in the observer (OnNext checks and discards), or topic-based routing (separate list per event type):

public class TypedEventBus
{
    private readonly Dictionary<Type, List<Delegate>> _handlers = new();

    public IDisposable Subscribe<TEvent>(Action<TEvent> handler)
    {
        var key = typeof(TEvent);
        if (!_handlers.TryGetValue(key, out var list))
            _handlers[key] = list = new();
        list.Add(handler);
        return new Subscription(() => list.Remove(handler));
    }

    public void Publish<TEvent>(TEvent evt)
    {
        if (_handlers.TryGetValue(typeof(TEvent), out var list))
            foreach (var h in list.ToList())
                ((Action<TEvent>)h)(evt); // only handlers for TEvent
    }
}

public interface IAsyncObserver<T>
{
    Task OnNextAsync(T value, CancellationToken ct = default);
}

public class AsyncSubject<T>
{
    private readonly List<IAsyncObserver<T>> _observers = new();

    // Sequential: each observer waits for previous to complete
    public async Task NotifySequentialAsync(T value, CancellationToken ct = default)
    {
        foreach (var o in _observers.ToList())
            await o.OnNextAsync(value, ct);
    }

    // Parallel: all observers run concurrently — use when independent
    public async Task NotifyParallelAsync(T value, CancellationToken ct = default)
    {
        await Task.WhenAll(_observers.ToList()
            .Select(o => o.OnNextAsync(value, ct)));
    }
}

Task.WhenAllAwaits all tasks to complete and returns when ALL finish (or throws AggregateException if any fail). Use for parallel independent observers. Use sequential await when order matters or when one observer's output affects the next. is preferred for independent observers. Sequential is needed when one observer must complete before the next runs (e.g., a validation pipeline).

Think of Observer like a radio station: one broadcaster sends a signal, and anyone with a radio picks it up. The station doesn't coordinate between listeners. Now think of Mediator like an air traffic controller: every plane talks only to the controller, and the controller coordinates who goes where. Planes never talk directly to each other.

Observer: Subject knows it has observers. One-directional notification. Best for event broadcasting, property change, "something happened — here's the data." The subject talks, observers listen.

Mediator: No object knows about others — all talk through the mediator. The mediator knows everyone and coordinates interactions. Best for complex object interactions, form coordination, or N-by-M connections where everything talks to everything. MediatR in .NET is the classic implementation.

The decision rule: If A is telling B "something changed" — use Observer. If A is asking "who handles this command?" — use Mediator. If your observer graph starts looking like a web instead of a tree, it's time to upgrade to a Mediator.

Standard Observer notifies in subscription order — but this is an implementation detail you should never rely on. If ordering matters, model it explicitly:

private readonly List<(IObserver<T> Observer, int Priority)> _observers = new();

public void Subscribe(IObserver<T> observer, int priority = 0)
{
    _observers.Add((observer, priority));
    _observers.Sort((a, b) => b.Priority.CompareTo(a.Priority)); // high first
}

public void Notify(T value)
{
    foreach (var (obs, _) in _observers.ToList())
        obs.OnNext(value);
}

Use WeakReference in your test to verify the subscriber is garbage-collected after unsubscribing:

[Fact]
public void Subscriber_CollectedAfterUnsubscribe()
{
    var subject = new StockPriceSubject("AAPL");
    WeakReference weakRef = null!;

    var sub = CreateSubscription(subject, out weakRef);
    sub.Dispose(); // unsubscribe

    GC.Collect();
    GC.WaitForPendingFinalizers();
    GC.Collect();

    Assert.False(weakRef.IsAlive, "Memory leak: subscriber still alive after Dispose()");
}

private IDisposable CreateSubscription(StockPriceSubject subject, out WeakReference weakRef)
{
    var observer = new FakeStockObserver();
    weakRef = new WeakReference(observer);
    return subject.Subscribe(observer);
    // observer goes out of scope here — only WeakReference and subject hold refs
}

Backpressure is what happens when a producer creates data faster than consumers can process it. Imagine a fire hose connected to a drinking fountain — the fountain can't handle the flow. Without a strategy, you get either crashed consumers (out of memory from queuing too many events) or lost data (events dropped on the floor). The key is choosing the right strategy for your domain.

// Option 1: BoundedChannel — blocks/drops when full
var channel = Channel.CreateBounded<StockPrice>(new BoundedChannelOptions(1000)
{
    FullMode = BoundedChannelFullMode.DropOldest
});

// Option 2: Rx Throttle — only emit after N ms quiet period
ticker.Prices
    .Throttle(TimeSpan.FromMilliseconds(100))
    .Subscribe(p => UpdateUI(p));

// Option 3: Rx Sample — emit latest every N ms regardless
ticker.Prices
    .Sample(TimeSpan.FromMilliseconds(500))
    .Subscribe(p => UpdateDashboard(p));

// Option 4: Rx Buffer — batch N events into windows
ticker.Prices
    .Buffer(TimeSpan.FromSeconds(1))
    .Subscribe(batch => ProcessBatch(batch));

Right strategy by domain: UI updates → Sample/Throttle (show latest). Audit → BoundedChannel (no data loss, backpressure to producer). Analytics → Buffer (batch efficiency).

Think of standard Observer like a live radio broadcast — if you weren't tuned in, you missed it forever. Event Sourcing is like recording every broadcast. Anyone who joins late can replay the recordings to catch up, then listen live from that point forward.

Event Sourcing stores every state change as an immutable event. Projections (read models) are observers of the event stream — they subscribe and build their own views by processing events sequentially. The event store is the Subject; projections are Observers. The key difference from basic Observer: ES events are persisted and replayable. Standard Observer notifications are ephemeral — missed events are gone. With ES, any projection can be rebuilt from scratch by replaying from event 0, making Observer's "fire-and-forget" limitation irrelevant.

When Observer needs to cross process boundaries — when the subject and observers live in different services on different machines — direct method calls won't work. You need a message broker sitting in the middle to carry notifications between services. This is Pub/Sub: the distributed version of Observer.

The subject publishes to a topic (Azure Service Bus, Kafka, RabbitMQ). Subscriber services consume from that topic — each service is an independent Observer. The broker provides durability (messages survive crashes), fan-out (one publish reaches many consumers), ordering, and at-least-once delivery. In .NET, MassTransit or NServiceBus abstract the broker: consumers implement IConsumer<TMessage> — the Observer interface mapped to a message type.

The critical addition over in-process Observer: idempotency. Network failures can cause the broker to deliver the same message twice. Your consumers must handle duplicates safely — typically by checking a "processed events" log before acting. Without idempotency, a redelivered "OrderPlaced" event could charge a customer twice.

Classic Observer is push-only — the subject pushes data at whatever rate it wants, and the consumer has to keep up or drown. Reactive Streams fixes this by adding a conversation between producer and consumer. The consumer says "I'm ready for 5 more items," and the producer only sends 5. This is called async backpressure.

In .NET, IAsyncEnumerable<T> with await foreach implements this naturally — each await MoveNextAsync() IS the subscriber requesting one item. The producer can't outrun the consumer because it must await each request. System.Threading.Channels provides the same pattern with BoundedChannel + ReadAllAsync().

CQRS splits your system into a Write side (handles commands, changes state) and a Read side (serves queries, displays data). The Observer pattern is the bridge between them — when the write side changes something, domain events notify the read side to update its projections.

The pipeline flows like this: Command arrives, the Aggregate processes it and raises DomainEvents, the Dispatcher (which is the Subject) delivers those events to every registered handler (each handler is an Observer). Projections update read models; side-effect handlers send emails, write audit logs, etc. Adding a new reaction to any event = one new handler class + one DI registration line. Zero changes to the aggregate or command handler.

When an event fires but one observer never receives it, the problem could be anywhere in the notification chain. Think of it like debugging a package that never arrived — was it never sent? Was it sent to the wrong address? Did it get intercepted along the way? Here's a systematic checklist to narrow it down:

Walk through each checkpoint in order: (1) Was the observer subscribed? — Add logging inside Subscribe(). (2) Did the event fire at all? — Log at the top of Notify(). (3) Was it unsubscribed too early? — Log inside Unsubscribe() with a stack trace to see who triggered it. (4) Did a prior observer throw and stop the chain? — Add per-observer try/catch to isolate failures. (5) Thread affinity? — Is the notification on a background thread while the observer needs the UI thread? (6) Filter dropped it? — Log at each Where() boundary in Rx pipelines.

In Rx pipelines, the .Do(x => logger.Log("Value: {V}", x)) operator is invaluable — it lets you "tap" into the stream at any point to see what values are flowing through, without changing the pipeline behavior. Place it between operators to find exactly where events get filtered or dropped.

DiagnosticListener is .NET's built-in Observer infrastructure. The framework publishes diagnostic events; your code subscribes. OpenTelemetry, Application Insights, and custom APM tools all hook in this way — zero changes to application code:

// Register as IObserver<DiagnosticListener> in DI
builder.Services.AddSingleton<IObserver<DiagnosticListener>, HttpRequestObserver>();

public class HttpRequestObserver : IObserver<DiagnosticListener>
{
    public void OnNext(DiagnosticListener listener)
    {
        if (listener.Name == "Microsoft.AspNetCore")
            listener.Subscribe(new HttpEventObserver());
    }
    public void OnError(Exception error) { }
    public void OnCompleted() { }
}

public class HttpEventObserver : IObserver<KeyValuePair<string, object?>>
{
    public void OnNext(KeyValuePair<string, object?> evt)
    {
        if (evt.Key == "Microsoft.AspNetCore.Hosting.HttpRequestIn.Start")
        {
            var httpContext = (HttpContext)evt.Value!;
            // Observe every incoming request — logging, metrics, tracing
        }
    }
    public void OnError(Exception error) { }
    public void OnCompleted() { }
}

Both — but the split matters. Unit tests verify each observer in isolation: given this input event, does the observer produce the correct side effect? Mock the subject; call OnNext() directly. Integration tests verify the wiring: does the subject actually reach all observers, in the right order, with the right data?

Key testing strategies: (1) Unit-test observers independently by calling their handler method directly. (2) Integration-test the dispatcher/subject with real DI wiring. (3) For Rx pipelines, use TestScheduler to control virtual time — never use Task.Delay in tests. (4) Use WeakReference tests to verify no memory leaks after unsubscription.

Domain events ARE Observer applied to DDD. The aggregate (Subject) raises events; domain event handlers (Observers) react. The aggregate never knows who's listening — it just records what happened.

The dispatch timing matters: pre-commit handlers (same transaction) are for consistency-critical reactions like inventory reservation. Post-commit handlers (after SaveChanges) are for side effects like email and analytics. This two-phase approach lets you guarantee correctness for critical operations while keeping non-critical work decoupled.

public abstract class Entity
{
    private readonly List<IDomainEvent> _events = new();
    public IReadOnlyList<IDomainEvent> DomainEvents => _events;
    protected void Raise(IDomainEvent evt) => _events.Add(evt);
    public void ClearEvents() => _events.Clear();
}

public class Order : Entity
{
    public void Place()
    {
        Status = OrderStatus.Placed;
        Raise(new OrderPlacedEvent(Id, Total, DateTime.UtcNow));
    }
}

// EF Core dispatches after SaveChanges
public class DomainEventDispatcher(IServiceProvider sp)
{
    public async Task DispatchAsync(IEnumerable<IDomainEvent> events, CancellationToken ct)
    {
        foreach (var evt in events)
        {
            var handlerType = typeof(IDomainEventHandler<>).MakeGenericType(evt.GetType());
            foreach (dynamic handler in sp.GetServices(handlerType))
                await handler.HandleAsync((dynamic)evt, ct);
        }
    }
}

Observer becomes a bottleneck when the math works against you. 500 observers multiplied by 10,000 events per second equals 5 million method calls per second. Each call allocates EventArgs, traverses the list, and waits for the observer to finish. At some point, the overhead overwhelms your system.

The three scenarios: Fan-out explosion (N observers x M events/sec = N*M calls), synchronous blocking (one slow observer blocks the chain), and GC pressure (EventArgs allocation at high frequency). Use the decision framework above to pick the right strategy for your throughput level.

// Fix 1: Object pooling — zero allocation at high frequency
var pool = ObjectPool.Create<PriceEvent>();
var evt = pool.Get();
evt.Init("AAPL", 185.50m);
subject.Notify(evt);
pool.Return(evt); // recycle

// Fix 2: Channel + batching — reduce observer call count
await foreach (var batch in channel.Reader.ReadAllAsync().Buffer(100))
    foreach (var observer in observers)
        observer.OnBatch(batch); // 100x fewer calls