Observer Pattern #

Sebuah pesanan baru masuk ke sistem e-commerce. Akibatnya, banyak hal harus terjadi: inventory harus dikurangi, email konfirmasi harus dikirim, tim fulfillment harus dinotifikasi, analytics harus diupdate, dan loyalty points harus ditambahkan ke akun pembeli. Tanpa Observer Pattern, OrderService harus memanggil semuanya secara eksplisit — setiap subsistem baru yang perlu tahu tentang pesanan baru berarti satu baris kode tambahan di OrderService. OrderService yang seharusnya hanya mengurusi logika pesanan, akhirnya mengurusi inventory, email, analytics, dan loyalty sekaligus — dan setiap kali ada subsistem baru, ia harus diubah. Observer Pattern membalik tanggung jawab ini: OrderService hanya mengumumkan “pesanan baru masuk”, dan siapapun yang peduli bisa mendaftarkan diri untuk mendengar pengumuman itu — tanpa OrderService perlu tahu siapa yang mendengar atau berapa banyak.

Apa itu Observer Pattern? #

Observer Pattern adalah behavioral design pattern yang mendefinisikan mekanisme subscription untuk memberi tahu banyak objek tentang kejadian yang terjadi pada objek yang mereka pantau, tanpa objek yang dipantau (Subject) perlu tahu siapa yang memantaunya atau apa yang mereka lakukan dengan informasi tersebut.

Dua peran utama dalam Observer Pattern:

Subject (Publisher) — objek yang menyimpan state dan memberi tahu observer ketika state berubah; tidak tahu siapa observer-nya, hanya tahu bahwa mereka ada
Observer (Subscriber) — objek yang mendaftarkan diri untuk mendapat notifikasi; bertanggung jawab atas apa yang dilakukan ketika notifikasi diterima

Tiga properti yang mendefinisikan Observer Pattern:

Loose coupling — Subject tidak bergantung pada Observer konkret; keduanya hanya saling kenal melalui interface
Dynamic subscription — Observer bisa didaftarkan dan dicabut saat runtime, tanpa Subject perlu diubah
Broadcast — satu notifikasi dari Subject menjangkau semua Observer yang terdaftar sekaligus

sequenceDiagram
    participant OS as OrderService (Subject)
    participant IS as InventoryObserver
    participant ES as EmailObserver
    participant AS as AnalyticsObserver

    Note over IS,AS: Register diri ke OrderService
    IS->>OS: Register(inventoryObs)
    ES->>OS: Register(emailObs)
    AS->>OS: Register(analyticsObs)

    Note over OS: Pesanan baru masuk
    OS->>OS: CreateOrder(req)
    OS->>IS: Notify(OrderCreatedEvent)
    OS->>ES: Notify(OrderCreatedEvent)
    OS->>AS: Notify(OrderCreatedEvent)

    IS-->>IS: kurangi stok
    ES-->>ES: kirim email konfirmasi
    AS-->>AS: record event

Masalah yang Dipecahkan #

Observer Pattern menyelesaikan dua masalah yang saling berkaitan: tight coupling antara Subject dan pihak yang bereaksi terhadap perubahannya, dan Open-Closed Principle yang dilanggar setiap kali ada reaksi baru yang perlu ditambahkan.

Masalah: Subject Tahu Terlalu Banyak #

// ANTI-PATTERN: OrderService memanggil semua subsistem secara langsung
func (s *OrderService) CreateOrder(req CreateOrderRequest) (*Order, error) {
    order, err := s.repo.Create(req)
    if err != nil {
        return nil, err
    }

    // OrderService harus tahu tentang inventory
    s.inventorySvc.DeductStock(order.Items)

    // OrderService harus tahu tentang email
    s.emailSvc.SendConfirmation(order.UserEmail, order)

    // OrderService harus tahu tentang analytics
    s.analyticsSvc.TrackPurchase(order)

    // OrderService harus tahu tentang loyalty
    s.loyaltySvc.AddPoints(order.UserID, order.TotalAmount)

    // Tambah WhatsApp notification? Ubah OrderService lagi.
    // Tambah fraud detection? Ubah OrderService lagi.
    // Total coupling yang terjadi: OrderService tahu 4+ subsistem

    return order, nil
}

