gateway

package module

Go to main page

Versions in this module

v0

v0.2.0

Jul 13, 2026

Changes in this version

+ const SSEGapEventName

+ var ErrConfirmationTimeout = errors.New("gateway: cross-node delivery confirmation timed out")

+ var ErrHistoryGap = errors.New("gateway: sse history gap, the requested Last-Event-ID is no longer retained")

+ var ErrOfflineFallbackOverloaded = errors.New("gateway: offline fallback dropped, concurrency limit reached")

+ var ErrOriginNotAllowed = errors.New("gateway: websocket origin not allowed")

+ var ErrOwnerHeld = errors.New("gateway: connection owner lease is held by another node")

+ var ErrOwnerLeaseUnsupported = errors.New("gateway: WithOwnerLease requires a LinearizableFencingCoordinator")

+ var ErrPayloadTooLarge = errors.New("gateway: inbound payload exceeds the configured read limit")

+ var ErrPresenceWatchUnsupported = errors.New("gateway: presence backend does not support Watch")

+ var ErrRateLimited = errors.New("gateway: inbound message rate limit exceeded")

+ var ErrRegistryClosed = errors.New("gateway: registry is closed")

+ var ErrSSESharedHistoryRequiresGenerationalHistory = errors.New("gateway: shared sse history requires GenerationalHistory")

+ var ErrSSESharedHistoryRequiresOwnerLease = errors.New("gateway: shared sse history requires WithOwnerLease")

+ var ErrStaleGeneration = errors.New(...)

+ var ErrStaleOwner = errors.New(...)

+ var ErrTakeoverTimeout = errors.New("gateway: cross-node connection takeover timed out")

+ func DecodeOutboxEnvelope(data []byte) (msgID string, seq uint64, payload []byte, err error)

+ func DecodeOutboxTextEnvelope(data []byte) (msgID string, seq uint64, payload []byte, err error)

+ func EncodeOutboxEnvelope(msgID string, seq uint64, payload []byte) ([]byte, error)

+ func EncodeOutboxTextEnvelope(msgID string, seq uint64, payload []byte) ([]byte, error)

+ func ReplayTailSeq(msgs []PersistedMessage) uint64

+ type BackpressurePolicy int

+ const BackpressureClose

+ const BackpressureDrop

+ type BroadcastOption func(*broadcastConfig)

+ func WithExclude(ids ...string) BroadcastOption

+ type CASCoordinator interface

+ CompareAndSwap func(ctx context.Context, key string, expected, newValue []byte, ttl time.Duration) (bool, error)

+ type ConnHandle struct

+ func (h *ConnHandle) UnregisterHandle(ctx context.Context) error

+ type ConnInfo struct

+ Group string

+ ID string

+ Meta map[string]string

+ Topics []string

+ type DeliveryResult struct

+ Delivered int

+ Dropped int

+ Remote int

+ func (d DeliveryResult) None() bool

+ func (d DeliveryResult) Total() int

+ type DomainPolicy func(ctx context.Context, domain string) (bool, error)

+ type FileCertStore struct

+ func NewFileCertStore(dir string) (*FileCertStore, error)

+ func (f *FileCertStore) Delete(_ context.Context, domain string) error

+ func (f *FileCertStore) Get(_ context.Context, domain string) (*Certificate, error)

+ func (f *FileCertStore) Put(_ context.Context, cert *Certificate) error

+ type GenerationalHistory interface

+ AdvanceGeneration func(ctx context.Context, connID string, generation uint64) error

+ AppendGenerational func(ctx context.Context, connID, eventID string, payload []byte, generation uint64) (seq uint64, err error)

+ type LinearizableFencingCoordinator interface

+ LinearizableFencing func()

type ManagerOption

+ func WithDomainPolicy(p DomainPolicy) ManagerOption

+ func WithFallbackCertificate(get func() (*tls.Certificate, error)) ManagerOption

+ func WithMaxCachedCerts(n int) ManagerOption

+ func WithNegativeCacheTTL(d time.Duration) ManagerOption

type MemoryCoordinator

+ func (*MemoryCoordinator) LinearizableFencing()

+ func (m *MemoryCoordinator) CompareAndSwap(_ context.Context, key string, expected, newValue []byte, ttl time.Duration) (bool, error)

+ type MemoryOutbox struct

+ func NewMemoryOutbox() *MemoryOutbox

+ func (o *MemoryOutbox) Ack(_ context.Context, connID, msgID string, generation uint64) error

