Go and Event-Driven WebSocket - Enhancing Flexibility

In this episode, we will explore Event-Driven Architecture and how to adapt a WebSocket Server to support event-driven operations, making your system more flexible and easier to scale!

What is Event-Driven Architecture?

Event-Driven Architecture is a design pattern that operates by sending and responding to events, such as button presses, new connections, or message deliveries.

Advantages of Event-Driven Architecture :

• Flexibility : New features can be added without impacting other parts of the system.
• Scalability : Breaking operations into events reduces redundant processing.
• Ideal for WebSocket : Better manages real-time communication.

Implementing Event-Driven in WebSocket

In a WebSocket Server, we can utilize Event-Driven design by :

• Separating operations by events such as onConnect, onMessage, onDisconnect.
• Using an Event Bus or Pub/Sub to distribute events within the system.
• Adding handlers for managing each event. 

Example of Event-Driven Operation on WebSocket

1. Define Event Structure

Name : Event name (e.g., message, disconnect).

Data : Information related to the event.

type Event struct {
    Name string      `json:"name"`
    Data interface{} `json:"data"`
}

2. Create Event Dispatcher

Register: Register events and handlers.

Dispatch: Distribute events to registered handlers.

package main

import "sync"

type EventHandler func(data interface{})

type EventDispatcher struct {
    handlers map[string][]EventHandler
    mu       sync.RWMutex
}

func NewEventDispatcher() *EventDispatcher {
    return &EventDispatcher{handlers: make(map[string][]EventHandler)}
}

func (d *EventDispatcher) Register(event string, handler EventHandler) {
    d.mu.Lock()
    defer d.mu.Unlock()
    d.handlers[event] = append(d.handlers[event], handler)
}

func (d *EventDispatcher) Dispatch(event string, data interface{}) {
    d.mu.RLock()
    defer d.mu.RUnlock()
    if handlers, found := d.handlers[event]; found {
        for _, handler := range handlers {
            go handler(data) // เรียกใช้ Handler แบบ Concurrent
        }
    }
}

3. Use Event Dispatcher in WebSocket

package main

import (
    "encoding/json"
    "log"
    "net/http"

    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
    CheckOrigin: func(r *http.Request) bool {
        return true
    },
}

var dispatcher = NewEventDispatcher()

func handleConnections(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println("Error upgrading connection:", err)
        return
    }
    defer conn.Close()

    dispatcher.Dispatch("connect", conn)

    for {
        _, msg, err := conn.ReadMessage()
        if err != nil {
            log.Println("Error reading message:", err)
            dispatcher.Dispatch("disconnect", conn)
            break
        }

        var event Event
        if err := json.Unmarshal(msg, &event); err != nil {
            log.Println("Error unmarshaling event:", err)
            continue
        }

        dispatcher.Dispatch(event.Name, event.Data)
    }
}

func main() {
    dispatcher.Register("connect", func(data interface{}) {
        conn := data.(*websocket.Conn)
        log.Println("New client connected:", conn.RemoteAddr())
    })

    dispatcher.Register("message", func(data interface{}) {
        log.Printf("Message received: %v", data)
    })

    dispatcher.Register("disconnect", func(data interface{}) {
        conn := data.(*websocket.Conn)
        log.Println("Client disconnected:", conn.RemoteAddr())
    })

    http.HandleFunc("/ws", handleConnections)
    log.Println("WebSocket server started at :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

Code Explanation

• EventDispatcher : Manages events and distributes them to relevant handlers.
• Register : Registers events such as connect, message, disconnect.
• Dispatch : Calls the relevant handler when an event occurs.
• onConnect : Handles when a user connects.
• onMessage : Handles received messages.
• onDisconnect : Handles when a user disconnects.

Additional Improvements

• Use Redis Pub/Sub : For distributing events between multiple WebSocket servers.
• Add Middleware : For authorization checks before dispatching events.
• Implement Event Queue : Use RabbitMQ or Kafka to manage large-scale event processing.

Summary

• Event-driven operations enhance flexibility and scalability.
• Use an Event Dispatcher to manage events within the WebSocket Server.
• Support adding new features without affecting other parts of the system.

Fun Activity

Try building a real-time chat system using an event-driven approach with events like joinRoom, sendMessage, and leaveRoom, and add a feature to broadcast messages to everyone in the room!

In the next episode, we will discuss connecting WebSocket with Microservices to support large systems and break operations into smaller parts!