broker

Documentation

Index

• func AuthMiddleware(auth Authenticator, redirectEnabled bool, redirectURL string, ...) http.Handler
• type Authenticator
• type Config
• type Endpoint
• type EndpointRegistry
  • func NewEndpointRegistry() *EndpointRegistry
  • func (r *EndpointRegistry) AddConsumer(endpointName string, session *brokerSession)
  • func (r *EndpointRegistry) ConsumerCount(endpointName string) int
  • func (r *EndpointRegistry) Forget(sessionID, role, endpoint string)
  • func (r *EndpointRegistry) GetEndpoint(name string) (*Endpoint, bool)
  • func (r *EndpointRegistry) GetOrCreate(name string) *Endpoint
  • func (r *EndpointRegistry) GetProviderYamux(endpointName string) (*yamux.Session, bool)
  • func (r *EndpointRegistry) HasProvider(endpointName string) bool
  • func (r *EndpointRegistry) NotifyProviderArrived(endpointName string)
  • func (r *EndpointRegistry) RegisterConsumerYamux(endpointName, sessionID string, ys *yamux.Session)
  • func (r *EndpointRegistry) RemoveProvider(endpointName string)
  • func (r *EndpointRegistry) SetProvider(endpointName string, session *brokerSession, yamuxSess *yamux.Session) error
  • func (r *EndpointRegistry) UnregisterConsumerYamux(endpointName, sessionID string)
  • func (r *EndpointRegistry) WaitForProvider(endpointName string, done <-chan struct{}) (*yamux.Session, bool)
• type NoopAuthenticator
  • func (a *NoopAuthenticator) Authenticate(r *http.Request) error
• type Relay
  • func NewRelay(registry *EndpointRegistry, logger *zap.Logger) *Relay
  • func (r *Relay) HandleConsumer(session *brokerSession)
  • func (r *Relay) HandleProvider(session *brokerSession)
• type Server
  • func NewServer(config Config, logger *zap.Logger) *Server
  • func (s *Server) Start() error
  • func (s *Server) Stop(ctx context.Context) error
• type TokenAuthenticator
  • func NewTokenAuthenticator(token string) *TokenAuthenticator
  • func (a *TokenAuthenticator) Authenticate(r *http.Request) error

Functions

func AuthMiddleware

func AuthMiddleware(auth Authenticator, redirectEnabled bool, redirectURL string, next http.Handler) http.Handler

AuthMiddleware wraps an http.Handler with authentication. If authentication fails:

• When redirectEnabled is true and redirectURL is set: redirects to the configured URL with 302
• Otherwise: returns a 401 Unauthorized response

Types

type Authenticator

type Authenticator interface {
	Authenticate(r *http.Request) error
}

Authenticator is the interface for request authentication.

type Config

type Config struct {
	ListenAddr                  string
	TLSCertFile                 string
	TLSKeyFile                  string
	UseTLS                      bool
	PollTimeout                 time.Duration // how long to hold poll before empty response (default 30s)
	SessionTimeout              time.Duration // inactive session cleanup interval (default 60s; must be >= 2x PollTimeout)
	CoalesceWindow              time.Duration // how long to let a poll response accumulate more data once any is available (default: pollmux.DefaultCoalesceWindow)
	PollBufferSize              int           // bytes per long-poll response (default: pollmux.DefaultPollBufferSize, 256KiB)
	MaxSendBytes                int           // cap on a single request body (default: pollmux.DefaultMaxSendBytes, 1MiB)
	HighWaterWarn               int           // log once when a session buffers this many bytes; 0 disables
	PollMode                    string        // "" defaults to stream mode; "batch" forces the older discrete poll mode; any other value panics at startup (see resolvePollMode)
	EnableWebSocket             bool          // whether to offer pollmux's WebSocket transport when a client asks for it (default: false)
	AuthEnabled                 bool          // whether authentication is enabled
	AuthToken                   string        // authentication token (used when AuthEnabled is true)
	StatusEndpointEnabled       bool          // whether to expose GET /status endpoint (default: false)
	UnauthorizedRedirectEnabled bool          // whether to redirect unauthorized requests instead of returning 401/404
	UnauthorizedRedirectURL     string        // redirect target URL for unauthorized requests
	Version                     string        // broker version
}

Config holds broker server configuration.

type Endpoint

type Endpoint struct {
	Name             string
	ProviderSession  *brokerSession            // nil if no provider connected
	ProviderYamux    *yamux.Session            // yamux client toward provider
	ConsumerSessions map[string]*brokerSession // session ID -> consumer session
	// contains filtered or unexported fields
}

Endpoint represents a named proxy endpoint with one provider and N consumers.

type EndpointRegistry

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

EndpointRegistry manages all named endpoints. Session lifecycle (creation, lookup, expiry) is entirely owned by pollmux.SessionStore; this registry is a derived topology view — which endpoint has a provider, which consumers are attached — updated from the connect/disconnect hooks.

func NewEndpointRegistry

func NewEndpointRegistry() *EndpointRegistry

NewEndpointRegistry creates a new empty EndpointRegistry.

func (*EndpointRegistry) AddConsumer

func (r *EndpointRegistry) AddConsumer(endpointName string, session *brokerSession)

AddConsumer registers a consumer session.

func (*EndpointRegistry) ConsumerCount

func (r *EndpointRegistry) ConsumerCount(endpointName string) int

