Level 4: Real-time Event Notification System (WebSockets & Redis Pub/Sub)

Welcome to the final level! You are now building a distributed, real-time messaging system.

In real-world production setups, single-node web servers can't scale to millions of concurrent websocket connections. We need to share state across multiple server instances. To do this, we use a message broker like Redis Pub/Sub.

graph TD
    ClientA["WebSocket Client A"] <--> Server1["Server Instance 1"]
    ClientB["WebSocket Client B"] <--> Server2["Server Instance 2"]
    Server1 <-->|"Pub/Sub"| Redis["Redis Message Broker"]
    Server2 <-->|"Pub/Sub"| Redis

Architectural Requirements

1. WebSocket Handler: Upgrade incoming HTTP requests to WebSocket connections.
2. Hub (Broadcaster): Maintain a registry of active client connections. Use channels to:
3. Register new clients.
4. Unregister disconnected clients.
5. Broadcast messages to all connected clients.
6. Redis Pub/Sub Integration:
7. Listen to a Redis channel for external events.
8. When an event is received, publish it to the local Hub to broadcast it to all connected WebSocket clients.
9. Docker Compose Setup:
10. Run the Go application.
11. Spin up a Redis container.

---

The Hub Pattern in Go

In Go, we don't use locks on global arrays to track connections. Instead, we use a central event-loop goroutine (the Hub) that handles registration and broadcasting through channels. This is called "sharing memory by communicating" (Go's concurrency philosophy).

type Hub struct {
    clients    map[*Client]bool
    broadcast  chan []byte
    register   chan *Client
    unregister chan *Client
}

func (h *Hub) Run() {
    for {
        select {
        case client := <-h.register:
            h.clients[client] = true
        case client := <-h.unregister:
            if _, ok := h.clients[client]; ok {
                delete(h.clients, client)
                close(client.send)
            }
        case message := <-h.broadcast:
            for client := range h.clients {
                select {
                case client.send <- message:
                default:
                    close(client.send)
                    delete(h.clients, client)
                }
            }
        }
    }
}

---

Getting Started

1. Look at hub.go to see the Hub and Client struct definitions.
2. Look at hub_test.go which simulates clients and verifies message distribution.
3. Run the tests:

   /usr/local/go/bin/go test -v ./level4_notification_system/...
   

4. Implement the Hub event loop and connection handling.

---

Central Event Hub (hub.go)

package notification

// Connection defines the interface needed for writing messages to clients.
// This decouples our Hub from the gorilla/websocket package, allowing easy testing.
type Connection interface {
    WriteMessage(message []byte) error
    Close() error
}

// Client represents a connected user subscription.
type Client struct {
    Hub  *Hub
    Conn Connection
    Send chan []byte
}

// Hub maintains the set of active clients and broadcasts messages to them.
type Hub struct {
    // Registered clients.
    Clients map[*Client]bool

    // Inbound messages from the clients / Redis.
    Broadcast chan []byte

    // Register requests from the clients.
    Register chan *Client

    // Unregister requests from clients.
    Unregister chan *Client
}

// NewHub initializes a new Hub.
func NewHub() *Hub {
    return &Hub{
        Broadcast:  make(chan []byte),
        Register:   make(chan *Client),
        Unregister: make(chan *Client),
        Clients:    make(map[*Client]bool),
    }
}

// Run executes the main event loop for client registration and broadcasting.
// This function runs forever, processing messages from the channels using select.
func (h *Hub) Run() {
    for {
        select {
        case client := <-h.Register:
            h.Clients[client] = true
        case client := <-h.Unregister:
            if _, ok := h.Clients[client]; ok {
                delete(h.Clients, client)
                close(client.Send)
            }
        case msg := <-h.Broadcast:
            for client := range h.Clients {
                select {
                case client.Send <- msg:
                default:
                    close(client.Send)
                    delete(h.Clients, client)
                }
            }
        }
    }
    // TODO: Implement the select-based event loop.
    // Hint: Loop forever using `for {}`.
    // Inside the loop, select on:
    // 1. <-h.Register: add client to h.Clients map.
    // 2. <-h.Unregister: check if client exists in h.Clients, delete it, and close client.Send.
    // 3. msg := <-h.Broadcast: iterate over h.Clients, send msg to client.Send.
    //    Hint: handle non-blocking sends using select:
    //    select {
    //    case client.Send <- msg:
    //    default:
    //        close(client.Send)
    //        delete(h.Clients, client)
    //    }
}

// WritePump pumps messages from the Hub to the WebSocket connection.
// A goroutine running WritePump is started for each client connection.
func (c *Client) WritePump() {
    defer c.Conn.Close()
    for msg := range c.Send {
        err := c.Conn.WriteMessage(msg)
        if err != nil {
            break
        }
    }
    // TODO: Loop over c.Send channel.
    // Hint: `for msg := range c.Send { ... }`
    // Inside the loop, call c.Conn.WriteMessage(msg).
    // If writing fails, break the loop.
    // Ensure the connection is closed when the pump exits (use defer c.Conn.Close()).
}