Documentation
¶
Overview ¶
Package websocket provides typed channels, room publication and the versioned Foundry WebSocket protocol. Applications own domain handlers and DTOs.
Index ¶
- Constants
- Variables
- func DeclaredModule(name foundation.ProviderID, key foundation.Key[*Hub], config ServerConfig, ...) foundation.Module
- func DisconnectConnection(ctx context.Context, source PublisherSource, id ConnectionID) error
- func DisconnectSubject[M model.Identifiable, K any](ctx context.Context, source PublisherSource, guard auth.Guard[M], ...) error
- func Module(name foundation.ProviderID, key foundation.Key[*Hub], config ServerConfig, ...) foundation.Module
- func NotPublished(err error) bool
- func Route(hub *Hub, path string, middleware ...foundryhttp.Middleware) foundryhttp.RouteRegistration
- func SharedModule(name foundation.ProviderID, key foundation.Key[*Hub], ...) foundation.Module
- func ValidateClusterConnection(ctx context.Context, key ClusterKey, instance InstanceID, ...) error
- func ValidateClusterOperation(ctx context.Context, key ClusterKey) error
- type Action
- type Anonymous
- type BackendConnection
- type Channel
- func OwnedRooms[C any, R comparable, S model.Identifiable](id ChannelID, rooms Rooms[R], guard auth.Guard[S], ...) Channel[C, R, S]
- func Private[C, R any, S model.Identifiable](id ChannelID, rooms Rooms[R], guard auth.Guard[S], ...) Channel[C, R, S]
- func Public[C, R any](id ChannelID, rooms Rooms[R]) Channel[C, R, Anonymous]
- type ChannelID
- type ChannelInfo
- type ChannelSnapshot
- type ClientDescription
- type ClientLimits
- type ClientRate
- type ClusterBackend
- type ClusterConfig
- type ClusterKey
- type ClusterLimits
- type ClusterMembership
- type Code
- type Config
- type Connection
- type ConnectionID
- type ConnectionName
- type Connections
- type Diagnostics
- type Direction
- type EventID
- type EventInfo
- type EventRegistration
- type Hooks
- type Hub
- func (h *Hub) Done() <-chan struct{}
- func (h *Hub) Probe(ctx context.Context) error
- func (h *Hub) Ready(ctx context.Context) (string, error)
- func (h *Hub) Registry() *Registry
- func (h *Hub) ServeHTTP(w stdhttp.ResponseWriter, r *stdhttp.Request)
- func (h *Hub) Snapshot() Snapshot
- func (h *Hub) Start(ctx context.Context) error
- func (h *Hub) Stop(ctx context.Context) error
- type Incoming
- func (e Incoming[C, R, S, P]) AcknowledgeAccepted() Incoming[C, R, S, P]
- func (e Incoming[C, R, S, P]) Authorize(check func(context.Context, MessageContext[R, S], P) error) Incoming[C, R, S, P]
- func (e Incoming[C, R, S, P]) Handle(handler func(context.Context, MessageContext[R, S], P) error) EventRegistration[C]
- func (e Incoming[C, R, S, P]) Relay(outgoing Outgoing[C, P]) EventRegistration[C]
- func (e Incoming[C, R, S, P]) RelayToOthers(outgoing Outgoing[C, P]) EventRegistration[C]
- type Instance
- type InstanceID
- type LeaveContext
- type Member
- type MemberFrame
- type MemberID
- type MessageContext
- type MessageID
- type NamedConnection
- type Option
- type Outgoing
- type PresenceChannel
- type PresenceSnapshot
- type ProtocolActions
- type ProtocolInfo
- type ProtocolResponses
- type Publication
- type PublishOption
- type Publisher
- type PublisherSource
- type Registration
- type Registry
- type ReplayConfig
- type Request
- type RequestID
- type Response
- type ResponseType
- type Rooms
- type Scope
- type ServerConfig
- type Snapshot
- type SubjectReference
- type Target
- type TicketRedeemer
Constants ¶
const KeyFeature keyspace.Feature = "websocket"
KeyFeature is the physical address family owned by this adapter.
const MaxChannelEvents = 256
const MaxChannels = 256
const MaxReplayMessages = 1024
const MaxRoomBytes = 512
const MaxTicketRedeemers = auth.MaxCredentials
MaxTicketRedeemers bounds the guards one hub accepts tickets for.
const ProtocolVersion = 1
const Subprotocol = "foundry.v1"
const TicketSubprotocolPrefix = "foundry.ticket."
TicketSubprotocolPrefix marks a single-use handshake ticket offered as one extra Sec-WebSocket-Protocol entry next to Subprotocol. The hub never selects or echoes it; only Subprotocol is negotiated.
Variables ¶
var MembershipConflict = safeerror.Sentinel(fault.Conflict, "WebSocket connection holds a different membership for this scope")
MembershipConflict reports that the authority still holds a different membership of this connection for the scope, typically left behind by a leave that failed transiently. The hub releases that record and joins again; it is a per-operation failure, never a namespace policy conflict.
var PolicyConflict = safeerror.Sentinel(fault.Conflict, "live WebSocket cluster policy differs")
PolicyConflict reports live cluster state written under a different channel registry or authority limits in the same namespace. It is the unrecoverable configuration fault that terminates a running hub; adapters return it (or wrap it) for such state. Other operation failures are treated as transient.
Functions ¶
func DeclaredModule ¶
func DeclaredModule(name foundation.ProviderID, key foundation.Key[*Hub], config ServerConfig, requires []foundation.ProviderID, construct func(foundation.Resolver) (*Hub, []foundryhttp.Middleware, error)) foundation.Module
DeclaredModule allows domain declarations to provide upgrade middleware during pure construction while preserving the same single-server kernel ownership.
func DisconnectConnection ¶
func DisconnectConnection(ctx context.Context, source PublisherSource, id ConnectionID) error
DisconnectConnection is trusted server-side management. Distributed delivery uses live pub/sub; an error never proves that no instance acted on the command.
func DisconnectSubject ¶
func DisconnectSubject[M model.Identifiable, K any](ctx context.Context, source PublisherSource, guard auth.Guard[M], reference model.Reference[M, K]) error
DisconnectSubject preserves the model/key type plus guard/provider namespace. Supply the generated reference containing the stored key, not a presentation getter. Call after committed revocation; periodic fresh authorization remains the backstop if a live disconnect message is lost.
func Module ¶
func Module(name foundation.ProviderID, key foundation.Key[*Hub], config ServerConfig, requires []foundation.ProviderID, construct func(foundation.Resolver) (*Hub, error)) foundation.Module
Module registers the WebSocket kernel and an injectable Hub. HTTP middleware wraps the exact upgrade route; trusted proxy/public-origin policy can therefore run before origin validation. List providers owning borrowed services in requires.
func NotPublished ¶
NotPublished reports that a Publish or Broadcast error happened before any live publication, fan-out or replay history was attempted (validation, local admission, a closing hub or payload encoding), so retrying cannot produce a duplicate. Other errors may follow partial distributed publication.
func Route ¶
func Route(hub *Hub, path string, middleware ...foundryhttp.Middleware) foundryhttp.RouteRegistration
Route serves this Hub's upgrades on an application router at an exact static path. Middleware wraps only the upgrade; router-wide middleware also applies.
func SharedModule ¶
func SharedModule(name foundation.ProviderID, key foundation.Key[*Hub], server foundation.Key[*foundryhttp.Server], requires []foundation.ProviderID, construct func(foundation.Resolver) (*Hub, error)) foundation.Module
SharedModule constructs a Hub served by an existing HTTP kernel instead of a dedicated listener: mount Route on that kernel's router; Ready reports that server's address. Boot confirms the distributed subscription before the listener starts; the hub drains its sockets when the application lifetime ends (after any stop delay), inside the HTTP shutdown grace. Upgrade sockets hold HTTP request and connection capacity for their lifetime, so size those limits for realtime load.
func ValidateClusterConnection ¶
func ValidateClusterConnection(ctx context.Context, key ClusterKey, instance InstanceID, connection ConnectionID) error
func ValidateClusterOperation ¶
func ValidateClusterOperation(ctx context.Context, key ClusterKey) error
Types ¶
type BackendConnection ¶
type BackendConnection struct {
// contains filtered or unexported fields
}
BackendConnection owns immutable cluster selection, borrowing its authority adapter. It registers no kernel or socket. A local connection is explicitly process-local.
func NewClusterConnection ¶
func NewClusterConnection(backend ClusterBackend, cluster ClusterConfig) (*BackendConnection, error)
func NewLocalConnection ¶
func NewLocalConnection() *BackendConnection
func (*BackendConnection) Local ¶
func (c *BackendConnection) Local() bool
Local reports a process-local connection that cannot publish across processes.
func (*BackendConnection) NewHub ¶
func (c *BackendConnection) NewHub(registry *Registry, authentication *foundryhttp.Authentication, config Config, opts ...Option) (*Hub, error)
func (*BackendConnection) NewPublisher ¶
type Channel ¶
type Channel[C, R, S any] struct { // contains filtered or unexported fields }
Channel binds a nominal owner, room key and concrete authenticated subject. Use a distinct owner type per channel. Runtime declaration identity also rejects separately declared channels sharing the same owner or string name.
func OwnedRooms ¶
func OwnedRooms[C any, R comparable, S model.Identifiable](id ChannelID, rooms Rooms[R], guard auth.Guard[S], reference model.Reference[S, R]) Channel[C, R, S]
OwnedRooms permits only a concrete room equal to the authenticated model's stored key. Reuse the generated zero model's reference metadata; presentation getters never decide ownership. Channel-wide subscription is forbidden.
func Private ¶
func Private[C, R any, S model.Identifiable](id ChannelID, rooms Rooms[R], guard auth.Guard[S], authorize func(context.Context, S, Target[R]) error) Channel[C, R, S]
Private always requires explicit room/channel authorization after the fresh typed guard check. Matching a guard alone does not authorize arbitrary rooms.
func (Channel[C, R, S]) WithReplay ¶
func (c Channel[C, R, S]) WithReplay(config ReplayConfig) Channel[C, R, S]
type ChannelInfo ¶
type ChannelInfo struct {
ID ChannelID `json:"id"`
Room foundryhttp.URLScalarInfo `json:"room"`
Private bool `json:"private"`
OwnedRooms bool `json:"owned_rooms"`
Guard auth.GuardName `json:"guard,omitempty"`
Provider auth.ProviderName `json:"provider,omitempty"`
Presence *contract.Schema `json:"presence,omitempty"`
Replay ReplayConfig `json:"replay,omitzero"`
Events []EventInfo `json:"events"`
}
type ChannelSnapshot ¶
type ClientDescription ¶
type ClientDescription struct {
Protocol ProtocolInfo `json:"protocol"`
Limits ClientLimits `json:"limits"`
Channels []ChannelInfo `json:"channels"`
}
func DescribeClient ¶
func DescribeClient(registry *Registry, config Config) (ClientDescription, error)
DescribeClient uses the same registry/config supplied to New or NewDistributed. It starts no hub, opens no sockets and invokes no handlers.
type ClientLimits ¶
type ClientLimits struct {
Subscriptions int `json:"subscriptions"`
FrameBytes int `json:"frame_bytes"`
PresenceMembers int `json:"presence_members"`
MemberBytes int `json:"member_bytes"`
DeduplicationEntries int `json:"deduplication_entries"`
OperationMilliseconds int64 `json:"operation_ms"`
// InboundQueue bounds a client's in-flight operations; MessageRate is the
// per-connection frame budget. Generated clients respect both.
InboundQueue int `json:"inbound_queue"`
MessageRate ClientRate `json:"message_rate"`
Payload contract.JSONLimits `json:"payload"`
}
ClientLimits exports the actual configured public budgets. Time is expressed in milliseconds rounded up; server policy/credential material remains private.
func (ClientLimits) Validate ¶
func (limits ClientLimits) Validate() error
Validate reuses runtime configuration validation with the smallest private capacity scope. Public metadata cannot infer a server's private connection counts, origins or credential refresh policy.
type ClientRate ¶
type ClientRate struct {
Requests uint32 `json:"requests"`
WindowMilliseconds int64 `json:"window_ms"`
}
ClientRate is a frame budget of Requests per WindowMilliseconds.
type ClusterBackend ¶
type ClusterBackend interface {
pubsub.Backend
WebSocketCheck(context.Context, ClusterKey) error
WebSocketOpen(context.Context, ClusterKey, InstanceID, ConnectionID) error
WebSocketTouch(context.Context, ClusterKey, InstanceID, ConnectionID) error
WebSocketJoin(context.Context, ClusterKey, InstanceID, ConnectionID, ClusterMembership) (PresenceSnapshot, error)
WebSocketLeave(context.Context, ClusterKey, InstanceID, ConnectionID, Scope) error
WebSocketClose(context.Context, ClusterKey, InstanceID, ConnectionID) error
WebSocketMembers(context.Context, ClusterKey, Scope) (PresenceSnapshot, error)
WebSocketAppend(context.Context, ClusterKey, ChannelID, ReplayConfig, []byte) error
WebSocketHistory(context.Context, ClusterKey, ChannelID, ReplayConfig) ([][]byte, error)
}
ClusterBackend borrows the existing live pub/sub transport and adds bounded atomic lease/membership/history capabilities. Mutations are attempted once; errors may represent an unknown outcome. Expired connections never resurrect. Members excludes expired leases and counts each subject's distinct connections. Close/Leave are idempotent and never delete another instance's live ownership.
type ClusterConfig ¶
type ClusterConfig struct {
Namespace keyspace.Namespace
MaxConnections int
MaxConnectionsPerSubject int
ConnectionTTL time.Duration
OperationTimeout time.Duration
Buffer pubsub.Limits
// RetainConnectionsOnGap keeps local sockets open across a lost fan-out
// stream; events published during the gap are then missed silently. The
// default closes them with a retryable status so clients reconnect and
// request replay.
RetainConnectionsOnGap bool
// ExcludeRemoteConnections lets ExceptConnection name a connection hosted
// by another instance by carrying the exclusion in the fan-out envelope.
// Instances before this field reject such envelopes, so enable it only
// once every instance in the namespace runs a release that has it.
// Excluding a connection of this instance (RelayToOthers) never needs it.
ExcludeRemoteConnections bool
}
ClusterConfig distinguishes authority-wide quotas from Config's local limits. Transient authority failures fail only the affected operation. A lost fan-out stream is resubscribed with backoff; no mutation retry is inferred from the Redis client's ability to reconnect. Only a policy conflict terminates a Hub.
func DefaultClusterConfig ¶
func DefaultClusterConfig(namespace keyspace.Namespace) ClusterConfig
func (ClusterConfig) Validate ¶
func (c ClusterConfig) Validate() error
type ClusterKey ¶
type ClusterKey struct {
// contains filtered or unexported fields
}
func NewClusterKey ¶
func NewClusterKey(namespace keyspace.Namespace, policy string, limits ClusterLimits) (ClusterKey, error)
func (ClusterKey) Limits ¶
func (k ClusterKey) Limits() ClusterLimits
func (ClusterKey) Namespace ¶
func (k ClusterKey) Namespace() keyspace.Namespace
func (ClusterKey) Policy ¶
func (k ClusterKey) Policy() string
func (ClusterKey) String ¶
func (k ClusterKey) String() string
func (ClusterKey) Validate ¶
func (k ClusterKey) Validate() error
type ClusterLimits ¶
type ClusterLimits struct {
Connections int `json:"connections"`
ConnectionsPerSubject int `json:"connections_per_subject"`
Subscriptions int `json:"subscriptions"`
PresenceMembers int `json:"presence_members"`
MemberBytes int `json:"member_bytes"`
FrameBytes int `json:"frame_bytes"`
ConnectionTTL time.Duration `json:"connection_ttl"`
Retention time.Duration `json:"retention"`
}
ClusterLimits are immutable authority policy, shared by every process in one namespace. Metadata lives as long as connections or replay can remain live.
func (ClusterLimits) Validate ¶
func (l ClusterLimits) Validate() error
type ClusterMembership ¶
type ClusterMembership struct {
Scope Scope
Subject MemberID
Presence bool
Data json.RawMessage
}
ClusterMembership is an adapter boundary. Data contains only the channel's safe member DTO. Subject is an opaque guard/provider/stored-identity digest.
func (ClusterMembership) Validate ¶
func (m ClusterMembership) Validate(limits ClusterLimits) error
type Code ¶
type Code string
const ( Malformed Code = "malformed" UnsupportedVersion Code = "unsupported_version" UnknownChannel Code = "unknown_channel" UnknownEvent Code = "unknown_event" WrongDirection Code = "wrong_direction" Unauthenticated Code = "unauthenticated" Forbidden Code = "forbidden" NotSubscribed Code = "not_subscribed" AlreadySubscribed Code = "already_subscribed" InvalidPayload Code = "invalid_payload" CapacityExceeded Code = "capacity_exceeded" OperationFailed Code = "operation_failed" OperationTimedOut Code = "operation_timed_out" Stopping Code = "stopping" RateLimited Code = "rate_limited" // authority timeout or exhausted management capacity. Unavailable Code = "unavailable" )
type Config ¶
type Config struct {
MaxPresenceScopes int
MaxConnections int
MaxSubscriptions int
MaxFrameBytes int
// InboundQueue bounds frames read but not yet processed. Generated clients
// limit their in-flight operations to it; overflowing it closes the socket.
InboundQueue int
// OutboundQueue bounds queued frames per connection. MaxQueuedBytes bounds
// their actual bytes, including inbound frames and pending-admission
// buffers; MaxTotalQueuedBytes bounds all connections together. Exceeding
// either disconnects the slow connection instead of blocking publishers.
OutboundQueue int
MaxQueuedBytes int
MaxTotalQueuedBytes int64
// MaxOperations bounds concurrent trusted publication, presence and
// disconnect operations. Callers wait briefly, then receive fault.Overloaded.
MaxOperations int
MaxPresenceMembers int
MaxMemberBytes int
OperationTimeout time.Duration
WriteTimeout time.Duration
HeartbeatInterval time.Duration
PongTimeout time.Duration
AuthRefreshInterval time.Duration
DrainTimeout time.Duration
MaxConnectionsPerIP int
MaxConnectionsPerSubject int
DeduplicationEntries int
MessageRate ratelimit.Limit
Payload contract.JSONLimits
// AdditionalOrigins are exact HTTP(S) origins, never wildcard patterns.
// Same-origin is accepted after trusted-proxy handling. Originless clients
// must be explicitly allowed; serialized opaque/null origins never are.
AdditionalOrigins []foundryhttp.Origin `config:",json"`
AllowOriginless bool
}
func DefaultConfig ¶
func DefaultConfig() Config
type Connection ¶
type Connection struct{}
type ConnectionID ¶
type ConnectionID = model.ID[Connection]
type ConnectionName ¶
type ConnectionName string
ConnectionName selects an instance, distinct from names of other service families.
func (ConnectionName) Validate ¶
func (n ConnectionName) Validate() error
type Connections ¶
type Connections struct {
// contains filtered or unexported fields
}
Connections is immutable; default and named selection alias the same instance.
func NewConnections ¶
func NewConnections(selected ConnectionName, entries ...NamedConnection) (*Connections, error)
func (*Connections) Connection ¶
func (r *Connections) Connection(name ConnectionName) (*BackendConnection, error)
func (*Connections) Default ¶
func (r *Connections) Default() (*BackendConnection, error)
func (*Connections) DefaultName ¶
func (r *Connections) DefaultName() ConnectionName
func (*Connections) Names ¶
func (r *Connections) Names() []ConnectionName
type Diagnostics ¶
type Diagnostics struct {
Runtime Snapshot
QueuedFrames, QueuedBytes int
HeartbeatFailures, RateRejected, Revocations, ForcedDisconnects uint64
// CleanupFailures counts cluster membership releases or stream closes that
// failed; leases expire by TTL and presence is reconciled.
CleanupFailures uint64
Channels []ChannelSnapshot
}
func Diagnose ¶
func Diagnose[M any](ctx context.Context, hub *Hub, guard auth.Guard[M], authorize func(context.Context, M) error) (Diagnostics, error)
Diagnose requires both a typed authenticated guard and explicit management policy. No public HTTP endpoint is installed. Compose with the existing HTTP authentication adapter; its scope determines the administrative subject.
type EventRegistration ¶
type EventRegistration[C any] struct { // contains filtered or unexported fields }
type Hooks ¶
type Hooks[R, S any] struct { Joined func(context.Context, MessageContext[R, S]) error Left func(context.Context, LeaveContext[R]) error }
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub owns connections, bounded queues and active callbacks. Construction starts no I/O or goroutines. Keep borrowed authentication/services alive until Done. A Hub must not be copied. Stop is permanent; construct a new Hub to restart.
func New ¶
func New(registry *Registry, authentication *foundryhttp.Authentication, config Config, opts ...Option) (*Hub, error)
func NewDistributed ¶
func NewDistributed(registry *Registry, authentication *foundryhttp.Authentication, config Config, backend ClusterBackend, cluster ClusterConfig, opts ...Option) (*Hub, error)
func (*Hub) Probe ¶
Probe reports local serving health for readiness checks without I/O: nil while the hub admits connections and, when distributed, its fan-out stream is subscribed. It fails while stopping, after a terminal fault or during a stream gap/resubscription. Transient per-operation failures do not fail it.
func (*Hub) Ready ¶
Ready waits for this Hub's Module-selected listener and returns its bound address. Manually embedded HTTP handlers use their host's readiness API.
func (*Hub) ServeHTTP ¶
func (h *Hub) ServeHTTP(w stdhttp.ResponseWriter, r *stdhttp.Request)
ServeHTTP remains active until all socket loops and callbacks actually exit. Native HTTP middleware can wrap this handler. Route only the intended upgrade path to it. A manual host must call Stop; Module owns both listener and Hub.
func (*Hub) Snapshot ¶
Snapshot is local operational metadata. It contains no credentials or models.
func (*Hub) Start ¶
Start confirms the distributed subscription before upgrades may be accepted. A local Hub needs no separate startup. The wait context bounds waiting; Stop owns any setup still in progress. A failed initial subscription is terminal; later stream loss is recovered by resubscribing without stopping the hub.
func (*Hub) Stop ¶
Stop seals admission, cancels sockets and waits for actual handler/hook exit. Its context bounds waiting, not callback ownership. Calling it from one of this Hub's active callbacks is rejected instead of waiting on itself. It returns the unrecoverable cluster fault that terminated the hub, if any.
type Incoming ¶
type Incoming[C, R, S, P any] struct { // contains filtered or unexported fields }
func DefineIncoming ¶
func RawIncoming ¶
func RawIncoming[C, R, S any](channel Channel[C, R, S], id EventID) Incoming[C, R, S, json.RawMessage]
RawIncoming/RawOutgoing deliberately expose dynamic JSON. Metadata marks this escape hatch; it is excluded from claims of typed payload schemas.
func (Incoming[C, R, S, P]) AcknowledgeAccepted ¶
AcknowledgeAccepted emits an accepted frame after decoding and authorization, before invoking the handler. Completion still emits ack or a stable error.
func (Incoming[C, R, S, P]) Authorize ¶
func (e Incoming[C, R, S, P]) Authorize(check func(context.Context, MessageContext[R, S], P) error) Incoming[C, R, S, P]
Authorize adds a payload-aware policy after channel authorization and decoding. Return an authorization error on denial. A nil policy invalidates the binding.
func (Incoming[C, R, S, P]) Handle ¶
func (e Incoming[C, R, S, P]) Handle(handler func(context.Context, MessageContext[R, S], P) error) EventRegistration[C]
func (Incoming[C, R, S, P]) Relay ¶
func (e Incoming[C, R, S, P]) Relay(outgoing Outgoing[C, P]) EventRegistration[C]
Relay is an ordinary typed handler; private room and payload policies still run before the corresponding outgoing event is published.
func (Incoming[C, R, S, P]) RelayToOthers ¶
func (e Incoming[C, R, S, P]) RelayToOthers(outgoing Outgoing[C, P]) EventRegistration[C]
RelayToOthers is Relay without live delivery back to the sending connection.
type InstanceID ¶
type LeaveContext ¶
type LeaveContext[R any] struct { Publisher *Hub Connection ConnectionID Target Target[R] Subject SubjectReference }
LeaveContext deliberately contains no retained model or authentication scope.
type MemberFrame ¶
type MemberFrame struct {
ID MemberID `json:"id"`
Data json.RawMessage `json:"data"`
Connections int `json:"connections"`
}
type MessageContext ¶
type MessageContext[R, S any] struct { Publisher *Hub Connection ConnectionID Target Target[R] Subject S }
type MessageID ¶
type MessageID = model.ID[Publication]
func Broadcast ¶
func Broadcast[C, R, S, P any](ctx context.Context, source PublisherSource, channel Channel[C, R, S], event Outgoing[C, P], payload P, options ...PublishOption) (MessageID, error)
Broadcast reaches every connection subscribed anywhere in this channel, once per connection even when it has overlapping whole-channel/room subscriptions.
func Publish ¶
func Publish[C, R, S, P any](ctx context.Context, source PublisherSource, channel Channel[C, R, S], room R, event Outgoing[C, P], payload P, options ...PublishOption) (MessageID, error)
Publish reaches subscriptions for exactly this room. Whole-channel subscribers do not implicitly subscribe to every private room. Success means local queue admission or distributed live publication was attempted, not recipient delivery. Slow peers disconnect. Distributed errors may follow partial publication/history.
type NamedConnection ¶
type NamedConnection struct {
Name ConnectionName
Value *BackendConnection
}
NamedConnection borrows one existing instance. Selection never creates another owner.
type Option ¶
type Option func(*options) error
Option configures optional hub collaborators at construction.
func WithLogger ¶
WithLogger reports cluster degradation, stream loss/recovery and terminal faults with safe structured diagnostics. Without it the hub does not log.
func WithTickets ¶
func WithTickets(redeemers ...TicketRedeemer) Option
WithTickets lets browsers, which cannot send an Authorization header on a WebSocket, authenticate with a ticket from TicketSubprotocolPrefix. The hub redeems it after the origin check, trying redeemers in order; the first that knows it authenticates that guard's channels for the connection. Every later auth scope re-verifies the bound credential, so revocation still disconnects.
type Outgoing ¶
type Outgoing[C, P any] struct { // contains filtered or unexported fields }
func DefineOutgoing ¶
func RawOutgoing ¶
func (Outgoing[C, P]) Registration ¶
func (e Outgoing[C, P]) Registration() EventRegistration[C]
type PresenceChannel ¶
type PresenceChannel[C, R, S, D any] struct { // contains filtered or unexported fields }
func WithPresence ¶
func WithPresence[C, R any, S model.Identifiable, D any](channel Channel[C, R, S], payload contract.JSON[D], member func(context.Context, S) (D, error)) PresenceChannel[C, R, S, D]
WithPresence requires an explicit safe DTO descriptor and authenticated model. contract.JSON rejects persistence models. Only the DTO bytes and immutable subject reference survive the operation; full models are never stored.
func (PresenceChannel[C, R, S, D]) Channel ¶
func (p PresenceChannel[C, R, S, D]) Channel() Channel[C, R, S]
func (PresenceChannel[C, R, S, D]) Count ¶
func (p PresenceChannel[C, R, S, D]) Count(ctx context.Context, hub *Hub, target Target[R]) (int, error)
Count uses the same authoritative safe-member snapshot as Members. It counts subjects, not tabs, and reports errors instead of a fallback local cluster count.
func (PresenceChannel[C, R, S, D]) Members ¶
func (p PresenceChannel[C, R, S, D]) Members(ctx context.Context, hub *Hub, target Target[R]) ([]Member[D], error)
Members is a trusted server-side inspection API, not an authorization check. Local results describe this Hub only; distributed results use its authority.
type PresenceSnapshot ¶
type PresenceSnapshot struct {
Revision uint64 `json:"revision"`
Members []MemberFrame `json:"members"`
}
PresenceSnapshot carries authority order so concurrent refreshes cannot replace newer counts with an older result. Revisions are bounded positive Redis integers.
type ProtocolActions ¶
type ProtocolInfo ¶
type ProtocolInfo struct {
Version int `json:"version"`
Subprotocol string `json:"subprotocol"`
TicketPrefix string `json:"ticket_subprotocol_prefix"`
MaxRoomBytes int `json:"max_room_bytes"`
MaxReplayMessages int `json:"max_replay_messages"`
Actions ProtocolActions `json:"actions"`
Responses ProtocolResponses `json:"responses"`
Codes []Code `json:"codes"`
}
ProtocolInfo is the versioned transport vocabulary used by runtime and client exporters. It contains no live connection state or credentials.
func ProtocolDescription ¶
func ProtocolDescription() ProtocolInfo
type ProtocolResponses ¶
type ProtocolResponses struct {
Subscribed ResponseType `json:"subscribed"`
Unsubscribed ResponseType `json:"unsubscribed"`
Acknowledged ResponseType `json:"acknowledged"`
Accepted ResponseType `json:"accepted"`
Error ResponseType `json:"error"`
Event ResponseType `json:"event"`
PresenceJoined ResponseType `json:"presence_joined"`
PresenceLeft ResponseType `json:"presence_left"`
PresenceUpdated ResponseType `json:"presence_updated"`
}
type Publication ¶
type Publication struct{}
type PublishOption ¶
type PublishOption func(*publishOptions)
PublishOption adjusts one publication's live delivery.
func ExceptConnection ¶
func ExceptConnection(id ConnectionID) PublishOption
ExceptConnection skips live delivery to one connection, typically the sender of a relayed message ("to others"). Recent replay still includes the event.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher owns publication operations without a socket listener or auth scopes. Its constructor binds the distributed backend and immutable registry.
func NewPublisher ¶
func NewPublisher(registry *Registry, config Config, backend ClusterBackend, cluster ClusterConfig, opts ...Option) (*Publisher, error)
NewPublisher constructs trusted server-side publication without authentication scopes or socket ownership. Use the exact same channel and cluster policies as the socket servers. Stop publication before closing the borrowed Redis client.
type PublisherSource ¶
type PublisherSource interface {
// contains filtered or unexported methods
}
PublisherSource is implemented by a socket Hub and a server-side Publisher. The same typed declaration/room/payload boundary serves HTTP, jobs and sockets.
type Registration ¶
type Registration struct {
// contains filtered or unexported fields
}
func Register ¶
func Register[C, R, S any](channel Channel[C, R, S], events ...EventRegistration[C]) Registration
Register is the only erasure boundary. Event ownership is checked by the Go compiler and by nominal declaration identity when the registry is built.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
func NewRegistry ¶
func NewRegistry(registrations ...Registration) (*Registry, error)
func (*Registry) Channels ¶
func (r *Registry) Channels() []ChannelInfo
Channels returns deterministic owned contract snapshots. Contract exporters consume these descriptors; no second payload or room schema is maintained.
type ReplayConfig ¶
type ReplayConfig struct {
Messages int `json:"messages"`
Bytes int `json:"bytes"`
TTL time.Duration `json:"ttl"`
}
ReplayConfig retains a recent per-channel bucket. Room replay filters that bucket; traffic in other rooms can evict older messages. Zero disables history. TTL is a duration in nanoseconds in Go/JSON metadata, not a reconnect cursor.
func (ReplayConfig) Validate ¶
func (c ReplayConfig) Validate() error
type Request ¶
type Request struct {
Version int `json:"v"`
Action Action `json:"action"`
ID RequestID `json:"id"`
Channel ChannelID `json:"channel"`
Room *string `json:"room,omitempty"`
Event EventID `json:"event,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
Replay *int `json:"replay,omitempty"`
}
Request is the wire boundary only. Normal domain handlers receive typed MessageContext and payload values; no JSON map inspection is necessary.
func DecodeRequest ¶
DecodeRequest enforces the exact versioned envelope in one strict pass: duplicate/unknown members, invalid UTF-8 and unpaired surrogates are rejected. Callers must bound frame buffering first. A malformed request still reports its ID when one was decoded and valid, so errors can correlate.
type Response ¶
type Response struct {
Version int `json:"v"`
Type ResponseType `json:"type"`
ID RequestID `json:"id,omitempty"`
Channel ChannelID `json:"channel,omitempty"`
Room *string `json:"room,omitempty"`
Event EventID `json:"event,omitempty"`
MessageID MessageID `json:"message_id,omitzero"`
Payload json.RawMessage `json:"payload,omitempty"`
Code Code `json:"code,omitempty"`
Members []MemberFrame `json:"members,omitempty"`
Member *MemberFrame `json:"member,omitempty"`
Replayed bool `json:"replayed,omitempty"`
}
type ResponseType ¶
type ResponseType string
const ( Subscribed ResponseType = "subscribed" Unsubscribed ResponseType = "unsubscribed" Acknowledged ResponseType = "ack" Accepted ResponseType = "accepted" ErrorResponse ResponseType = "error" EventResponse ResponseType = "event" PresenceJoined ResponseType = "presence_joined" PresenceLeft ResponseType = "presence_left" PresenceUpdated ResponseType = "presence_updated" )
type Rooms ¶
type Rooms[R any] struct { // contains filtered or unexported fields }
func DefineRooms ¶
func DefineRooms[R any](codec foundryhttp.PathCodec[R]) Rooms[R]
DefineRooms reuses the existing text codec and its same-value metadata.
func (Rooms[R]) Description ¶
func (r Rooms[R]) Description() (foundryhttp.URLScalarInfo, error)
type Scope ¶
type Scope struct {
Channel ChannelID `json:"channel"`
Room string `json:"room"`
HasRoom bool `json:"has_room"`
}
Scope distinguishes whole-channel membership from an exact room. Adapters use its opaque Hash; applications retain Channel/Target rather than constructing it.
type ServerConfig ¶
type ServerConfig struct {
HTTP foundryhttp.ServerConfig
Path string
Middleware []foundryhttp.Middleware
}
func DefaultServerConfig ¶
func DefaultServerConfig() ServerConfig
func (ServerConfig) Validate ¶
func (c ServerConfig) Validate() error
type Snapshot ¶
type Snapshot struct {
Distributed, Degraded bool
Streaming bool
BackgroundTasks int
Stopping bool
Connections int
Subscriptions int
ActiveOperations int
QueuedBytes int64
Accepted, Rejected, Publications, SlowConsumers, Failures uint64
Gaps, Resubscriptions, DroppedEnvelopes, OperationOverloads uint64
}
type SubjectReference ¶
type SubjectReference struct {
Guard auth.GuardName
Provider auth.ProviderName
Identity model.Identity
}
SubjectReference is immutable authenticated attribution for cleanup and management. It is not a credential and cannot authenticate another operation.
type Target ¶
Target distinguishes a whole-channel subscription from a concrete room. The same value type is retained through authorization and incoming handlers.
func WholeChannel ¶
type TicketRedeemer ¶
type TicketRedeemer interface {
TicketSource() auth.CredentialName
RedeemTicket(context.Context, secret.String) (value.Optional[auth.BoundCredential], error)
}
TicketRedeemer exchanges a single-use handshake ticket for a bound credential of the guard whose credential source it names. token.Tokens implements it. An unknown, expired or already redeemed ticket is omitted, not an error.
Source Files
¶
- backend_connection.go
- channel.go
- client_metadata.go
- cluster_backend.go
- cluster_config.go
- cluster_membership.go
- cluster_presence.go
- cluster_runtime.go
- cluster_wire.go
- code_diagnostics.go
- config.go
- connection.go
- connection_limits.go
- diagnostics.go
- disconnect.go
- drain.go
- event.go
- heartbeat.go
- hub.go
- module.go
- named.go
- operation.go
- pending_subscription.go
- presence.go
- protocol.go
- publish.go
- publisher_source.go
- registry.go
- replay.go
- room.go
- subscription.go
- ticket.go
- upgrade.go
- wire_decode.go