ConsumerCount returns the number of active consumers for an endpoint.

func (*EndpointRegistry) Forget

func (r *EndpointRegistry) Forget(sessionID, role, endpoint string)

Forget removes a session's topology bookkeeping (consumer or provider) for the given role and endpoint. Called from Server.onDisconnect once pollmux has already removed the session from its own SessionStore — this only cleans up the derived registry view.

func (*EndpointRegistry) GetEndpoint

func (r *EndpointRegistry) GetEndpoint(name string) (*Endpoint, bool)

GetEndpoint retrieves an endpoint by name.

func (*EndpointRegistry) GetOrCreate

func (r *EndpointRegistry) GetOrCreate(name string) *Endpoint

GetOrCreate returns the endpoint with the given name, creating it if needed.

func (*EndpointRegistry) GetProviderYamux

func (r *EndpointRegistry) GetProviderYamux(endpointName string) (*yamux.Session, bool)

GetProviderYamux returns the provider yamux session for an endpoint.

func (*EndpointRegistry) HasProvider

func (r *EndpointRegistry) HasProvider(endpointName string) bool

HasProvider returns true if the endpoint has an active provider.

func (*EndpointRegistry) NotifyProviderArrived

func (r *EndpointRegistry) NotifyProviderArrived(endpointName string)

NotifyProviderArrived broadcasts on the endpoint's cond so that any goroutines blocked in WaitForProvider wake up and retry.

func (*EndpointRegistry) RegisterConsumerYamux

func (r *EndpointRegistry) RegisterConsumerYamux(
	endpointName, sessionID string,
	ys *yamux.Session,
)

RegisterConsumerYamux stores the consumer's yamux session so it can be closed when the provider disconnects.

func (*EndpointRegistry) RemoveProvider

func (r *EndpointRegistry) RemoveProvider(endpointName string)

RemoveProvider removes the provider from an endpoint and closes all consumer yamux sessions so that consumers detect the disconnection immediately. The consumerYamuxSessions map is cleared so stale entries don't accumulate.

func (*EndpointRegistry) SetProvider

func (r *EndpointRegistry) SetProvider(
	endpointName string,
	session *brokerSession,
	yamuxSess *yamux.Session,
) error

SetProvider registers a provider session for an endpoint. Returns error if a provider is already registered.

func (*EndpointRegistry) UnregisterConsumerYamux

func (r *EndpointRegistry) UnregisterConsumerYamux(endpointName, sessionID string)

UnregisterConsumerYamux removes a consumer's yamux session from the endpoint.

func (*EndpointRegistry) WaitForProvider

func (r *EndpointRegistry) WaitForProvider(
	endpointName string,
	done <-chan struct{},
) (*yamux.Session, bool)

WaitForProvider returns the provider yamux session for an endpoint, blocking until one appears or the done channel is closed (consumer yamux session ended).

Unlike a fixed timeout, this waits indefinitely for the provider — the caller passes the yamux session's CloseChan() as done so that if the consumer disconnects, the wait is cancelled immediately rather than leaking a goroutine.

Returns (session, true) when a provider is available. Returns (nil, false) if done is closed before a provider arrives.

type NoopAuthenticator

type NoopAuthenticator struct{}

NoopAuthenticator allows all requests through.

func (*NoopAuthenticator) Authenticate

func (a *NoopAuthenticator) Authenticate(r *http.Request) error

Authenticate always returns nil, allowing all requests.

type Relay

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

Relay manages yamux sessions and bridges streams between consumers and providers.

func NewRelay

func NewRelay(registry *EndpointRegistry, logger *zap.Logger) *Relay

NewRelay creates a new Relay with the given registry and logger.

func (*Relay) HandleConsumer

func (r *Relay) HandleConsumer(session *brokerSession)

HandleConsumer sets up yamux on the consumer session and starts accepting streams. Blocks until the consumer disconnects.

func (*Relay) HandleProvider

func (r *Relay) HandleProvider(session *brokerSession)

HandleProvider sets up yamux on the provider session and registers it. Blocks until the provider disconnects, then closes all consumer yamux sessions for this endpoint so consumers detect the failure immediately.

type Server

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

Server is the broker HTTP server.

func NewServer

func NewServer(config Config, logger *zap.Logger) *Server

NewServer creates a new broker Server.

func (*Server) Start

func (s *Server) Start() error

Start starts the HTTP server and the session sweeper. Blocks until the server stops.

func (*Server) Stop

func (s *Server) Stop(ctx context.Context) error

Stop gracefully stops the server. Safe to call multiple times.

Shutdown sequence:

1. Stop the HTTP server (no new requests accepted; in-flight requests drain).
2. Close every live session so relay goroutines exit cleanly and connected consumers/providers detect the closure and reconnect.
3. Stop the sweeper. It returns only once its goroutine has exited, so no further OnDisconnect calls arrive after Stop returns.

type TokenAuthenticator

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

TokenAuthenticator validates Bearer token from Authorization header.

func NewTokenAuthenticator

func NewTokenAuthenticator(token string) *TokenAuthenticator

NewTokenAuthenticator creates a new TokenAuthenticator with the given token.

func (*TokenAuthenticator) Authenticate

func (a *TokenAuthenticator) Authenticate(r *http.Request) error

Authenticate checks if the request contains a valid Bearer token. Expected format: Authorization: Bearer <token>