+ func (o *MemoryOutbox) AdvanceGeneration(_ context.Context, connID string, generation uint64) error

+ func (o *MemoryOutbox) Append(_ context.Context, connID string, payload []byte) (string, uint64, error)

+ func (o *MemoryOutbox) DropConn(_ context.Context, connID string) error

+ func (o *MemoryOutbox) Unacked(_ context.Context, connID string) ([]PersistedMessage, error)

+ type MemoryPresence struct

+ func NewMemoryPresence() *MemoryPresence

+ func (p *MemoryPresence) Entries(_ context.Context, group string) ([]PresenceEntry, error)

+ func (p *MemoryPresence) Join(_ context.Context, group, connID string, ttl time.Duration) error

+ func (p *MemoryPresence) JoinWithMeta(_ context.Context, group, connID string, meta map[string]string, ...) error

+ func (p *MemoryPresence) Leave(_ context.Context, group, connID string) error

+ func (p *MemoryPresence) LeaveGen(_ context.Context, group, connID string, generation uint64) error

+ func (p *MemoryPresence) Members(_ context.Context, group string) ([]string, error)

+ func (p *MemoryPresence) Online(_ context.Context, group string) (bool, error)

+ func (p *MemoryPresence) Refresh(_ context.Context, group, connID string, ttl time.Duration) error

+ func (p *MemoryPresence) RefreshGen(_ context.Context, group, connID string, generation uint64, ttl time.Duration) error

+ func (p *MemoryPresence) Watch(ctx context.Context, group string) (<-chan PresenceEvent, func(), error)

+ type MemorySSEHistory struct

+ func NewMemorySSEHistory(perConn int, opts ...MemorySSEHistoryOption) *MemorySSEHistory

+ func (h *MemorySSEHistory) AdvanceGeneration(_ context.Context, connID string, generation uint64) error

+ func (h *MemorySSEHistory) Append(_ context.Context, connID, eventID string, payload []byte) error

+ func (h *MemorySSEHistory) AppendGenerational(_ context.Context, connID, eventID string, payload []byte, generation uint64) (uint64, error)

+ func (h *MemorySSEHistory) Drop(connID string)

+ func (h *MemorySSEHistory) Len() int

+ func (h *MemorySSEHistory) Since(_ context.Context, connID, lastEventID string) ([]SSEEvent, error)

+ type MemorySSEHistoryOption func(*MemorySSEHistory)

+ func WithSSEHistoryMaxConnections(n int) MemorySSEHistoryOption

+ type Observer interface

+ BroadcastFanout func(topic string, localMembers int)

+ ConnectionRegistered func(id, group string)

+ ConnectionReplaced func(id, group string)

+ ConnectionUnregistered func(id, group string)

+ DeliveryDropped func(id, group string)

+ DeliveryFailed func(id string, err error)

+ type OfflineChannel interface

+ Deliver func(ctx context.Context, group string, payload []byte) error

+ type OfflineObserver interface

+ OfflineFallback func(group string, err error)

+ type OfflineOption func(*boundedOffline)

+ func WithOfflineMaxConcurrent(n int) OfflineOption

+ func WithOfflineTimeout(d time.Duration) OfflineOption

+ type Outbox interface

+ Ack func(ctx context.Context, connID, msgID string, generation uint64) error

+ Append func(ctx context.Context, connID string, payload []byte) (msgID string, seq uint64, err error)

+ DropConn func(ctx context.Context, connID string) error

+ Unacked func(ctx context.Context, connID string) ([]PersistedMessage, error)

+ type OutboxGenerationAdvancer interface

+ AdvanceGeneration func(ctx context.Context, connID string, generation uint64) error

+ type PersistedMessage struct

+ ID string

+ Payload []byte

+ Seq uint64

+ type Presence interface

+ Join func(ctx context.Context, group, connID string, ttl time.Duration) error

+ Leave func(ctx context.Context, group, connID string) error

+ Members func(ctx context.Context, group string) ([]string, error)

+ Online func(ctx context.Context, group string) (bool, error)

+ Refresh func(ctx context.Context, group, connID string, ttl time.Duration) error

+ type PresenceDirectory interface

+ Entries func(ctx context.Context, group string) ([]PresenceEntry, error)

+ type PresenceEntry struct

+ ConnID string

+ Meta map[string]string

+ type PresenceEvent struct

+ ConnID string

+ Group string

+ Kind PresenceEventKind

