websocket

package
v0.11.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 10, 2026 License: Apache-2.0 Imports: 10 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrConnectionClosed is returned when attempting to send on a closed connection
	ErrConnectionClosed = &ConnectionError{Message: "connection is closed"}
)

Common connection errors

Functions

func IsOrgConnectionLimitError 

func IsOrgConnectionLimitError(err error) bool

IsOrgConnectionLimitError checks if an error is an OrgConnectionLimitError

Types

type Connection 

type Connection struct {
	// GatewayID identifies the gateway instance (UUID from gateway registration)
	GatewayID string

	// ConnectionID provides a unique identifier for this specific connection instance.
	// Used to distinguish between multiple connections from the same gateway (clustering).
	ConnectionID string

	// OrganizationID identifies the organization that owns this gateway connection.
	// Used for per-organization connection limit tracking and cleanup.
	OrganizationID string

	// ConnectedAt records when the connection was established
	ConnectedAt time.Time

	// LastHeartbeat records the timestamp of the most recent heartbeat (pong) received.
	// Updated automatically by the pong handler to track connection liveness.
	LastHeartbeat time.Time

	// Transport provides the underlying protocol implementation for message delivery.
	// Abstraction allows swapping WebSocket for other protocols without changing business logic.
	Transport Transport

	// AuthToken stores the API key used to authenticate this connection.
	// Can be used for re-validation or audit logging.
	AuthToken string

	// DeliveryStats tracks event delivery statistics for this connection
	DeliveryStats *DeliveryStats
	// contains filtered or unexported fields
}

Connection represents an active gateway connection with metadata and lifecycle management. This wrapper decouples connection management logic from the underlying transport protocol.

Design rationale: By wrapping the Transport interface, we can:

  • Track connection metadata (gateway ID, connection time, heartbeat status)
  • Support multiple transport implementations (WebSocket, SSE, gRPC)
  • Manage connection lifecycle (connect, heartbeat, disconnect) uniformly

func NewConnection 

func NewConnection(gatewayID, connectionID string, transport Transport, authToken string, orgID string) *Connection

NewConnection creates a new Connection wrapper with the provided parameters.

Parameters:

  • gatewayID: UUID of the authenticated gateway
  • connectionID: Unique identifier for this connection instance
  • transport: Transport implementation (e.g., WebSocketTransport)
  • authToken: API key used for authentication
  • orgID: UUID of the organization that owns the gateway

Returns a fully initialized Connection ready for message delivery.

func (*Connection) Close 

func (c *Connection) Close(code int, reason string) error

Close terminates the connection gracefully with a close code and reason. This method is idempotent - calling it multiple times is safe.

Parameters:

  • code: Protocol-specific close code (e.g., 1000 for normal WebSocket closure)
  • reason: Human-readable reason for closure

Returns an error if the close operation fails (ignored if already closed).

func (*Connection) GetLastHeartbeat 

func (c *Connection) GetLastHeartbeat() time.Time

GetLastHeartbeat returns the timestamp of the most recent heartbeat. Used by the connection manager to detect stale/dead connections.

Thread-safe for concurrent access.

func (*Connection) GetStatus 

func (c *Connection) GetStatus(heartbeatTimeout time.Duration) ConnectionStatus

GetStatus returns the current connection status for monitoring and stats API.

Status values:

  • "connected": Connection is active and receiving heartbeats
  • "stale": No heartbeat received within timeout period (but not yet closed)
  • "closed": Connection has been explicitly closed

Parameters:

  • heartbeatTimeout: Duration after which a connection is considered stale

Thread-safe for concurrent access.

func (*Connection) IsClosed 

func (c *Connection) IsClosed() bool

IsClosed returns true if the connection has been explicitly closed. Thread-safe for concurrent access.

func (*Connection) Send 

func (c *Connection) Send(message []byte) error

Send delivers a message to the gateway through the underlying transport. This method is thread-safe and can be called concurrently.

Parameters:

  • message: The message payload to send (typically JSON-encoded)

Returns an error if the send fails or the connection is already closed.

func (*Connection) UpdateHeartbeat 

func (c *Connection) UpdateHeartbeat()

UpdateHeartbeat records the current time as the last heartbeat timestamp. Called automatically by the pong handler when heartbeat frames are received.

Thread-safe for concurrent access.

type ConnectionError 

type ConnectionError struct {
	Message string
}