// BENAR: OrderService hanya tahu bahwa ada event, bukan siapa yang perlu tahu
func (s *OrderService) CreateOrder(req CreateOrderRequest) (*Order, error) {
    order, err := s.repo.Create(req)
    if err != nil {
        return nil, err
    }

    // Satu baris — tidak peduli siapa yang mendengar atau berapa banyak
    s.eventBus.Publish(OrderCreatedEvent{Order: order})

    return order, nil
}
// Tambah WhatsApp? Daftarkan WhatsAppObserver — OrderService tidak berubah
// Tambah fraud detection? Daftarkan FraudObserver — OrderService tidak berubah

Struktur dan Komponen #

classDiagram
    class Subject {
        <<interface>>
        +Register(observer Observer)
        +Unregister(observer Observer)
        +Notify(event Event)
    }

    class Observer {
        <<interface>>
        +OnEvent(event Event)
    }

    class OrderEventBus {
        -observers map[EventType][]Observer
        -mu sync.RWMutex
        +Register(eventType EventType, observer Observer)
        +Unregister(eventType EventType, observer Observer)
        +Publish(event Event)
    }

    class InventoryObserver {
        -inventorySvc InventoryService
        +OnEvent(event Event)
    }

    class EmailObserver {
        -emailSvc EmailService
        +OnEvent(event Event)
    }

    class AnalyticsObserver {
        -analyticsSvc AnalyticsService
        +OnEvent(event Event)
    }

    Subject <|.. OrderEventBus
    Observer <|.. InventoryObserver
    Observer <|.. EmailObserver
    Observer <|.. AnalyticsObserver
    OrderEventBus o-- Observer : manages list

Komponen	Peran	Karakteristik
Subject interface	Kontrak untuk mendaftarkan dan menotifikasi observer	`Register`, `Unregister`, `Notify`
Concrete Subject	Menyimpan daftar observer, men-trigger notifikasi	Thread-safe dengan mutex
Observer interface	Kontrak yang harus dipenuhi semua subscriber	Biasanya satu method: `OnEvent` atau `Update`
Concrete Observer	Bereaksi terhadap event sesuai tanggung jawabnya	Independen satu sama lain

Implementasi Lengkap: Order Event System #

Event Types dan Interfaces #

package events

import "time"

// EventType mendefinisikan jenis event yang bisa dipublikasikan.
type EventType string

const (
    EventOrderCreated   EventType = "order.created"
    EventOrderPaid      EventType = "order.paid"
    EventOrderShipped   EventType = "order.shipped"
    EventOrderCancelled EventType = "order.cancelled"
)

// Event adalah interface untuk semua event dalam sistem.
type Event interface {
    Type() EventType
    OccurredAt() time.Time
}

// OrderCreatedEvent dipublikasikan ketika pesanan baru berhasil dibuat.
type OrderCreatedEvent struct {
    OrderID     string
    UserID      string
    UserEmail   string
    Items       []OrderItem
    TotalAmount float64
    CreatedAt   time.Time
}

func (e OrderCreatedEvent) Type() EventType      { return EventOrderCreated }
func (e OrderCreatedEvent) OccurredAt() time.Time { return e.CreatedAt }

// OrderPaidEvent dipublikasikan ketika pembayaran berhasil dikonfirmasi.
type OrderPaidEvent struct {
    OrderID       string
    UserID        string
    UserEmail     string
    TransactionID string
    Amount        float64
    PaidAt        time.Time
}

func (e OrderPaidEvent) Type() EventType      { return EventOrderPaid }
func (e OrderPaidEvent) OccurredAt() time.Time { return e.PaidAt }

// OrderShippedEvent dipublikasikan ketika pesanan dikirim.
type OrderShippedEvent struct {
    OrderID        string
    UserID         string
    UserEmail      string
    TrackingNumber string
    Courier        string
    ShippedAt      time.Time
}

func (e OrderShippedEvent) Type() EventType      { return EventOrderShipped }
func (e OrderShippedEvent) OccurredAt() time.Time { return e.ShippedAt }

// OrderItem adalah item dalam sebuah pesanan.
type OrderItem struct {
    ProductID string
    Quantity  int
    Price     float64
}

