Observer Pattern Derinlemesine: event, IObservable<T>, IChangeToken ve Domain Event'ler

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile derlenip denenmiştir.

Önce Tahmin Et

Bookstore'da sipariş verildiğinde iki iş daha yapılmalı: müşteriye onay e-postası gitmeli ve stok rezerve edilmeli. Geliştirici OrderService'e bir event ekleyip iki işi ona bağlamış. Böylece service, e-posta ve stok kodunu tanımak zorunda kalmıyor:

// Naif hali: event, kayıttan önce ve senkron tetikleniyor
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}");

Stok servisi zaman aşımına uğruyor. Müşteri onay e-postası alır mı? Sipariş veritabanına yazılır mı? Ada ertesi gün "siparişim nerede" diye aradığında destek ekibi neyi bulur?

Haber Verildi, Kayıt Yok

Gerçek çıktı:

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

Ada 1001 numaralı siparişin onay e-postasını aldı, ama veritabanında böyle bir sipariş yok. Üç ayrı hata bir arada: event kayıt yapılmadan önce tetiklendi; subscriber'lar senkron ve sırayla çağrıldı, biri exception fırlatınca diğerlerinin ve Place'in geri kalanının önü kesildi; stok servisindeki bir arıza, onunla ilgisi olmayan sipariş kaydını da düşürdü. Event'in kendisi doğru bir fikirdi. Yanlış olan, ne zaman ve hangi garantilerle tetiklendiğiydi.

C# OOP (Object-Oriented Programming) serisindeki "C#'ta Birkaç Design Pattern: Factory, Strategy ve Observer" yazısında bir subscriber'ın exception'ının diğerlerini nasıl kestiğini görmüş ve GetInvocationList() ile her birini ayrı try/catch'e almıştık. Bu yazı o temelin üstüne kuruluyor: BCL'nin (Base Class Library) IObservable<T> sözleşmesi, ASP.NET Core'un içindeki observer'lar ve uygulama seviyesinde domain event'ler.

GoF'un Tanımı: Bire Çok Bağımlılık