ConnectionError represents connection-specific errors

func (*ConnectionError) Error 

func (e *ConnectionError) Error() string

type ConnectionStatus 

type ConnectionStatus struct {
	GatewayID     string    `json:"gatewayId"`
	ConnectionID  string    `json:"connectionId"`
	ConnectedAt   time.Time `json:"connectedAt"`
	LastHeartbeat time.Time `json:"lastHeartbeat"`
	Status        string    `json:"status"` // "connected", "stale", "closed"
}

ConnectionStatus represents the current state of a connection for monitoring.

type DeliveryStats 

type DeliveryStats struct {
	// TotalEventsSent tracks the cumulative number of events sent to all gateways.
	// Updated atomically using atomic.AddInt64() to ensure thread-safety.
	TotalEventsSent int64

	// FailedDeliveries tracks the cumulative number of event delivery failures.
	// A failure occurs when:
	//   - Gateway is not connected (no active WebSocket connection)
	//   - Send operation fails (e.g., connection closed during send)
	//   - Payload exceeds maximum size limit
	FailedDeliveries int64

	// LastFailureTime records the timestamp of the most recent delivery failure.
	// Not atomic - updated under lock or during single-threaded stats query.
	LastFailureTime time.Time

	// LastFailureReason contains a human-readable description of the most recent failure.
	// Examples: "gateway not connected", "send timeout", "payload too large"
	// Not atomic - updated under lock or during single-threaded stats query.
	LastFailureReason string
}

DeliveryStats tracks event delivery statistics in memory using atomic operations. This structure provides operational visibility into event delivery success/failure rates without requiring database persistence.

Design rationale: Atomic counters enable lock-free concurrent updates from multiple goroutines handling event delivery. Stats reset on server restart, which is acceptable for operational monitoring (persistent metrics can be added via Prometheus later).

func (*DeliveryStats) GetFailedCount 

func (s *DeliveryStats) GetFailedCount() int64

GetFailedCount returns the current value of failed deliveries. Uses atomic load to ensure visibility of concurrent updates.

func (*DeliveryStats) GetSuccessRate 

func (s *DeliveryStats) GetSuccessRate() float64

GetSuccessRate calculates the percentage of successful event deliveries. Returns 100.0 if no events have been sent (avoiding division by zero).

func (*DeliveryStats) GetTotalSent 

func (s *DeliveryStats) GetTotalSent() int64

GetTotalSent returns the current value of total events sent. Uses atomic load to ensure visibility of concurrent updates.

func (*DeliveryStats) IncrementFailed 

func (s *DeliveryStats) IncrementFailed(reason string)

IncrementFailed atomically increments the failed deliveries counter and records the failure details.

Parameters:

  • reason: Human-readable description of why the delivery failed

Note: LastFailureTime and LastFailureReason updates are not atomic. This is acceptable for monitoring purposes where exact synchronization is not critical.

func (*DeliveryStats) IncrementTotalSent 

func (s *DeliveryStats) IncrementTotalSent()

IncrementTotalSent atomically increments the total events sent counter. This method is thread-safe and can be called concurrently from multiple goroutines.

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

Manager handles the lifecycle of gateway WebSocket connections. It maintains an in-memory registry of active connections, manages heartbeats, and handles graceful/ungraceful disconnections.

Design rationale: sync.Map provides thread-safe concurrent access optimized for read-heavy workloads (event delivery lookups). The registry maps gateway IDs to slices of connections to support multiple connections per gateway (clustering).

func NewManager 

func NewManager(config ManagerConfig, gatewayRepo repository.GatewayRepository, slogger *slog.Logger) *Manager

NewManager creates a new connection manager with the provided configuration

func (*Manager) CanAcceptOrgConnection 

func (m *Manager) CanAcceptOrgConnection(orgID string) bool

CanAcceptOrgConnection checks if the organization can accept a new connection without actually adding it. Use this for pre-upgrade validation.

func (*Manager) GetAllConnections 

func (m *Manager) GetAllConnections() map[string][]*Connection

GetAllConnections returns all active connections across all gateways. Used by the stats API to provide operational visibility.

Returns a map of gatewayID -> []*Connection

func (*Manager) GetConnectionCount 

func (m *Manager) GetConnectionCount() int

GetConnectionCount returns the total number of active connections

func (*Manager) GetConnections 

func (m *Manager) GetConnections(gatewayID string) []*Connection