+ type PresenceEventKind int

+ const PresenceJoin

+ const PresenceLeave

+ type PresenceFencer interface

+ LeaveGen func(ctx context.Context, group, connID string, generation uint64) error

+ RefreshGen func(ctx context.Context, group, connID string, generation uint64, ...) error

+ type PresenceMetaJoiner interface

+ JoinWithMeta func(ctx context.Context, group, connID string, meta map[string]string, ...) error

+ type PresenceWatcher interface

+ Watch func(ctx context.Context, group string) (events <-chan PresenceEvent, cancel func(), err error)

+ type RegisterOption func(*registerConfig)

+ func WithConnCloseHook(hook func(reason string)) RegisterOption

+ func WithConnGroup(group string) RegisterOption

+ func WithConnMeta(meta map[string]string) RegisterOption

+ func WithConnTopics(topics ...string) RegisterOption

+ func WithReplaceExisting() RegisterOption

type Registry

+ func (r *Registry) Ack(ctx context.Context, connID, msgID string) error

+ func (r *Registry) Close(_ context.Context) error

+ func (r *Registry) Disconnect(ctx context.Context, id, reason string) error

+ func (r *Registry) DisconnectGroup(ctx context.Context, group, reason string) (int, error)

+ func (r *Registry) GroupMembers(ctx context.Context, group string) ([]PresenceEntry, error)

+ func (r *Registry) IsOnline(ctx context.Context, group string) (bool, error)

+ func (r *Registry) LocalConnectionsOf(group string) []string

+ func (r *Registry) RegisterHandle(ctx context.Context, id string, send func([]byte) error, ...) (*ConnHandle, error)

+ func (r *Registry) SendToGroup(ctx context.Context, group string, payload []byte) (DeliveryResult, error)

+ func (r *Registry) WatchPresence(ctx context.Context, group string) (<-chan PresenceEvent, func(), error)

+ type RegistryOption func(*Registry)

+ func WithConfirmationTimeout(d time.Duration) RegistryOption

+ func WithDeliveryConfirmation() RegistryOption

+ func WithObserver(o Observer) RegistryOption

+ func WithOfflineChannel(ch OfflineChannel, opts ...OfflineOption) RegistryOption

+ func WithOutbox(o Outbox) RegistryOption

+ func WithOutboxEnvelope() RegistryOption

+ func WithOwnerLease(c Coordinator) RegistryOption

+ func WithPresence(p Presence) RegistryOption

+ func WithPresenceTTL(d time.Duration) RegistryOption

+ type ReloadingCertificate struct

+ func NewReloadingCertificate(certPath, keyPath string, interval time.Duration, logger log.Logger) (*ReloadingCertificate, error)

+ func (c *ReloadingCertificate) Get() (*tls.Certificate, error)

+ func (c *ReloadingCertificate) Start(ctx context.Context)

+ func (c *ReloadingCertificate) Stop()

+ type SSEAuthFunc func(*http.Request) (*ConnInfo, error)

+ type SSEEvent struct

+ ID string

+ Payload []byte

type SSEHandlerOption

+ func WithSSEAuth(f SSEAuthFunc) SSEHandlerOption

+ func WithSSEBackpressurePolicy(p BackpressurePolicy) SSEHandlerOption

+ func WithSSEEventName(f func(payload []byte) string) SSEHandlerOption

+ func WithSSEHistory(h SSEHistory) SSEHandlerOption

+ func WithSSEReauth(interval time.Duration, f SSEAuthFunc) SSEHandlerOption

+ func WithSSERetry(d time.Duration) SSEHandlerOption

+ func WithSSEWriteTimeout(d time.Duration) SSEHandlerOption

+ type SSEHistory interface

+ Append func(ctx context.Context, connID, eventID string, payload []byte) error

+ Since func(ctx context.Context, connID, lastEventID string) (events []SSEEvent, err error)

type ServerOption

+ func WithServerErrorLog(logger *stdlog.Logger) ServerOption

+ type SharedSSEHistory interface

+ SharedSSEHistory func()

+ type WSAuthFunc func(*http.Request) (*ConnInfo, error)

type WSHandlerOption

+ func WithWSAuth(f WSAuthFunc) WSHandlerOption

+ func WithWSBackpressurePolicy(p BackpressurePolicy) WSHandlerOption

+ func WithWSCompression(mode websocket.CompressionMode) WSHandlerOption

