Documentation
¶
Overview ¶
Package websocket implements the real-time pub/sub hub that pushes server events to connected GUI clients. It uses gorilla/websocket under the hood and exposes a topic-based broadcast API consumed by the scheduler, gRPC handlers, and notification service.
Topic naming convention:
job:<uuid> — status updates for a specific backup job agent:<uuid> — online/offline/error transitions for an agent notifications:<user_id> — in-app notifications for a specific user
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client represents a single connected WebSocket peer. Each client runs two goroutines: readPump (detects disconnection, handles pong frames) and writePump (serialises outgoing messages onto the wire).
The send channel is the handoff point between the hub's Publish calls and the writePump. It is closed by the hub when the client is unregistered, which causes writePump to drain and exit cleanly.
func NewClient ¶
func NewClient(hub *Hub, w http.ResponseWriter, r *http.Request, topics []string, logger *zap.Logger) (*Client, error)
NewClient creates a Client and upgrades the HTTP connection to WebSocket. topics is the list of pub/sub channels the client wants to receive. The notifications:<user_id> topic must be included by the caller.
Returns an error if the upgrade fails (e.g. the request is not a valid WebSocket handshake).
func (*Client) Run ¶
func (c *Client) Run()
Run registers the client with the hub and starts the read and write pumps. It blocks until the connection closes. The caller should invoke it in a goroutine if they need to return from the HTTP handler immediately — however, since this is called from an HTTP handler that has already completed the upgrade, blocking is fine.
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub is the central pub/sub broker for WebSocket clients. It maintains the registry of connected clients and routes published messages to all clients subscribed to a given topic.
Design: single-writer event loop ¶
All mutations to the client registry (register, unregister) are serialised through a single goroutine — the Run loop — via channels. This eliminates the need for a mutex on the registry map and makes the data flow easy to reason about. Publish is the one exception: it holds a read-lock for the shortest possible time to copy the target set, then sends outside the lock to avoid blocking the event loop while waiting on slow client channels.
Topic format ¶
job:<uuid> — updates for a specific backup job agent:<uuid> — status changes for a specific agent notifications:<user_id> — in-app notifications for a user
func (*Hub) ConnectedCount ¶
ConnectedCount returns the current number of connected WebSocket clients. Intended for metrics and health endpoints.
func (*Hub) Publish ¶
Publish sends msg to every client subscribed to topic. It is safe to call from any goroutine (scheduler, gRPC handlers, etc.). Clients whose send buffer is full are disconnected to prevent backpressure from a slow consumer blocking all other subscribers on the same topic.
func (*Hub) Run ¶
func (h *Hub) Run(ctx interface{ Done() <-chan struct{} })
Run starts the hub's event loop. It must be called exactly once, in its own goroutine. It exits when ctx is cancelled (via server graceful shutdown).
go hub.Run(ctx)
func (*Hub) Subscribe ¶
Subscribe registers client with the hub and adds it to all its topics. Called by the HTTP upgrade handler after the client is initialised.
func (*Hub) Unsubscribe ¶
Unsubscribe removes client from the hub and all its topic subscriptions. Called by the client's readPump when the connection closes.
type Message ¶
type Message struct {
// Type identifies the kind of event so the client can route it correctly.
Type MessageType `json:"type"`
// Topic is the pub/sub channel this message was published on.
// Clients use it to associate the update with the correct UI element.
Topic string `json:"topic"`
// Payload carries the event-specific data. The shape varies by Type:
// - job.status: {"status":"running","started_at":"..."}
// - job.log: {"level":"info","message":"...","timestamp":"..."}
// - agent.status: {"status":"online","ip_address":"..."}
// - agent.metrics: {"cpu_percent":12.5,"mem_percent":60.1,"disk_percent":45.0}
// - notification: {"id":"...","type":"...","title":"...","body":"..."}
// - ping: {} (empty)
Payload any `json:"payload"`
}
Message is the envelope for every WebSocket frame sent to clients. The GUI deserializes this struct and dispatches on Type.
JSON example:
{"type":"job.status","topic":"job:018f...","payload":{"status":"running"}}
type MessageType ¶
type MessageType string
MessageType identifies the kind of event carried by a Message. The GUI uses this field to route the payload to the correct store update.
const ( // MsgJobStatus is sent when a job transitions between states // (pending → running → succeeded | failed). MsgJobStatus MessageType = "job.status" // MsgJobLog is sent for each streamed log line during an active backup. MsgJobLog MessageType = "job.log" // MsgAgentStatus is sent when an agent connects, disconnects, or errors. MsgAgentStatus MessageType = "agent.status" // MsgAgentMetrics is sent on every agent heartbeat with a snapshot of // current host resource utilization (CPU, memory, disk percentages). // Published on the "agent:<uuid>" topic so the detail page can display // live gauges without polling the REST API. MsgAgentMetrics MessageType = "agent.metrics" // MsgNotification is sent when a new in-app notification is created for // the subscribed user. MsgNotification MessageType = "notification" // MsgPing is sent by the hub periodically to keep the connection alive // and let the client detect stale connections. MsgPing MessageType = "ping" )
Source Files