Documentation
¶
Overview ¶
Package devicesocket keeps the WebSocket connections of signed-in client devices. It tracks the sockets each user holds, caps how many one user may keep open, answers heartbeats and drops silent sockets, and lets the application push to or kick a device. It knows users and devices only by their IDs: the online, offline and kick callbacks leave recording presence and ending sessions to the application, which owns the accounts.
Every socket is served by two goroutines of its own: a read loop and a writer. Pushes are queued to the writer without waiting, so a device that stops draining its socket costs nobody else anything: it fails its write deadline and is dropped. Handler adapts the socket to the HTTP API.
Index ¶
- Variables
- func Handler(manager *DeviceManager, deps HandlerDeps) app.HandlerFunc
- type Device
- type DeviceManager
- func (dm *DeviceManager) AddDevice(w http.ResponseWriter, r *http.Request, session string, userID int64, ...)
- func (dm *DeviceManager) Broadcast(message string)
- func (dm *DeviceManager) KickDevice(userID int64, deviceID string)
- func (dm *DeviceManager) Online() int
- func (dm *DeviceManager) SendToDevice(userID int64, deviceID string, message string) error
- func (dm *DeviceManager) Shutdown(ctx context.Context)
- func (dm *DeviceManager) StartHeartbeatCheck()
- func (dm *DeviceManager) Stop()
- func (dm *DeviceManager) UpdateHeartbeat(userID int64, deviceID string)
- func (dm *DeviceManager) UpgradeHertz(c *app.RequestContext, session string, userID int64, deviceID string, ...) error
- type HandlerDeps
- type HandshakeError
- type Operator
Constants ¶
This section is empty.
Variables ¶
var ( // ErrSendQueueFull reports a push refused because the device is not // draining its socket. ErrSendQueueFull = errors.New("device send queue is full") // ErrDeviceClosed reports a push to a device whose socket was retired. ErrDeviceClosed = errors.New("device socket is closed") )
Functions ¶
func Handler ¶ added in v1.21.0
func Handler(manager *DeviceManager, deps HandlerDeps) app.HandlerFunc
Handler returns the Hertz handler of the device WebSocket route. It runs behind the session authentication middleware, which puts the account and its session in the context; the handler refuses a request without them, or whose user id is not the signed-in user's, and then upgrades the connection to the device's socket. The device identifier is the label the user's client chose; kicks and pushes address the device by it.
@Summary Device WebSocket @Description Upgrades the connection to the signed-in user's device socket. The client sends "ping" (or "heartbeat") as its heartbeat and is answered "ping"; the server pushes JSON messages such as {"method":"subscribe_update"}, {"method":"kicked_device"} and {"method":"kicked_admin"}. @Tags user @Security BearerAuth @Param userid path int true "User ID of the signed-in user" @Param identifier path string true "Device identifier" @Success 101 {string} string "Switching Protocols" @Router /v1/app/ws/{userid}/{identifier} [get]
Types ¶
type Device ¶
type Device struct {
Session string
DeviceID string
Conn *websocket.Conn
CreatedAt time.Time
// contains filtered or unexported fields
}
Device is one connected socket of a user.
func (*Device) LastPingTime ¶
LastPingTime returns when the device last sent a heartbeat.
type DeviceManager ¶
type DeviceManager struct {
// event callbacks
OnDeviceOnline func(userID int64, deviceID, session string)
OnDeviceOffline func(userID int64, deviceID, session string, createAt time.Time)
OnDeviceKicked func(userID int64, deviceID, session string, operator Operator)
OnMessage func(userID int64, deviceID, session string, message string)
// contains filtered or unexported fields
}
DeviceManager manages devices
func NewDeviceManager ¶
func NewDeviceManager(heartbeatTimeout, checkInterval int, allowedOrigins []string) *DeviceManager
NewDeviceManager creates a device manager that drops a device silent for heartbeatTimeout seconds, sweeping every checkInterval seconds (a value that is not positive falls back to 30). allowedOrigins are the browser origins admitted to connect; an empty list admits every origin, as the CORS middleware does.
func (*DeviceManager) AddDevice ¶
func (dm *DeviceManager) AddDevice(w http.ResponseWriter, r *http.Request, session string, userID int64, deviceID string, maxDevices int)
AddDevice upgrades a net/http request to the device's socket, registers the device and serves it in the background.
func (*DeviceManager) Broadcast ¶
func (dm *DeviceManager) Broadcast(message string)
Broadcast queues a message for every connected device. Devices that are not draining their sockets are skipped.
func (*DeviceManager) KickDevice ¶
func (dm *DeviceManager) KickDevice(userID int64, deviceID string)
KickDevice kicks a device (supports individual device or entire user)
func (*DeviceManager) Online ¶ added in v1.21.0
func (dm *DeviceManager) Online() int
Online returns how many device sockets are connected.
func (*DeviceManager) SendToDevice ¶
func (dm *DeviceManager) SendToDevice(userID int64, deviceID string, message string) error
SendToDevice queues a message for a device of the user, or for all of them when deviceID is empty. It does not wait for the socket: a device that is not draining its socket refuses the message with ErrSendQueueFull.
func (*DeviceManager) Shutdown ¶
func (dm *DeviceManager) Shutdown(ctx context.Context)
Shutdown waits for ctx to end, then stops the sweep and closes every device socket.
func (*DeviceManager) StartHeartbeatCheck ¶
func (dm *DeviceManager) StartHeartbeatCheck()
StartHeartbeatCheck periodically checks for heartbeat timeout devices
func (*DeviceManager) Stop ¶
func (dm *DeviceManager) Stop()
Stop terminates the heartbeat sweep. It is idempotent so lifecycle hooks can call it unconditionally.
func (*DeviceManager) UpdateHeartbeat ¶
func (dm *DeviceManager) UpdateHeartbeat(userID int64, deviceID string)
UpdateHeartbeat records a heartbeat of the device and answers it.
func (*DeviceManager) UpgradeHertz ¶ added in v1.21.0
func (dm *DeviceManager) UpgradeHertz(c *app.RequestContext, session string, userID int64, deviceID string, maxDevices int) error
UpgradeHertz accepts a Hertz request as the device's socket. Hertz writes a response only after the handler returns and hands the connection over afterwards, so the handshake is validated here, its response is set on the request context for Hertz to write, and the device is registered and served in the hijack handler Hertz then runs until the socket closes. The caller answers a HandshakeError with its status; nil means the socket is accepted and the handler must write nothing more.
type HandlerDeps ¶ added in v1.21.0
type HandlerDeps struct {
// MaxDevices returns how many sockets the user may keep open at once;
// nil, or a result of zero or less, leaves the manager's default cap.
MaxDevices func(ctx context.Context, userID int64) int
}
HandlerDeps is what the device socket route needs besides the manager.
type HandshakeError ¶ added in v1.21.0
HandshakeError reports a refused WebSocket handshake with the HTTP status to answer it with.
func (*HandshakeError) Error ¶ added in v1.21.0
func (e *HandshakeError) Error() string
Source Files