+ func WithWSGroupRate(perSecond float64, burst int) WSHandlerOption

+ func WithWSInboundRate(perSecond float64, burst int) WSHandlerOption

+ func WithWSInsecureSkipOriginCheck() WSHandlerOption

+ func WithWSMessageType(t websocket.MessageType) WSHandlerOption

+ func WithWSOriginPatterns(patterns ...string) WSHandlerOption

+ func WithWSPingInterval(d time.Duration) WSHandlerOption

+ func WithWSPongTimeout(d time.Duration) WSHandlerOption

+ func WithWSReadLimit(n int64) WSHandlerOption

+ func WithWSReauth(interval time.Duration, f WSAuthFunc) WSHandlerOption

+ func WithWSReplaceExisting() WSHandlerOption

+ func WithWSSubprotocols(protocols ...string) WSHandlerOption

+ func WithWSWriteTimeout(d time.Duration) WSHandlerOption

v0.1.0

Jul 12, 2026

Changes in this version

+ var ErrBackpressure = errors.New("gateway: connection send buffer full")

+ var ErrCertificateNotFound = errors.New("gateway: certificate not found")

+ var ErrConnectionClosed = errors.New("gateway: connection closed")

+ var ErrConnectionExists = errors.New("gateway: connection already registered")

+ var ErrConnectionNotFound = errors.New("gateway: connection not found")

+ var ErrDomainNotAllowed = errors.New("gateway: domain not allowed")

+ var ErrIssuanceLockExpired = errors.New("gateway: certificate issuance took longer than the issuance lock ttl")

+ var ErrIssuanceTimeout = errors.New("gateway: timed out waiting for coordinated certificate issuance")

+ var ErrLockNotAcquired = errors.New("gateway: lock not acquired")

+ var ErrNoIssuer = errors.New("gateway: no certificate issuer configured")

+ var ErrOriginPullVerificationFailed = errors.New("gateway: origin pull client certificate verification failed")

+ var ErrPubSubUnavailable = errors.New("gateway: pub/sub is not enabled on this actor system")

+ var ErrUnauthorized = errors.New("gateway: unauthorized")

+ func NewRSACertificateRequest(bits int) func(domain string) (csrPEM, keyPEM []byte, err error)

+ type AuthenticatedOriginPulls struct

+ CAPEM []byte

+ func (a *AuthenticatedOriginPulls) Apply(cfg *tls.Config) error

+ func (a *AuthenticatedOriginPulls) Verify(state tls.ConnectionState) error

+ type CertIssuer interface

+ Issue func(ctx context.Context, domain string) (*Certificate, error)

+ type CertStore interface

+ Delete func(ctx context.Context, domain string) error

+ Get func(ctx context.Context, domain string) (*Certificate, error)

+ Put func(ctx context.Context, cert *Certificate) error

+ type Certificate struct

+ CertPEM []byte

+ Domain string

+ KeyPEM []byte

+ NotAfter time.Time

+ type CloudflareOriginCAIssuer struct

+ APIToken string

+ BaseURL string

+ HTTPClient *http.Client

+ RequestCertificate func(domain string) (csrPEM, keyPEM []byte, err error)

+ RequestedValidityDays int

+ func (c *CloudflareOriginCAIssuer) Issue(ctx context.Context, domain string) (*Certificate, error)

+ type Coordinator interface

+ Get func(ctx context.Context, key string) (value []byte, ok bool, err error)

+ Put func(ctx context.Context, key string, value []byte, ttl time.Duration) error

+ TryLock func(ctx context.Context, key string, ttl time.Duration) (unlock func(context.Context) error, err error)

+ type Drainer interface

+ Drain func()

+ type Manager struct

+ func NewManager(system actor.ActorSystem, logger log.Logger, opts ...ManagerOption) *Manager

+ func (m *Manager) EnsureCertificate(ctx context.Context, domain string) (*tls.Certificate, error)

+ func (m *Manager) GetCertificate(hello *tls.ClientHelloInfo) (*tls.Certificate, error)

+ func (m *Manager) Start(ctx context.Context) error

+ func (m *Manager) Stop(ctx context.Context) error

+ func (m *Manager) TLSConfig() *tls.Config

+ type ManagerOption func(*Manager)

+ func WithAllowedDomains(domains ...string) ManagerOption

+ func WithCertIssuer(issuer CertIssuer) ManagerOption

+ func WithCertStore(store CertStore) ManagerOption

+ func WithCoordinator(coordinator Coordinator) ManagerOption