// Observer adalah interface yang harus diimplementasikan semua subscriber.
type Observer interface {
    OnEvent(event Event)
    ObserverName() string // untuk logging dan debugging
}

Concrete Subject: EventBus yang Thread-Safe #

package events

import (
    "fmt"
    "log/slog"
    "sync"
)

// EventBus adalah Subject — mengelola daftar observer per tipe event
// dan mendistribusikan event ke observer yang relevan.
type EventBus struct {
    mu        sync.RWMutex
    observers map[EventType][]Observer
    logger    *slog.Logger
}

// NewEventBus membuat EventBus baru yang siap digunakan.
func NewEventBus(logger *slog.Logger) *EventBus {
    return &EventBus{
        observers: make(map[EventType][]Observer),
        logger:    logger,
    }
}

// Register mendaftarkan observer untuk tipe event tertentu.
// Observer yang sama bisa didaftarkan untuk beberapa tipe event.
func (b *EventBus) Register(eventType EventType, observer Observer) {
    b.mu.Lock()
    defer b.mu.Unlock()

    // Cek apakah observer sudah terdaftar untuk event type ini
    for _, existing := range b.observers[eventType] {
        if existing == observer {
            b.logger.Warn("observer already registered",
                "event_type", eventType,
                "observer", observer.ObserverName())
            return
        }
    }

    b.observers[eventType] = append(b.observers[eventType], observer)
    b.logger.Info("observer registered",
        "event_type", eventType,
        "observer", observer.ObserverName(),
        "total", len(b.observers[eventType]),
    )
}

// Unregister mencabut pendaftaran observer dari tipe event tertentu.
func (b *EventBus) Unregister(eventType EventType, observer Observer) {
    b.mu.Lock()
    defer b.mu.Unlock()

    observers := b.observers[eventType]
    for i, existing := range observers {
        if existing == observer {
            b.observers[eventType] = append(observers[:i], observers[i+1:]...)
            b.logger.Info("observer unregistered",
                "event_type", eventType,
                "observer", observer.ObserverName(),
            )
            return
        }
    }
}

// Publish mendistribusikan event ke semua observer yang terdaftar.
// Notifikasi dilakukan secara sinkron — semua observer selesai sebelum Publish return.
func (b *EventBus) Publish(event Event) {
    b.mu.RLock()
    observers := make([]Observer, len(b.observers[event.Type()]))
    copy(observers, b.observers[event.Type()])
    b.mu.RUnlock()

    b.logger.Info("publishing event",
        "event_type", event.Type(),
        "observer_count", len(observers),
        "occurred_at", event.OccurredAt(),
    )

    for _, observer := range observers {
        func() {
            defer func() {
                if r := recover(); r != nil {
                    b.logger.Error("observer panicked",
                        "observer", observer.ObserverName(),
                        "event_type", event.Type(),
                        "panic", r,
                    )
                }
            }()
            observer.OnEvent(event)
        }()
    }
}

// PublishAsync mendistribusikan event secara asinkron via goroutine.
// Gunakan untuk observer yang bisa berjalan paralel dan independent.
func (b *EventBus) PublishAsync(event Event) {
    b.mu.RLock()
    observers := make([]Observer, len(b.observers[event.Type()]))
    copy(observers, b.observers[event.Type()])
    b.mu.RUnlock()

    for _, observer := range observers {
        go func(obs Observer) {
            defer func() {
                if r := recover(); r != nil {
                    b.logger.Error("async observer panicked",
                        "observer", obs.ObserverName(),
                        "event_type", event.Type(),
                        "panic", fmt.Sprintf("%v", r),
                    )
                }
            }()
            obs.OnEvent(event)
        }(observer)
    }
}

// ObserverCount mengembalikan jumlah observer terdaftar untuk event type tertentu.
func (b *EventBus) ObserverCount(eventType EventType) int {
    b.mu.RLock()
    defer b.mu.RUnlock()
    return len(b.observers[eventType])
}

Concrete Observers #

package observers

import (
    "context"
    "fmt"
    "log/slog"

    "myapp/events"
)

// InventoryObserver mengurangi stok ketika pesanan dibuat.
type InventoryObserver struct {
    inventorySvc InventoryService
    logger       *slog.Logger
}

func NewInventoryObserver(svc InventoryService, logger *slog.Logger) *InventoryObserver {
    return &InventoryObserver{inventorySvc: svc, logger: logger}
}

func (o *InventoryObserver) OnEvent(event events.Event) {
    e, ok := event.(events.OrderCreatedEvent)
    if !ok {
        return // event yang tidak relevan, abaikan
    }

    ctx := context.Background()
    if err := o.inventorySvc.DeductStock(ctx, e.Items); err != nil {
        o.logger.Error("failed to deduct stock",
            "order_id", e.OrderID,
            "error", err,
        )
        return
    }

    o.logger.Info("stock deducted", "order_id", e.OrderID, "items", len(e.Items))
}

func (o *InventoryObserver) ObserverName() string { return "InventoryObserver" }


// EmailObserver mengirim email konfirmasi ketika pesanan dibuat dan dibayar.
type EmailObserver struct {
    emailSvc EmailService
    logger   *slog.Logger
}

func NewEmailObserver(svc EmailService, logger *slog.Logger) *EmailObserver {
    return &EmailObserver{emailSvc: svc, logger: logger}
}

func (o *EmailObserver) OnEvent(event events.Event) {
    ctx := context.Background()

    switch e := event.(type) {
    case events.OrderCreatedEvent:
        if err := o.emailSvc.SendOrderConfirmation(ctx, e.UserEmail, e.OrderID, e.TotalAmount); err != nil {
            o.logger.Error("failed to send confirmation email",
                "order_id", e.OrderID,
                "email", e.UserEmail,
                "error", err,
            )
        }

    case events.OrderShippedEvent:
        if err := o.emailSvc.SendShippingNotification(ctx, e.UserEmail, e.OrderID, e.TrackingNumber, e.Courier); err != nil {
            o.logger.Error("failed to send shipping email",
                "order_id", e.OrderID,
                "error", err,
            )
        }

    default:
        // Event lain tidak perlu dihandle oleh EmailObserver
    }
}

func (o *EmailObserver) ObserverName() string { return "EmailObserver" }


// AnalyticsObserver mencatat semua event ke analytics platform.
type AnalyticsObserver struct {
    analyticsSvc AnalyticsService
    logger       *slog.Logger
}

func NewAnalyticsObserver(svc AnalyticsService, logger *slog.Logger) *AnalyticsObserver {
    return &AnalyticsObserver{analyticsSvc: svc, logger: logger}
}

func (o *AnalyticsObserver) OnEvent(event events.Event) {
    ctx := context.Background()

    properties := map[string]interface{}{
        "event_type":  string(event.Type()),
        "occurred_at": event.OccurredAt(),
    }

    switch e := event.(type) {
    case events.OrderCreatedEvent:
        properties["order_id"] = e.OrderID
        properties["user_id"] = e.UserID
        properties["amount"] = e.TotalAmount
        properties["item_count"] = len(e.Items)

    case events.OrderPaidEvent:
        properties["order_id"] = e.OrderID
        properties["user_id"] = e.UserID
        properties["amount"] = e.Amount
        properties["transaction_id"] = e.TransactionID
    }

    if err := o.analyticsSvc.Track(ctx, string(event.Type()), properties); err != nil {
        o.logger.Warn("failed to track analytics event", "error", err)
        // Non-fatal: analytics gagal tidak boleh menghentikan flow utama
    }
}

func (o *AnalyticsObserver) ObserverName() string { return "AnalyticsObserver" }


// LoyaltyObserver menambahkan loyalty points ketika pembayaran berhasil.
type LoyaltyObserver struct {
    loyaltySvc LoyaltyService
    logger     *slog.Logger
}

func NewLoyaltyObserver(svc LoyaltyService, logger *slog.Logger) *LoyaltyObserver {
    return &LoyaltyObserver{loyaltySvc: svc, logger: logger}
}

func (o *LoyaltyObserver) OnEvent(event events.Event) {
    e, ok := event.(events.OrderPaidEvent)
    if !ok {
        return
    }

    // Hitung poin: 1 poin per 10.000 yang dibayarkan
    points := int(e.Amount / 10000)
    if points == 0 {
        return
    }

    ctx := context.Background()
    if err := o.loyaltySvc.AddPoints(ctx, e.UserID, points); err != nil {
        o.logger.Error("failed to add loyalty points",
            "user_id", e.UserID,
            "points", points,
            "error", err,
        )
        return
    }

    o.logger.Info("loyalty points added", "user_id", e.UserID, "points", points)
}

