Documentation
¶
Index ¶
- func NotificationsStream(userID string) presence.StreamKey
- type ChannelJoinPayload
- type ChannelLeavePayload
- type ChannelMessageRemovePayload
- type ChannelMessageSendPayload
- type ChannelMessageUpdatePayload
- type ConnectionRegistry
- func (cr *ConnectionRegistry) ActiveSessionCount() int
- func (cr *ConnectionRegistry) Add(s *Session)
- func (cr *ConnectionRegistry) AllSessions() []*Session
- func (cr *ConnectionRegistry) GetBySession(sessionID string) (*Session, bool)
- func (cr *ConnectionRegistry) GetUserSessionIDs(userID string) []string
- func (cr *ConnectionRegistry) GetUserSessions(userID string) []*Session
- func (cr *ConnectionRegistry) SendToSession(sessionID string, payload []byte)
- func (cr *ConnectionRegistry) SendToUser(userID string, payload []byte)
- func (cr *ConnectionRegistry) StartGracePeriod(sessionID string, cleanupFn func())
- type Envelope
- type GatewayHandler
- func (gh *GatewayHandler) BroadcastChannelMessage(channelID string, msg *chat.ChannelMessage)
- func (gh *GatewayHandler) BroadcastGroupChannelMessage(ctx context.Context, groupID string, code int16, ...)
- func (gh *GatewayHandler) CleanupRelayedMatch(matchID string)
- func (gh *GatewayHandler) DeliverClusterChat(channelID string, payload []byte)
- func (gh *GatewayHandler) DeliverClusterParty(partyID string, payload []byte)
- func (gh *GatewayHandler) DeliverClusterPresence(kind string, payload []byte)
- func (gh *GatewayHandler) DisconnectSession(sessionID string) error
- func (gh *GatewayHandler) EmitSessionEvent(name, userID, sessionID, username string)
- func (gh *GatewayHandler) RouteMessage(s *Session, payload []byte)
- func (gh *GatewayHandler) SendNotifications(userID string, notifs []*notification.Notification)
- func (gh *GatewayHandler) SendNotificationsToAll(notifs []*notification.Notification)
- func (gh *GatewayHandler) SetClusterMesh(mesh *cluster.Mesh)
- func (gh *GatewayHandler) SetDBPool(pool *pgxpool.Pool)
- func (gh *GatewayHandler) SetEventDispatcher(fn func(name, userID, sessionID, username string))
- func (gh *GatewayHandler) SetHookRegistry(hr *runtime.HookRegistry)
- func (gh *GatewayHandler) SetMatchmaker(mm *matchmaker.Matchmaker)
- func (gh *GatewayHandler) SetMessageRouter(r *presence.LocalMessageRouter)
- func (gh *GatewayHandler) SetNodeID(nodeID string)
- func (gh *GatewayHandler) SetPartyRegistry(reg *party.Registry)
- func (gh *GatewayHandler) SetPresenceLimits(maxStatusBytes, maxSubscriptions int)
- func (gh *GatewayHandler) SetPresenceTracker(pt *presence.PresenceTracker)
- func (gh *GatewayHandler) SetRPCInvoker(...)
- func (gh *GatewayHandler) SetRedisClient(rdb *redis.Client)
- func (gh *GatewayHandler) SetRtHookExecutor(ex *runtime.RtHookExecutor)
- func (gh *GatewayHandler) SetStatusRegistry(sr *presence.StatusRegistry)
- func (gh *GatewayHandler) SetStreamTracker(st *presence.StreamTracker)
- func (gh *GatewayHandler) StartRelayFanoutListener(parent context.Context)
- func (gh *GatewayHandler) StatusFollow(sessionID string, userIDs []string) error
- func (gh *GatewayHandler) StatusUnfollow(sessionID string, userIDs []string) error
- func (gh *GatewayHandler) TrackStatusOnConnect(s *Session)
- func (gh *GatewayHandler) UntrackStatusOnDisconnect(sessionID, userID, username string)
- func (gh *GatewayHandler) Upgrade(w http.ResponseWriter, r *http.Request)
- type MatchCreatePayload
- type MatchDataSendPayload
- type MatchJoinPayload
- type MatchLeavePayload
- type MatchmakerAddPayload
- type MatchmakerRemovePayload
- type PartyAcceptPayload
- type PartyClosePayload
- type PartyCreatePayload
- type PartyDataSendPayload
- type PartyJoinPayload
- type PartyJoinRequestListPayload
- type PartyLeavePayload
- type PartyMatchmakerAddPayload
- type PartyMatchmakerRemovePayload
- type PartyPresencePayload
- type PartyPromotePayload
- type PartyRemovePayload
- type PartyUpdatePayload
- type PingPayload
- type RelayedMatch
- type RpcPayload
- type Session
- type StatusFollowPayload
- type StatusUnfollowPayload
- type StatusUpdatePayload
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NotificationsStream ¶
NotificationsStream returns the stream key for a user's notification inbox.
Types ¶
type ChannelJoinPayload ¶
type ChannelLeavePayload ¶
type ChannelLeavePayload struct {
ChannelID string `json:"channel_id"`
}
type ChannelMessageSendPayload ¶
type ChannelMessageSendPayload struct {
ChannelID string `json:"channel_id"`
Content json.RawMessage `json:"content"`
}
type ChannelMessageUpdatePayload ¶
type ChannelMessageUpdatePayload struct {
ChannelID string `json:"channel_id"`
MessageID string `json:"message_id"`
Content json.RawMessage `json:"content"`
}
type ConnectionRegistry ¶
type ConnectionRegistry struct {
GracePeriod time.Duration // Reconnection grace period
// contains filtered or unexported fields
}
ConnectionRegistry maintains active and recovering user WebSocket sessions.
func NewConnectionRegistry ¶
func NewConnectionRegistry() *ConnectionRegistry
NewConnectionRegistry creates a new ConnectionRegistry.
func (*ConnectionRegistry) ActiveSessionCount ¶
func (cr *ConnectionRegistry) ActiveSessionCount() int
ActiveSessionCount returns sessions currently tracked (including grace-period recovery).
func (*ConnectionRegistry) Add ¶
func (cr *ConnectionRegistry) Add(s *Session)
Add registers a new session, cancelling any active grace timers for reconnection.
func (*ConnectionRegistry) AllSessions ¶
func (cr *ConnectionRegistry) AllSessions() []*Session
AllSessions returns a snapshot of all active sessions.
func (*ConnectionRegistry) GetBySession ¶
func (cr *ConnectionRegistry) GetBySession(sessionID string) (*Session, bool)
GetBySession retrieves a session by ID.
func (*ConnectionRegistry) GetUserSessionIDs ¶
func (cr *ConnectionRegistry) GetUserSessionIDs(userID string) []string
GetUserSessionIDs retrieves active session IDs for a user.
func (*ConnectionRegistry) GetUserSessions ¶
func (cr *ConnectionRegistry) GetUserSessions(userID string) []*Session
GetUserSessions returns active sessions for a user.
func (*ConnectionRegistry) SendToSession ¶
func (cr *ConnectionRegistry) SendToSession(sessionID string, payload []byte)
SendToSession sends a message to a specific active session. When the send buffer is full the session is closed (backpressure).
func (*ConnectionRegistry) SendToUser ¶
func (cr *ConnectionRegistry) SendToUser(userID string, payload []byte)
SendToUser sends a message to all active sessions of a user. When a session send buffer is full that session is closed (backpressure).
func (*ConnectionRegistry) StartGracePeriod ¶
func (cr *ConnectionRegistry) StartGracePeriod(sessionID string, cleanupFn func())
StartGracePeriod kicks off the connection recovery window.
type Envelope ¶
type Envelope struct {
Cid string `json:"cid,omitempty"`
MatchCreate *MatchCreatePayload `json:"match_create,omitempty"`
MatchJoin *MatchJoinPayload `json:"match_join,omitempty"`
MatchLeave *MatchLeavePayload `json:"match_leave,omitempty"`
MatchDataSend *MatchDataSendPayload `json:"match_data_send,omitempty"`
MatchmakerAdd *MatchmakerAddPayload `json:"matchmaker_add,omitempty"`
MatchmakerRemove *MatchmakerRemovePayload `json:"matchmaker_remove,omitempty"`
PartyMatchmakerAdd *PartyMatchmakerAddPayload `json:"party_matchmaker_add,omitempty"`
PartyMatchmakerRemove *PartyMatchmakerRemovePayload `json:"party_matchmaker_remove,omitempty"`
PartyCreate *PartyCreatePayload `json:"party_create,omitempty"`
PartyJoin *PartyJoinPayload `json:"party_join,omitempty"`
PartyLeave *PartyLeavePayload `json:"party_leave,omitempty"`
PartyPromote *PartyPromotePayload `json:"party_promote,omitempty"`
PartyAccept *PartyAcceptPayload `json:"party_accept,omitempty"`
PartyRemove *PartyRemovePayload `json:"party_remove,omitempty"`
PartyClose *PartyClosePayload `json:"party_close,omitempty"`
PartyUpdate *PartyUpdatePayload `json:"party_update,omitempty"`
PartyJoinRequestList *PartyJoinRequestListPayload `json:"party_join_request_list,omitempty"`
PartyDataSend *PartyDataSendPayload `json:"party_data_send,omitempty"`
StatusFollow *StatusFollowPayload `json:"status_follow,omitempty"`
StatusUnfollow *StatusUnfollowPayload `json:"status_unfollow,omitempty"`
StatusUpdate *StatusUpdatePayload `json:"status_update,omitempty"`
Ping *PingPayload `json:"ping,omitempty"`
ChannelJoin *ChannelJoinPayload `json:"channel_join,omitempty"`
ChannelLeave *ChannelLeavePayload `json:"channel_leave,omitempty"`
ChannelMessageSend *ChannelMessageSendPayload `json:"channel_message_send,omitempty"`
ChannelMessageUpdate *ChannelMessageUpdatePayload `json:"channel_message_update,omitempty"`
ChannelMessageRemove *ChannelMessageRemovePayload `json:"channel_message_remove,omitempty"`
Rpc *RpcPayload `json:"rpc,omitempty"`
}
Envelope defines client-to-server and server-to-client WS structures.
type GatewayHandler ¶
type GatewayHandler struct {
Router *match.Router
Matchmaker *matchmaker.Matchmaker
PartyRegistry *party.Registry
HookRegistry *runtime.HookRegistry
PresenceTracker *presence.PresenceTracker
StreamTracker *presence.StreamTracker
StatusRegistry *presence.StatusRegistry
MessageRouter *presence.LocalMessageRouter
MaxStatusBytes int
MaxSubscriptions int
// contains filtered or unexported fields
}
GatewayHandler upgrades connections and starts connection lifecycle loops.
func NewGatewayHandler ¶
func NewGatewayHandler( logger *zap.Logger, tm *auth.TokenManager, reg *ConnectionRegistry, onConnect func(s *Session), onDisconnect func(sessionID, userID, username string), router *match.Router, ) *GatewayHandler
NewGatewayHandler creates a new GatewayHandler.
func (*GatewayHandler) BroadcastChannelMessage ¶
func (gh *GatewayHandler) BroadcastChannelMessage(channelID string, msg *chat.ChannelMessage)
BroadcastChannelMessage implements chat.MessageRouter for local WS fan-out.
func (*GatewayHandler) BroadcastGroupChannelMessage ¶
func (gh *GatewayHandler) BroadcastGroupChannelMessage(ctx context.Context, groupID string, code int16, content, senderID, username string)
BroadcastGroupChannelMessage implements social.ChannelBroadcaster.
func (*GatewayHandler) CleanupRelayedMatch ¶
func (gh *GatewayHandler) CleanupRelayedMatch(matchID string)
CleanupRelayedMatch deletes the relayed match state.
func (*GatewayHandler) DeliverClusterChat ¶
func (gh *GatewayHandler) DeliverClusterChat(channelID string, payload []byte)
DeliverClusterChat delivers a cross-node chat payload to local channel subscribers.
func (*GatewayHandler) DeliverClusterParty ¶
func (gh *GatewayHandler) DeliverClusterParty(partyID string, payload []byte)
DeliverClusterParty delivers a remote party envelope to local party members.
func (*GatewayHandler) DeliverClusterPresence ¶
func (gh *GatewayHandler) DeliverClusterPresence(kind string, payload []byte)
DeliverClusterPresence applies a remote status join/leave/update to local followers.
func (*GatewayHandler) DisconnectSession ¶
func (gh *GatewayHandler) DisconnectSession(sessionID string) error
DisconnectSession force-closes a session (runtime SessionDisconnect).
func (*GatewayHandler) EmitSessionEvent ¶
func (gh *GatewayHandler) EmitSessionEvent(name, userID, sessionID, username string)
EmitSessionEvent dispatches session_start / session_end via configured dispatcher.
func (*GatewayHandler) RouteMessage ¶
func (gh *GatewayHandler) RouteMessage(s *Session, payload []byte)
func (*GatewayHandler) SendNotifications ¶
func (gh *GatewayHandler) SendNotifications(userID string, notifs []*notification.Notification)
SendNotifications implements notification.Deliverer for a single user.
func (*GatewayHandler) SendNotificationsToAll ¶
func (gh *GatewayHandler) SendNotificationsToAll(notifs []*notification.Notification)
SendNotificationsToAll implements notification.Deliverer for connected clients.
func (*GatewayHandler) SetClusterMesh ¶
func (gh *GatewayHandler) SetClusterMesh(mesh *cluster.Mesh)
SetClusterMesh attaches the Redis social mesh for chat/presence/party fan-out.
func (*GatewayHandler) SetDBPool ¶
func (gh *GatewayHandler) SetDBPool(pool *pgxpool.Pool)
SetDBPool configures an optional Postgres pool for chat persistence.
func (*GatewayHandler) SetEventDispatcher ¶
func (gh *GatewayHandler) SetEventDispatcher(fn func(name, userID, sessionID, username string))
SetEventDispatcher configures session_start/session_end event emission.
func (*GatewayHandler) SetHookRegistry ¶
func (gh *GatewayHandler) SetHookRegistry(hr *runtime.HookRegistry)
SetHookRegistry configures runtime hooks for realtime interceptors.
func (*GatewayHandler) SetMatchmaker ¶
func (gh *GatewayHandler) SetMatchmaker(mm *matchmaker.Matchmaker)
SetMatchmaker configures the Matchmaker instance for ticket routing.
func (*GatewayHandler) SetMessageRouter ¶
func (gh *GatewayHandler) SetMessageRouter(r *presence.LocalMessageRouter)
SetMessageRouter configures the local message router for status delivery.
func (*GatewayHandler) SetNodeID ¶
func (gh *GatewayHandler) SetNodeID(nodeID string)
SetNodeID configures the cluster node label used for match metadata and Pub/Sub.
func (*GatewayHandler) SetPartyRegistry ¶
func (gh *GatewayHandler) SetPartyRegistry(reg *party.Registry)
SetPartyRegistry configures the party registry for party matchmaker add/remove.
func (*GatewayHandler) SetPresenceLimits ¶
func (gh *GatewayHandler) SetPresenceLimits(maxStatusBytes, maxSubscriptions int)
SetPresenceLimits configures status string and follow caps.
func (*GatewayHandler) SetPresenceTracker ¶
func (gh *GatewayHandler) SetPresenceTracker(pt *presence.PresenceTracker)
SetPresenceTracker configures the online index for friends∩online (legacy field name).
func (*GatewayHandler) SetRPCInvoker ¶
func (gh *GatewayHandler) SetRPCInvoker(fn func(ctx context.Context, userID, username, id, payload string) (string, int, error))
SetRPCInvoker configures WebSocket custom RPC dispatch.
func (*GatewayHandler) SetRedisClient ¶
func (gh *GatewayHandler) SetRedisClient(rdb *redis.Client)
SetRedisClient configures the Redis client instance.
func (*GatewayHandler) SetRtHookExecutor ¶
func (gh *GatewayHandler) SetRtHookExecutor(ex *runtime.RtHookExecutor)
SetRtHookExecutor configures BeforeRt/AfterRt for WebSocket envelopes.
func (*GatewayHandler) SetStatusRegistry ¶
func (gh *GatewayHandler) SetStatusRegistry(sr *presence.StatusRegistry)
SetStatusRegistry configures the status follow graph and fan-out.
func (*GatewayHandler) SetStreamTracker ¶
func (gh *GatewayHandler) SetStreamTracker(st *presence.StreamTracker)
SetStreamTracker configures the Local-first stream tracker (party/channel/match/status modes).
func (*GatewayHandler) StartRelayFanoutListener ¶
func (gh *GatewayHandler) StartRelayFanoutListener(parent context.Context)
StartRelayFanoutListener subscribes to cross-node relayed match envelopes and delivers locally.
func (*GatewayHandler) StatusFollow ¶
func (gh *GatewayHandler) StatusFollow(sessionID string, userIDs []string) error
StatusFollow adds follow edges for a session (runtime nk.status_follow).
func (*GatewayHandler) StatusUnfollow ¶
func (gh *GatewayHandler) StatusUnfollow(sessionID string, userIDs []string) error
StatusUnfollow removes follow edges for a session (runtime nk.status_unfollow).
func (*GatewayHandler) TrackStatusOnConnect ¶
func (gh *GatewayHandler) TrackStatusOnConnect(s *Session)
TrackStatusOnConnect auto-follows self and optionally Tracks status stream (?status=true).
func (*GatewayHandler) UntrackStatusOnDisconnect ¶
func (gh *GatewayHandler) UntrackStatusOnDisconnect(sessionID, userID, username string)
UntrackStatusOnDisconnect removes status presence and follow edges after grace expiry.
func (*GatewayHandler) Upgrade ¶
func (gh *GatewayHandler) Upgrade(w http.ResponseWriter, r *http.Request)
Upgrade upgrades HTTP requests to WebSocket connection stream.
type MatchCreatePayload ¶
type MatchCreatePayload struct{}
type MatchDataSendPayload ¶
type MatchJoinPayload ¶
type MatchLeavePayload ¶
type MatchLeavePayload struct {
MatchID string `json:"match_id"`
}
type MatchmakerAddPayload ¶
type MatchmakerAddPayload struct {
QueueName string `json:"queue_name"`
MinCount int `json:"min_count"`
MaxCount int `json:"max_count"`
StringProperties map[string]string `json:"string_properties"`
NumericProperties map[string]float64 `json:"numeric_properties"`
CountMultiple int `json:"count_multiple"`
ReversePrecision bool `json:"reverse_precision"`
Query string `json:"query"`
}
type MatchmakerRemovePayload ¶
type MatchmakerRemovePayload struct {
TicketID string `json:"ticket_id"`
}
type PartyAcceptPayload ¶
type PartyAcceptPayload struct {
PartyID string `json:"party_id"`
Presence PartyPresencePayload `json:"presence"`
}
type PartyClosePayload ¶
type PartyClosePayload struct {
PartyID string `json:"party_id"`
}
type PartyCreatePayload ¶
type PartyDataSendPayload ¶
type PartyJoinPayload ¶
type PartyJoinPayload struct {
PartyID string `json:"party_id"`
}
type PartyJoinRequestListPayload ¶
type PartyJoinRequestListPayload struct {
PartyID string `json:"party_id"`
}
type PartyLeavePayload ¶
type PartyLeavePayload struct {
PartyID string `json:"party_id"`
}
type PartyMatchmakerAddPayload ¶
type PartyMatchmakerAddPayload struct {
PartyID string `json:"party_id"`
QueueName string `json:"queue_name"`
MinCount int `json:"min_count"`
MaxCount int `json:"max_count"`
StringProperties map[string]string `json:"string_properties"`
NumericProperties map[string]float64 `json:"numeric_properties"`
CountMultiple int `json:"count_multiple"`
Query string `json:"query"`
}
type PartyPresencePayload ¶
type PartyPromotePayload ¶
type PartyPromotePayload struct {
PartyID string `json:"party_id"`
Presence PartyPresencePayload `json:"presence"`
}
type PartyRemovePayload ¶
type PartyRemovePayload struct {
PartyID string `json:"party_id"`
Presence PartyPresencePayload `json:"presence"`
}
type PartyUpdatePayload ¶
type PingPayload ¶
type PingPayload struct{}
type RelayedMatch ¶
type RelayedMatch struct {
MatchID string
Sessions map[string]*Session
// contains filtered or unexported fields
}
RelayedMatch represents a client-relayed room session.
type RpcPayload ¶
type Session ¶
type Session struct {
ID string
UserID string
Username string
Conn *websocket.Conn
Send chan []byte
CloseOnce sync.Once
IsActive bool
TrackStatus bool
// contains filtered or unexported fields
}
Session wraps a live WebSocket client connection.
func (*Session) CloseConn ¶
func (s *Session) CloseConn()
CloseConn closes the underlying WebSocket connection (nil-safe).
type StatusFollowPayload ¶
type StatusUnfollowPayload ¶
type StatusUnfollowPayload struct {
UserIDs []string `json:"user_ids"`
}
type StatusUpdatePayload ¶
type StatusUpdatePayload struct {
Status *string `json:"status"`
}
StatusUpdatePayload distinguishes JSON null (offline) from empty string (online blank).