Event Bus y el patrón Observer en Go (Golang): implementación pub/sub

Event Bus y el patrón Observer en Go (Golang): de la teoría a la práctica

El patrón Observer (observador) y su implementación como Event Bus (bus de eventos) son herramientas fundamentales para construir sistemas débilmente acoplados y escalables. En Go, con su filosofía de «comunícate a través de canales, no a través de memoria compartida», estos patrones adquieren una elegancia especial. En este artículo, analizaremos cómo crear tu propio Event Bus en Go, usando gorutinas y canales, y veremos casos de uso típicos.

Patrón Observer: esencia y problema

Imagina que tienes un sistema donde un objeto (editor) debe notificar a muchos otros (suscriptores) sobre un cambio en su estado. Sin Observer, tendrías que acoplar rígidamente el editor con cada suscriptor. Observer resuelve esto introduciendo una interfaz de suscripción: el editor mantiene una lista de suscriptores y llama a su método Update() cuando ocurre un evento.

Event Bus es un bus centralizado que actúa como intermediario. Cualquier componente puede publicar un evento, y cualquier otro puede suscribirse a él, sin conocer la existencia del otro.

¿Por qué Go es ideal para Event Bus?

• Gorutinas y canales: procesamiento asíncrono de eventos sin bloqueos.
• Interfaces: permiten definir fácilmente contratos para los manejadores.
• Ausencia de herencia: la composición a través de canales y funciones es más natural que la implementación clásica de POO.

Implementación simple de Event Bus en Go

Empecemos con una versión básica, donde los eventos son cadenas y los suscriptores son funciones manejadoras.

package main

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

// EventBus — nuestro bus de eventostype EventBus struct {    subscribers map[string][]func(string)    mu          sync.RWMutex}

// NewEventBus crea un nuevo busfunc NewEventBus() *EventBus {    return &EventBus{        subscribers: make(map[string][]func(string)),    }}

// Subscribe suscribe un manejador a un eventofunc (eb *EventBus) Subscribe(event string, handler func(string)) {    eb.mu.Lock()    defer eb.mu.Unlock()    eb.subscribers[event] = append(eb.subscribers[event], handler)}

// Publish publica un evento y llama a todos los suscriptores de forma síncronafunc (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("Inicio de sesión: %s", data)    })        bus.Subscribe("user.login", func(data string) {        fmt.Printf("Auditoría: %s inició sesión", data)    })        bus.Publish("user.login", "alice@example.com")    // Salida:    // Inicio de sesión: alice@example.com    // Auditoría: alice@example.com inició sesión}

Este código funciona, pero hay un detalle: los manejadores se llaman de forma síncrona. Si un manejador se bloquea, bloqueará todo el bus.

Versión asíncrona con canales

Añadamos asincronía a través de canales con búfer. Cada suscriptor recibirá su propio canal.

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