GoF (Gang of Four: Gamma, Helm, Johnson ve Vlissides'in 1994 tarihli Design Patterns kitabı) Observer'ı behavioral pattern'ler arasında sayar. Niyeti, object'ler arasında bire çok bir bağımlılık tanımlamak: bir object'in durumu değiştiğinde ona bağlı olanların hepsi haberdar edilir ve kendini günceller. Roller: Subject (observer'ları kaydeden ve haber veren), Observer (haber alma interface'i), ConcreteSubject ve ConcreteObserver. Kitap iki tasarım seçimini açıkça tartışır. Push modelinde subject değişikliğin ayrıntısını bildirimle birlikte gönderir; pull modelinde yalnız "değiştim" der, observer ihtiyacı olanı kendisi sorar. Kitabın uyardığı bedel de bizim açılıştaki hatamızla aynı aileden: subject observer'larının ne yaptığını bilmediği için, masum görünen bir değişiklik beklenmedik bir güncelleme zinciri başlatabilir.

C#'ta event bu pattern'in dile gömülü hâli. Event'in tetiklenmesi, subscription'ın nasıl tutulduğu ve subscription leak C# İleri serisindeki "C#'ta Event: EventHandler, Güvenli Tetikleme ve Subscription Leak" yazısında. Burada event'in yetmediği yerlere bakıyoruz.

IObservable<T> ve IObserver<T> Sözleşmesi

event tek bir şey söyler: "oldu". Bir değer akışını modellediğinde iki soru daha doğar: akış bitti mi, yoksa hatayla mı kesildi? BCL'deki IObservable<T> ve IObserver<T> interface'leri bu üç bildirimi sözleşmeye koyar: OnNext (yeni değer), OnError (akış hatayla bitti) ve OnCompleted (akış normal bitti). Sözleşmenin grameri Rx (Reactive Extensions) tasarım kurallarında net: sıfır ya da daha fazla OnNext, ardından en fazla bir kez OnError ya da OnCompleted; ondan sonra hiçbir şey. Subscribe bir IDisposable döndürür ve subscription'ı bitirmenin yolu onu dispose etmektir. Microsoft Learn'deki "Observer design pattern" sayfası tam olarak bu sözleşmeyi uygulayan bir örnekle anlatıyor. Kitap fiyatı değişikliklerini yayınlayan bir feed:

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

// Subject: IObservable<T> sözleşmesini elle uyguluyor
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)
    {
        // Sözleşme: OnCompleted'dan sonra OnNext yok
        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

Ada subscription'ını dispose edince ikinci fiyat düşüşünü almadı ve OnCompleted da ona gitmedi. Complete()'ten sonraki Publish sözleşme gereği yutuldu. Döngülerde _observers.ToArray() kullanmamızın nedeni serinin Iterator yazısında gördüğümüz "Collection was modified" hatası: bir observer OnNext içinde subscription'ını bırakırsa liste dolaşılırken değişir. Elle yazılan her subject bu tür ayrıntılarla dolu; o yüzden pratikte bu interface'leri kendin implement etmek yerine bir library'nin hazır subject'ini kullanırsın.

Rx.NET: Aynı Interface'e Operatör Eklemek

BCL interface'leri verir ama operatör vermez: filtrelemek, dönüştürmek, gruplamak için ya her observer'ı elle yazarsın ya da Rx.NET'i (System.Reactive paketi, burada 7.0.0) kullanırsın. Rx, IObservable<T> için LINQ'in (Language Integrated Query) IEnumerable<T> için yaptığını yapar. Üstelik bizim elle yazdığımız PriceFeed üzerinde de çalışır, çünkü tek beklentisi standart 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

Fiyatı artan kitap Where'de elendi, Buffer(2) indirimleri ikişerli paketledi. İlginç olan son satırlar: akış tamamlandığında Buffer elinde kalan tek indirimi kaybetmedi, yarım paketi gönderip sonra OnCompleted'ı iletti. Sözleşmedeki "bitti" bildirimi operatörlerin doğru davranabilmesi için gerekli; bir event ile bunu yapamazsın. Angular tarafında çalışıyorsan bu kod sana tanıdık gelmeli: RxJS aynı modelin JavaScript sürümü ve pipe(filter(...), map(...), bufferCount(2)) zinciri birebir aynı fikir. Rx'in bir kuralını da bil: aynı observer'a yapılan OnNext çağrıları üst üste binmez, sırayla gelir; kendi subject'ini yazıyorsan bu garantiyi de sen vermek zorundasın.

.NET'te Observer: IChangeToken ve IOptionsMonitor.OnChange

ASP.NET Core'un configuration altyapısı Observer'ın pull modelini kullanır. Merkezdeki tip IChangeToken: "bir şey değişti mi?" sorusuna HasChanged ile cevap verir ve RegisterChangeCallback ile observer kaydeder. Ayrıntısı şaşırtıcı: bir change token tek kullanımlık.

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();

// Token tek kullanımlık: ChangeToken.OnChange her tetiklenmeden sonra yenisini alıyor
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 bir kez değişti ve öyle kaldı; ikinci bir bildirim gönderemez. Configuration her reload'da eski token'ı tetikleyip yerine yenisini koyuyor, o yüzden tek bir token'a callback kaydetmek yalnız ilk değişikliği yakalar. ChangeToken.OnChange bu yüzden var: bir token üreten fonksiyon (producer) ve bir observer (consumer) alır, her tetiklenmeden sonra producer'dan yeni token alıp yeniden subscribe olur. IOptionsMonitor<T> da içeride aynı mekanizmayı kullanıyor ve OnChange'e kaydettiğin observer'a yeniden bind edilmiş options object'ini push ediyor. Gerçek bir uygulamada Reload()'u sen çağırmazsın; reloadOnChange: true ile eklenen bir JSON (JavaScript Object Notation) dosyası değiştiğinde file watcher çağırır. Çıktıda bir ayrıntıya daha dikkat et: aynı reload token'ına bağlı iki observer ilk reload'da bir sırayla, ikincisinde ters sırayla çalıştı. Observer'ların birbirine göre sırasına hiçbir zaman güvenme. Options pattern'in kendisi ASP.NET Core Web API (Application Programming Interface) serisindeki "ASP.NET Core'da Configuration ve Options Pattern: Provider Sırası, User Secrets ve Validation" yazısında.

Bu ailenin üçüncü üyesi masaüstü ve MVVM (Model-View-ViewModel) dünyasından: INotifyPropertyChanged. Bir view model property'si değiştiğinde PropertyChanged event'ini property'nin adıyla tetikler, UI (user interface) binding'i de o property'yi yeniden okur. Bu da pull modeli: bildirim yalnız adı taşır, değer observer tarafından okunur.

Domain Event'ler: Kayıttan Sonra, Birbirinden Bağımsız

Açılıştaki hatayı uygulama seviyesinde düzeltelim. Fikir şu: entity bir şey olduğunda onu doğrudan yayınlamaz, kaydeder. Service önce veriyi kalıcı hâle getirir, sonra kaydedilen event'leri dağıtır. Dispatcher her observer'ı birbirinden izole çalıştırır.

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;

    // Entity event'i kaydeder, kimin dinlediğini bilmez
    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)
                {
                    // Bir observer'ın hatası diğerlerini durdurmuyor; log'la ve devam et
                    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);
        // Event'ler kayıt başarılı olduktan sonra dağıtılıyor
        await dispatcher.DispatchAsync(order.DomainEvents, ct);
        order.ClearDomainEvents();
    }
}