GetConnections retrieves all connections for a specific gateway ID. Returns an empty slice if the gateway has no active connections.

Thread-safe for concurrent access.

func (*Manager) GetOrgConnectionStats 

func (m *Manager) GetOrgConnectionStats(orgID string) OrgConnectionStats

GetOrgConnectionStats returns connection statistics for a specific organization

func (*Manager) IncrementDisconnections 

func (m *Manager) IncrementDisconnections()

func (*Manager) IncrementFailedConnections 

func (m *Manager) IncrementFailedConnections()

func (*Manager) IncrementSuccessfulConnections 

func (m *Manager) IncrementSuccessfulConnections()

func (*Manager) IncrementTotalEventsSent 

func (m *Manager) IncrementTotalEventsSent()

func (*Manager) Register 

func (m *Manager) Register(gatewayID string, transport Transport, authToken string,
	orgID string) (*Connection, error)

Register adds a new connection to the registry and starts heartbeat monitoring. Returns an error if the maximum connection limit is reached.

Parameters:

  • gatewayID: UUID of the authenticated gateway
  • transport: Transport implementation for message delivery
  • authToken: API key used for authentication
  • orgID: UUID of the organization that owns the gateway

Returns the Connection instance and any error encountered.

Design decision: Support multiple connections per gateway ID by storing connections in a slice. This enables gateway clustering where multiple instances share the same gateway identity.

func (*Manager) SetConnectionHooks 

func (m *Manager) SetConnectionHooks(onConnect, onDisconnect func(gatewayID string) error)

SetConnectionHooks registers optional callbacks invoked on every gateway connect and disconnect. onConnect is called after a connection is stored; onDisconnect after it is removed. Callbacks are invoked synchronously and must not acquire the manager's locks. This method must be called only during initialization, before the gateway begins accepting connections. It is not safe to call concurrently with Register or Unregister.

func (*Manager) Shutdown 

func (m *Manager) Shutdown()

Shutdown gracefully closes all connections and stops heartbeat monitoring. Waits for all connection handler goroutines to exit before returning.

This method should be called during server shutdown to cleanly terminate all gateway connections with a normal closure code.

func (*Manager) Unregister 

func (m *Manager) Unregister(gatewayID, connectionID string)

Unregister removes a connection from the registry and closes it gracefully. This method is idempotent - calling it multiple times is safe.

Parameters:

  • gatewayID: UUID of the gateway
  • connectionID: Unique identifier of the connection to remove

type ManagerConfig 

type ManagerConfig struct {
	MaxConnections       int           // Maximum concurrent connections (default 1000)
	HeartbeatInterval    time.Duration // Ping interval (default 20s)
	HeartbeatTimeout     time.Duration // Pong timeout (default 30s)
	MaxConnectionsPerOrg int           // Maximum connections per organization (default 3)
	MetricsLogEnabled    bool          // Enable periodic metrics logging (default true)
	MetricsLogInterval   time.Duration // Interval between metrics log entries (default 10s)
}

ManagerConfig contains configuration parameters for the connection manager

func DefaultManagerConfig 

func DefaultManagerConfig() ManagerConfig

DefaultManagerConfig returns sensible default configuration values

type OrgConnectionLimitError 

type OrgConnectionLimitError struct {
	OrganizationID string
	CurrentCount   int
	MaxAllowed     int
}

OrgConnectionLimitError is returned when an organization has reached its connection limit

func (*OrgConnectionLimitError) Error 

func (e *OrgConnectionLimitError) Error() string

type OrgConnectionStats 

type OrgConnectionStats struct {
	OrganizationID string `json:"organizationId"`
	CurrentCount   int    `json:"currentCount"`
	MaxAllowed     int    `json:"maxAllowed"`
}

type Transport 

type Transport interface {
	// Send delivers a message to the connected client.
	// Returns an error if the send operation fails (e.g., connection closed, timeout).
	//
	// Parameters:
	//   - message: The message payload to send (typically JSON-encoded)
	//
	// The implementation should handle protocol-specific framing and encoding.
	Send(message []byte) error

	// Close terminates the transport connection gracefully.
	// The implementation should send appropriate close frames/messages as per
	// the protocol specification (e.g., WebSocket close frame with code 1000).
	//
	// Parameters:
	//   - code: Protocol-specific close code (e.g., WebSocket close codes)
	//   - reason: Human-readable reason for connection closure
	Close(code int, reason string) error

	// SetReadDeadline sets the deadline for reading from the transport.
	// Used for implementing heartbeat/keepalive timeout detection.
	//
	// Parameters:
	//   - deadline: Absolute time when read operations should timeout
	//
	// A zero time value disables the deadline.
	SetReadDeadline(deadline time.Time) error

	// SetWriteDeadline sets the deadline for writing to the transport.
	// Used to prevent indefinite blocking on send operations.
	//
	// Parameters:
	//   - deadline: Absolute time when write operations should timeout
	//
	// A zero time value disables the deadline.
	SetWriteDeadline(deadline time.Time) error

	// EnablePongHandler configures automatic handling of pong frames for heartbeat.
	// For WebSocket, this sets up the pong handler. Other transports may implement
	// equivalent keepalive mechanisms.
	//
	// Parameters:
	//   - handler: Callback function invoked when a pong frame is received
	//
	// This is called to reset read deadlines and maintain connection liveness.
	EnablePongHandler(handler func(string) error)

	// SendPing sends a ping frame to test connection liveness.
	// Returns an error if the ping cannot be sent.
	//
	// For protocols without built-in ping/pong, implementations may use
	// application-level heartbeat messages.
	SendPing() error
}

Transport defines an abstraction layer for protocol-independent message delivery. This interface allows the system to switch between different transport mechanisms (WebSocket, Server-Sent Events, gRPC, etc.) without modifying business logic.

Design rationale: The transport abstraction isolates protocol-specific code from event routing and connection management logic, enabling future protocol changes without rewriting the core system.

func NewWebSocketTransport 

func NewWebSocketTransport(conn *websocket.Conn) Transport

NewWebSocketTransport creates a new WebSocket transport wrapper.

Parameters:

  • conn: The established gorilla/websocket connection

Returns a Transport implementation backed by WebSocket.

type WebSocketTransport 

type WebSocketTransport struct {
	// contains filtered or unexported fields
}

WebSocketTransport implements the Transport interface using gorilla/websocket. This provides the concrete WebSocket protocol implementation while isolating WebSocket-specific code from business logic.

Design rationale: By implementing the Transport interface, we can swap WebSocket for other protocols (SSE, gRPC) without changing the Connection or Manager code.

func (*WebSocketTransport) Close 

func (t *WebSocketTransport) Close(code int, reason string) error

Close terminates the WebSocket connection with a close frame.

Parameters:

  • code: WebSocket close code (e.g., 1000 for normal closure)
  • reason: Human-readable close reason

Returns an error if sending the close frame fails.

func (*WebSocketTransport) EnablePongHandler 

func (t *WebSocketTransport) EnablePongHandler(handler func(string) error)

EnablePongHandler configures the automatic pong frame handler. Called when a pong frame is received in response to a ping.

Parameters:

  • handler: Callback invoked when pong frame arrives

func (*WebSocketTransport) ReadMessage 

func (t *WebSocketTransport) ReadMessage() (messageType int, payload []byte, err error)

ReadMessage reads the next message from the WebSocket connection. This is used by connection handlers to detect disconnections and handle incoming messages.

Returns:

  • messageType: The type of message (Text, Binary, Close, Ping, Pong)
  • payload: The message data
  • error: Any error encountered during read

func (*WebSocketTransport) Send 

func (t *WebSocketTransport) Send(message []byte) error

Send delivers a message to the WebSocket client as a text frame.

Parameters:

  • message: The message payload (typically JSON-encoded)

Returns an error if the write fails or the connection is closed.

func (*WebSocketTransport) SendPing 

func (t *WebSocketTransport) SendPing() error

SendPing sends a WebSocket ping frame to test connection liveness. The client should respond with a pong frame.

Returns an error if the ping cannot be sent.

func (*WebSocketTransport) SetReadDeadline 

func (t *WebSocketTransport) SetReadDeadline(deadline time.Time) error

SetReadDeadline sets the deadline for read operations. Used to detect heartbeat timeouts.

Parameters:

  • deadline: Absolute time when reads should timeout (zero disables)

func (*WebSocketTransport) SetWriteDeadline 

func (t *WebSocketTransport) SetWriteDeadline(deadline time.Time) error

SetWriteDeadline sets the deadline for write operations. Prevents indefinite blocking on slow clients.

Parameters:

  • deadline: Absolute time when writes should timeout (zero disables)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.