func (o *LoyaltyObserver) ObserverName() string { return "LoyaltyObserver" }

Subject: OrderService yang Bersih #

package order

import (
    "context"
    "fmt"
    "time"

    "myapp/events"
)

// OrderService adalah Subject — mempublikasikan event, tidak peduli siapa observernya.
type OrderService struct {
    repo     OrderRepository
    eventBus *events.EventBus
}

func NewOrderService(repo OrderRepository, eventBus *events.EventBus) *OrderService {
    return &OrderService{repo: repo, eventBus: eventBus}
}

func (s *OrderService) CreateOrder(ctx context.Context, req CreateOrderRequest) (*Order, error) {
    order, err := s.repo.Create(ctx, req)
    if err != nil {
        return nil, fmt.Errorf("failed to create order: %w", err)
    }

    // Publikasikan event — tidak tahu dan tidak peduli siapa yang subscribe
    s.eventBus.Publish(events.OrderCreatedEvent{
        OrderID:     order.ID,
        UserID:      req.UserID,
        UserEmail:   req.UserEmail,
        Items:       req.Items,
        TotalAmount: order.TotalAmount,
        CreatedAt:   time.Now(),
    })

    return order, nil
}

func (s *OrderService) ConfirmPayment(ctx context.Context, orderID, transactionID string, amount float64) error {
    order, err := s.repo.UpdateStatus(ctx, orderID, "paid")
    if err != nil {
        return fmt.Errorf("failed to update order status: %w", err)
    }

    s.eventBus.Publish(events.OrderPaidEvent{
        OrderID:       orderID,
        UserID:        order.UserID,
        UserEmail:     order.UserEmail,
        TransactionID: transactionID,
        Amount:        amount,
        PaidAt:        time.Now(),
    })

    return nil
}

func (s *OrderService) ShipOrder(ctx context.Context, orderID, trackingNumber, courier string) error {
    order, err := s.repo.UpdateStatus(ctx, orderID, "shipped")
    if err != nil {
        return fmt.Errorf("failed to update order status: %w", err)
    }

    s.eventBus.Publish(events.OrderShippedEvent{
        OrderID:        orderID,
        UserID:         order.UserID,
        UserEmail:      order.UserEmail,
        TrackingNumber: trackingNumber,
        Courier:        courier,
        ShippedAt:      time.Now(),
    })

    return nil
}

Perakitan di main.go #

func main() {
    logger := slog.Default()
    eventBus := events.NewEventBus(logger)

    // Buat observers
    inventoryObs := observers.NewInventoryObserver(inventorySvc, logger)
    emailObs     := observers.NewEmailObserver(emailSvc, logger)
    analyticsObs := observers.NewAnalyticsObserver(analyticsSvc, logger)
    loyaltyObs   := observers.NewLoyaltyObserver(loyaltySvc, logger)

    // Daftarkan observer ke event yang relevan
    eventBus.Register(events.EventOrderCreated, inventoryObs)
    eventBus.Register(events.EventOrderCreated, emailObs)
    eventBus.Register(events.EventOrderCreated, analyticsObs)

    eventBus.Register(events.EventOrderPaid, loyaltyObs)
    eventBus.Register(events.EventOrderPaid, analyticsObs)

    eventBus.Register(events.EventOrderShipped, emailObs)
    eventBus.Register(events.EventOrderShipped, analyticsObs)

    // OrderService tidak perlu tahu semua ini
    orderSvc := order.NewOrderService(orderRepo, eventBus)

    // Tambah FraudDetectionObserver di kemudian hari:
    // eventBus.Register(events.EventOrderCreated, fraudObs)
    // — tidak ada kode lain yang berubah
}

Async Observer: Notifikasi Non-Blocking #

Untuk observer yang bisa berjalan secara independen dan tidak perlu menunggu satu sama lain, async notification via goroutine lebih tepat.

// AsyncEventBus mendistribusikan event ke setiap observer dalam goroutine terpisah.
// Gunakan ketika observer tidak perlu urutan eksekusi yang dijamin.
type AsyncEventBus struct {
    EventBus
    errorHandler func(obs Observer, event Event, err interface{})
}

func NewAsyncEventBus(logger *slog.Logger) *AsyncEventBus {
    bus := &AsyncEventBus{}
    bus.observers = make(map[EventType][]Observer)
    bus.logger = logger
    bus.errorHandler = func(obs Observer, event Event, err interface{}) {
        logger.Error("async observer error",
            "observer", obs.ObserverName(),
            "event_type", event.Type(),
            "error", fmt.Sprintf("%v", err),
        )
    }
    return bus
}

func (b *AsyncEventBus) Publish(event Event) {
    b.mu.RLock()
    observers := make([]Observer, len(b.observers[event.Type()]))
    copy(observers, b.observers[event.Type()])
    b.mu.RUnlock()

    var wg sync.WaitGroup
    for _, observer := range observers {
        wg.Add(1)
        go func(obs Observer) {
            defer wg.Done()
            defer func() {
                if r := recover(); r != nil {
                    b.errorHandler(obs, event, r)
                }
            }()
            obs.OnEvent(event)
        }(observer)
    }
    // Tunggu semua selesai — atau tidak tergantung kebutuhan
    // wg.Wait() // uncomment jika perlu tunggu semua observer selesai
}

Perbandingan sync vs async:

Aspek	Synchronous Publish	Asynchronous Publish
Urutan eksekusi	Dijamin — observer dipanggil berurutan	Tidak dijamin — berjalan paralel
Error handling	Error observer bisa menunda observer berikutnya	Error satu observer tidak mempengaruhi lain
Rollback	Bisa implement rollback jika observer gagal	Sulit — event sudah dipublish
Performa	Observer lambat memblok subject	Observer berjalan paralel
Cocok untuk	Observer yang saling bergantung, butuh urutan	Observer independent, best-effort

Mencegah Memory Leak #

Salah satu masalah terbesar Observer Pattern yang tidak ditangani dengan baik adalah memory leak — observer yang tidak lagi aktif masih terdaftar dan mencegah garbage collection.

// ANTI-PATTERN: observer tidak pernah di-unregister
type SessionHandler struct {
    eventBus *events.EventBus
}

func (h *SessionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // Observer baru didaftarkan setiap request — tidak pernah dihapus!
    obs := &RequestScopedObserver{requestID: r.Header.Get("X-Request-ID")}
    h.eventBus.Register(events.EventOrderCreated, obs)
    // Setelah request selesai, obs tidak dihapus → accumulate terus
}

// BENAR: selalu unregister observer yang tidak lagi dibutuhkan
func (h *SessionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    obs := &RequestScopedObserver{requestID: r.Header.Get("X-Request-ID")}
    h.eventBus.Register(events.EventOrderCreated, obs)
    defer h.eventBus.Unregister(events.EventOrderCreated, obs) // cleanup otomatis

    // handle request...
}

// Atau: gunakan WeakReference-style dengan timeout
type TimedObserver struct {
    observer  Observer
    expiresAt time.Time
}

// EventBus dengan auto-cleanup observer yang sudah expired
func (b *EventBus) CleanupExpired() {
    b.mu.Lock()
    defer b.mu.Unlock()
    now := time.Now()
    for eventType, observers := range b.observers {
        var active []Observer
        for _, obs := range observers {
            if timed, ok := obs.(*TimedObserver); ok {
                if now.Before(timed.expiresAt) {
                    active = append(active, obs)
                }
            } else {
                active = append(active, obs)
            }
        }
        b.observers[eventType] = active
    }
}

Observer Hidup Lebih Lama dari yang Diharapkan

Jika Subject menyimpan referensi ke Observer, Observer tidak akan di-garbage collect meski semua referensi lain ke Observer sudah tidak ada. Ini adalah sumber memory leak yang sangat umum. Selalu sediakan mekanisme Unregister dan pastikan lifecycle Observer dikelola dengan benar — terutama untuk Observer yang dibuat per-request atau per-session.

Testing Observer Pattern #

// MockObserver untuk testing — mencatat semua event yang diterima
type MockObserver struct {
    ReceivedEvents []events.Event
    mu             sync.Mutex
    ShouldPanic    bool
}

func (m *MockObserver) OnEvent(event events.Event) {
    if m.ShouldPanic {
        panic("mock panic")
    }
    m.mu.Lock()
    defer m.mu.Unlock()
    m.ReceivedEvents = append(m.ReceivedEvents, event)
}

func (m *MockObserver) ObserverName() string { return "MockObserver" }

func (m *MockObserver) EventCount() int {
    m.mu.Lock()
    defer m.mu.Unlock()
    return len(m.ReceivedEvents)
}


func TestEventBus_Publish_NotifiesAllObservers(t *testing.T) {
    bus := events.NewEventBus(slog.Default())

    obs1 := &MockObserver{}
    obs2 := &MockObserver{}

    bus.Register(events.EventOrderCreated, obs1)
    bus.Register(events.EventOrderCreated, obs2)

    event := events.OrderCreatedEvent{
        OrderID: "ORD-001", UserID: "user-1",
        TotalAmount: 150000, CreatedAt: time.Now(),
    }
    bus.Publish(event)

    if obs1.EventCount() != 1 {
        t.Errorf("expected obs1 to receive 1 event, got %d", obs1.EventCount())
    }
    if obs2.EventCount() != 1 {
        t.Errorf("expected obs2 to receive 1 event, got %d", obs2.EventCount())
    }
}

func TestEventBus_Unregister_StopsNotification(t *testing.T) {
    bus := events.NewEventBus(slog.Default())
    obs := &MockObserver{}

    bus.Register(events.EventOrderCreated, obs)
    bus.Publish(events.OrderCreatedEvent{OrderID: "ORD-001", CreatedAt: time.Now()})

    bus.Unregister(events.EventOrderCreated, obs)
    bus.Publish(events.OrderCreatedEvent{OrderID: "ORD-002", CreatedAt: time.Now()})

    if obs.EventCount() != 1 {
        t.Errorf("expected 1 event after unregister, got %d", obs.EventCount())
    }
}

func TestEventBus_PanicInObserver_DoesNotStopOthers(t *testing.T) {
    bus := events.NewEventBus(slog.Default())

    panicObs := &MockObserver{ShouldPanic: true}
    normalObs := &MockObserver{}

    bus.Register(events.EventOrderCreated, panicObs)
    bus.Register(events.EventOrderCreated, normalObs)

    // Publish tidak boleh panic meski salah satu observer panic
    require.NotPanics(t, func() {
        bus.Publish(events.OrderCreatedEvent{OrderID: "ORD-001", CreatedAt: time.Now()})
    })

    // Observer yang normal tetap menerima event
    if normalObs.EventCount() != 1 {
        t.Errorf("normal observer should still receive event despite panic in another observer")
    }
}

func TestEventBus_TypeFiltering(t *testing.T) {
    bus := events.NewEventBus(slog.Default())
    obs := &MockObserver{}

    // Hanya subscribe ke EventOrderPaid
    bus.Register(events.EventOrderPaid, obs)

    // Publish EventOrderCreated — tidak seharusnya diterima obs
    bus.Publish(events.OrderCreatedEvent{OrderID: "ORD-001", CreatedAt: time.Now()})
    if obs.EventCount() != 0 {
        t.Error("observer should not receive event for unsubscribed type")
    }

    // Publish EventOrderPaid — seharusnya diterima
    bus.Publish(events.OrderPaidEvent{OrderID: "ORD-001", Amount: 100000, PaidAt: time.Now()})
    if obs.EventCount() != 1 {
        t.Errorf("expected 1 event for subscribed type, got %d", obs.EventCount())
    }
}

func TestEventBus_ConcurrentPublish(t *testing.T) {
    bus := events.NewEventBus(slog.Default())
    obs := &MockObserver{}
    bus.Register(events.EventOrderCreated, obs)

    const goroutines = 100
    var wg sync.WaitGroup
    for i := 0; i < goroutines; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            bus.Publish(events.OrderCreatedEvent{
                OrderID:   fmt.Sprintf("ORD-%d", i),
                CreatedAt: time.Now(),
            })
        }(i)
    }
    wg.Wait()

    if obs.EventCount() != goroutines {
        t.Errorf("expected %d events from concurrent publish, got %d", goroutines, obs.EventCount())
    }
}

Observer vs Event Bus vs Mediator #

