Event Bus and Observer Pattern in Go (Golang): pub/sub Implementation

Event Bus and Observer Pattern in Go (Golang): From Theory to Practice

The Observer pattern and its implementation as an Event Bus are fundamental tools for building loosely coupled, scalable systems. In Go, with its philosophy of "communicate via channels, not via shared memory," these patterns acquire a special elegance. In this article, we will analyze how to create your own Event Bus in Go using goroutines and channels, and consider typical usage scenarios.

The Observer Pattern: Essence and Problem

Imagine you have a system where one object (the publisher) must notify many others (subscribers) about a change in its state. Without Observer, you would have to tightly couple the publisher to each subscriber. Observer solves this by introducing a subscription interface: the publisher stores a list of subscribers and calls their Update() method when an event occurs.

Event Bus is a centralized bus that acts as an intermediary. Any component can publish an event, and any other can subscribe to it without knowing about each other's existence.

Why Go is Ideal for Event Bus?

• Goroutines and channels: asynchronous event processing without locks.
• Interfaces: allow you to easily define contracts for handlers.
• No inheritance: composition via channels and functions is more natural than a classic OOP implementation.

Simple Event Bus Implementation in Go

Let's start with a basic version where events are strings and subscribers are handler functions.

package main

import (    "fmt"    "sync"    "time")

// EventBus — our event bustype EventBus struct {    subscribers map[string][]func(string)    mu          sync.RWMutex}

// NewEventBus creates a new busfunc NewEventBus() *EventBus {    return &EventBus{        subscribers: make(map[string][]func(string)),    }}

// Subscribe subscribes a handler to an eventfunc (eb *EventBus) Subscribe(event string, handler func(string)) {    eb.mu.Lock()    defer eb.mu.Unlock()    eb.subscribers[event] = append(eb.subscribers[event], handler)}

// Publish publishes an event and calls all subscribers synchronouslyfunc (eb *EventBus) Publish(event string, data string) {    eb.mu.RLock()    defer eb.mu.RUnlock()    if handlers, ok := eb.subscribers[event]; ok {        for _, handler := range handlers {            handler(data)        }    }}

func main() {    bus := NewEventBus()        bus.Subscribe("user.login", func(data string) {        fmt.Printf("Login: %s", data)    })        bus.Subscribe("user.login", func(data string) {        fmt.Printf("Audit: %s logged in", data)    })        bus.Publish("user.login", "alice@example.com")    // Output:    // Login: alice@example.com    // Audit: alice@example.com logged in}

This code works, but there is a nuance: handlers are called synchronously. If one handler hangs, it will block the entire bus.

Asynchronous Version with Channels

Let's add asynchrony using buffered channels. Each subscriber will receive its own channel.

type AsyncEventBus struct {    subscribers map[string][]chan string    mu          sync.RWMutex}

func NewAsyncEventBus() *AsyncEventBus {    return &AsyncEventBus{        subscribers: make(map[string][]chan string),    }}

func (eb *AsyncEventBus) Subscribe(event string, buffer int) <-chan string {    eb.mu.Lock()    defer eb.mu.Unlock()    ch := make(chan string, buffer)    eb.subscribers[event] = append(eb.subscribers[event], ch)    return ch}

func (eb *AsyncEventBus) Publish(event string, data string) {    eb.mu.RLock()    defer eb.mu.RUnlock()    if channe