Handler'lar DI (dependency injection) container'a kayıtlı üç class: SendConfirmationEmail, hâlâ zaman aşımına uğrayan ReserveStock ve yeni eklenen UpdateBestsellers. OrderService hiçbirini tanımıyor:

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

Sipariş önce kaydedildi, e-posta ondan sonra gitti, stok hatası çok satanlar listesini engellemedi. Yeni bir observer eklemek tek bir registration satırı. Ama dürüst olalım, bu çözümün de bir boşluğu var: sipariş kaydedildikten hemen sonra süreç çökerse e-posta hiç gitmez, çünkü event'ler yalnız bellekte. Kaydı ve "gönderilecek mesajı" aynı transaction'da yazıp ayrı bir süreçle dağıtmak Outbox pattern'in işi; bu ilerideki Yazılım Mimarisi serilerinin konusu. Domain event'lerin DDD (Domain-Driven Design) içindeki yeri, aggregate sınırı ve event'in ne zaman toplandığı da ilerideki DDD serisinde.

Sık Düşülen Tuzaklar

Senkron observer'da yavaşlık. event ve bizim dispatcher'ımız observer'ları sırayla await ediyor; e-posta servisi iki saniye bekletirse sipariş endpoint'i de iki saniye bekler. Yavaş işi observer'ın içinde bir queue'ya bırak, işlemeyi background job'a ver.

Exception'ın subject'e geri dönmesi. Açılıştaki gibi, izolasyonsuz bir observer listesinde ilk exception hem diğer observer'ları hem de subject'in geri kalan işini keser. Ya her observer'ı ayrı ayrı koru, ya da bu davranışı bilerek seç ("stok rezerve edilemezse sipariş de olmasın" diyorsan stok bir observer değil, işlemin parçasıdır).

Subscription sızıntısı. Uzun yaşayan bir subject (singleton bir feed, static bir event) kısa yaşayan observer'ları referansla tutar ve garbage collector onları toplayamaz. Subscribe'ın döndürdüğü IDisposable'ı sakla ve dispose et; IOptionsMonitor.OnChange'in döndürdüğü de dahil.

Event'i kayıttan önce yayınlamak. "Sipariş verildi" event'i, sipariş gerçekten verildiğinde, yani kayıt başarılı olduktan sonra yayınlanmalı. Aksi hâlde dünyaya hiç olmamış bir şeyi duyurursun.

Observer zincirleri ve sıra. Bir observer'ın tepkisi başka bir event'i, o da başka bir observer'ı tetikleyebilir. GoF'un "beklenmedik güncellemeler" uyarısı bu: zincir uzadıkça bir değişikliğin nereye kadar yayıldığını kodu okuyarak bulamazsın. Gördüğümüz gibi observer'lar arasındaki sıraya da güvenemezsin.

Ne Zaman Kullanma

Tek bir listener ve değişmeyecek bir ilişki için Observer kurma; doğrudan bir method çağrısı daha okunur ve debug etmesi kolaydır. İşlemin parçası olan, başarısı işlemin başarısını belirleyen adımları observer'a çevirme: stok rezervasyonu siparişin bir koşuluysa event değil, service'in içindeki açık bir çağrıdır. IObservable<T> ve Rx'i, gerçekten zamana yayılmış bir değer akışın olduğunda (fiyat feed'i, sensör verisi, kullanıcı girdisi) kullan; tek seferlik bir bildirim için event ya da bir Task yeterli. Async ve sıralı bir sequence tüketiyorsan push yerine pull olan IAsyncEnumerable<T> çoğu zaman daha basit bir modeldir.

Küçük Bir Deneme

★☆☆ PriceFeed'e Fail(Exception) method'u ekle: bütün observer'lara OnError göndersin ve sonrasında gelen Publish ve Complete çağrılarını yok saysın. Rx zincirindeki Buffer hata gelince elindeki yarım paketi gönderiyor mu?

★★☆ appsettings.json'ı reloadOnChange: true ile ekleyen bir console uygulaması yaz, IOptionsMonitor.OnChange'e bir observer bağla ve dosyayı bir editörde kaydet. Tek bir kaydetmede callback kaç kez çalıştı? Gereksiz tekrarları nasıl engellersin?

★★★ DomainEventDispatcher'ı, her handler'ı kendi DI scope'unda çalıştıracak şekilde değiştir (IServiceScopeFactory). Sonra OrderService.PlaceAsync'i, event'leri bir Channel<IDomainEvent>'a yazıp bir BackgroundService'in okuyacağı şekilde yeniden düzenle. Hangi garantiyi kazandın, hangisini kaybettin?

Observer bir object'in durumu değişince başkalarına haber vermesiydi. Bir sonraki yazıda durumun kendisi davranışa dönüşüyor: siparişin her method'unda tekrarlanan switch (Status) bloklarını state object'lerine taşıyan State pattern.