Observer Pattern bisa diimplementasikan dengan berbagai pendekatan. Penting memahami perbedaannya.

Aspek	Observer Klasik	Event Bus (Typed)	Mediator
Subject tahu Observer?	Ya — menyimpan list	Tidak — hanya tahu EventType	Tidak — semua lewat mediator
Observer tahu Subject?	Tidak — hanya tahu event	Tidak — hanya tahu EventType	Tidak
Coupling	Subject-Observer loosely coupled	Paling loosely coupled	Komponen tidak tahu satu sama lain
Event filtering	Observer filter sendiri	Filter by type di bus	Mediator memutuskan siapa yang perlu tahu
Cocok untuk	Satu subject, beberapa observer	Banyak subject, banyak observer	Komunikasi kompleks antar komponen

Kapan Menggunakan dan Kapan Tidak #

GUNAKAN Observer jika:
  ✓ Perubahan pada satu objek perlu diketahui objek lain tanpa tight coupling
  ✓ Jumlah dan identitas observer tidak diketahui di compile-time
  ✓ Observer bisa didaftarkan dan dicabut secara dinamis
  ✓ Membangun sistem event-driven atau reactive
  ✓ Ingin menerapkan Open-Closed Principle untuk reaksi terhadap event

HINDARI Observer jika:
  ✗ Chain notifikasi terlalu panjang dan membuat debugging sulit
  ✗ Observer perlu tahu tentang observer lain — gunakan Mediator
  ✗ Urutan observer sangat kritis dan tidak bisa dijamin
  ✗ Sistem sangat kecil dengan 1-2 reaksi yang tidak akan berubah

Checklist Review Observer #

DESAIN:
  □ Subject tidak bergantung pada concrete Observer — hanya pada interface
  □ Observer bisa didaftarkan dan dicabut tanpa mengubah Subject
  □ Event membawa cukup informasi sehingga Observer tidak perlu query balik ke Subject
  □ Setiap Observer handle hanya event yang relevan — abaikan yang lain

IMPLEMENTASI:
  □ EventBus thread-safe — Register, Unregister, Publish menggunakan mutex
  □ Publish mengambil salinan list observer sebelum memanggil (hindari deadlock)
  □ Panic di satu observer tidak menghentikan notifikasi ke observer lain
  □ Error di observer tidak di-propagate ke Subject — Subject tidak peduli

LIFECYCLE:
  □ Ada mekanisme Unregister yang mudah digunakan
  □ Observer di-unregister ketika tidak lagi dibutuhkan
  □ Tidak ada referensi sirkular antara Subject dan Observer

TESTING:
  □ MockObserver mencatat semua event untuk di-assert
  □ Test bahwa semua observer terotifikasi
  □ Test bahwa Unregister menghentikan notifikasi
  □ Test bahwa panic di satu observer tidak mempengaruhi lain
  □ Test thread-safety dengan concurrent Publish

Ringkasan #

Observer mendefinisikan mekanisme subscription — Subject mengumumkan perubahan, Observer bereaksi; keduanya tidak perlu saling kenal secara konkret.

Loose coupling yang sesungguhnya: Subject tidak tahu siapa observernya, berapa banyak, atau apa yang mereka lakukan — hanya memanggil OnEvent() melalui interface.

Typed Event Bus adalah evolusi Observer klasik — filter event by type di bus, bukan di setiap observer; lebih scalable untuk sistem dengan banyak event.

Thread safety wajib: Register, Unregister, dan Publish sering dipanggil dari goroutine berbeda; gunakan sync.RWMutex dan salin list observer sebelum iterasi.

Panic recovery di dalam Publish — satu observer yang panic tidak boleh menghentikan notifikasi ke observer lain; wrap setiap panggilan observer dengan recover().

Memory leak adalah risiko nyata — observer yang tidak di-unregister mencegah garbage collection; selalu gunakan defer Unregister() untuk observer dengan lifecycle terbatas.

Sync vs Async: synchronous untuk observer yang butuh urutan; async (goroutine) untuk observer independent yang bisa berjalan paralel.

Bedakan dari Mediator: Observer untuk “satu subject, banyak subscriber”; Mediator untuk “banyak komponen yang saling berkomunikasi melalui perantara”.

← Sebelumnya: Strategy Berikutnya: Command →