+ func WithIssuanceLockTTL(d time.Duration) ManagerOption

+ func WithRenewBefore(d time.Duration) ManagerOption

+ func WithRenewInterval(cronExpression string) ManagerOption

+ type MemoryCertStore struct

+ func NewMemoryCertStore() *MemoryCertStore

+ func (m *MemoryCertStore) Delete(_ context.Context, domain string) error

+ func (m *MemoryCertStore) Get(_ context.Context, domain string) (*Certificate, error)

+ func (m *MemoryCertStore) Put(_ context.Context, cert *Certificate) error

+ type MemoryCoordinator struct

+ func NewMemoryCoordinator() *MemoryCoordinator

+ func (m *MemoryCoordinator) Get(_ context.Context, key string) ([]byte, bool, error)

+ func (m *MemoryCoordinator) Put(_ context.Context, key string, value []byte, ttl time.Duration) error

+ func (m *MemoryCoordinator) TryLock(_ context.Context, key string, ttl time.Duration) (func(context.Context) error, error)

+ type Registry struct

+ func NewRegistry(system actor.ActorSystem, logger log.Logger) *Registry

+ func (r *Registry) Broadcast(ctx context.Context, topic string, payload []byte) error

+ func (r *Registry) Has(id string) bool

+ func (r *Registry) Join(_ context.Context, id, topic string) error

+ func (r *Registry) Leave(_ context.Context, id, topic string) error

+ func (r *Registry) Len() int

+ func (r *Registry) Register(ctx context.Context, id string, send func([]byte) error, topics ...string) error

+ func (r *Registry) SendToConnection(ctx context.Context, id string, payload []byte) error

+ func (r *Registry) Unregister(ctx context.Context, id string) error

+ type SSEHandler struct

+ func NewSSEHandler(registry *Registry, opts ...SSEHandlerOption) *SSEHandler

+ func (h *SSEHandler) Drain()

+ func (h *SSEHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ type SSEHandlerOption func(*SSEHandler)

+ func WithSSEAuthFunc(f func(*http.Request) error) SSEHandlerOption

+ func WithSSEIDFunc(f func(*http.Request) string) SSEHandlerOption

+ func WithSSEKeepAlive(d time.Duration) SSEHandlerOption

+ func WithSSELogger(logger log.Logger) SSEHandlerOption

+ func WithSSEOnConnect(f func(ctx context.Context, id string, r *http.Request)) SSEHandlerOption

+ func WithSSEOnDisconnect(f func(id string)) SSEHandlerOption

+ func WithSSESendBuffer(size int) SSEHandlerOption

+ func WithSSETopics(f func(*http.Request) []string) SSEHandlerOption

+ type Server struct

+ func NewServer(addr string, handler http.Handler, opts ...ServerOption) (*Server, error)

+ func (s *Server) ListenAndServe(ctx context.Context) error

+ func (s *Server) Shutdown(ctx context.Context) error

+ type ServerOption func(*Server)

+ func WithAuthenticatedOriginPulls(pulls *AuthenticatedOriginPulls) ServerOption

+ func WithDrainOnShutdown(drainers ...Drainer) ServerOption

+ func WithReadHeaderTimeout(d time.Duration) ServerOption

+ func WithTLSManager(manager *Manager) ServerOption

+ type StaticIssuer struct

+ func NewStaticIssuer(cert *Certificate, domains ...string) *StaticIssuer

+ func (s *StaticIssuer) Issue(_ context.Context, domain string) (*Certificate, error)

+ type WSHandler struct

+ func NewWSHandler(registry *Registry, opts ...WSHandlerOption) *WSHandler

+ func (h *WSHandler) Drain()

+ func (h *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ type WSHandlerOption func(*WSHandler)

+ func WithWSAuthFunc(f func(*http.Request) error) WSHandlerOption

+ func WithWSIDFunc(f func(*http.Request) string) WSHandlerOption

+ func WithWSLogger(logger log.Logger) WSHandlerOption

+ func WithWSOnConnect(f func(ctx context.Context, id string, r *http.Request)) WSHandlerOption

+ func WithWSOnDisconnect(f func(id string)) WSHandlerOption

+ func WithWSOnMessage(f func(ctx context.Context, id string, payload []byte)) WSHandlerOption

+ func WithWSSendBuffer(size int) WSHandlerOption

+ func WithWSTopics(f func(*http.Request) []string) WSHandlerOption

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL