transport

Documentation

Index

• type Broker
  • func New(opts ...Option) *Broker
  • func (b *Broker) Addr() net.Addr
  • func (b *Broker) AuthzEngine() *authz.Engine
  • func (b *Broker) Close() error
  • func (b *Broker) Handle(cmd protocol.Command, h Handler)
  • func (b *Broker) HandleStream(jig context.Context, conn *quic.Conn, stream *quic.Stream, clientIDStr string)
  • func (b *Broker) Listen(ctx context.Context, addr string, tlsConf *tls.Config) error
  • func (b *Broker) ReplayAAL(ctx context.Context, path string, key []byte) (int, error)
  • func (b *Broker) Router() *broker.Router
  • func (b *Broker) SetAuthzEngine(e *authz.Engine)
  • func (b *Broker) Shutdown(ctx context.Context) error
• type Handler
• type Option
  • func WithAAL(l *aal.Log) Option
  • func WithAALReplay(path string, key []byte) Option
  • func WithAuthz(e *authz.Engine) Option
  • func WithCompression(engine broker.CompressionEngine, minBatchSize int) Option
  • func WithLogger(l *slog.Logger) Option
  • func WithMaxBufSize(n int) Option
  • func WithMaxDecompressedPerStream(n int) Option
  • func WithReadBufSize(n int) Option
  • func WithRouter(r *broker.Router) Option
  • func WithTracer(t *tracing.Tracer) Option

Types

type Broker

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

Broker is a QUIC-based message broker that accepts connections and multiplexes streams. Each stream is handled by a dedicated goroutine that reads binary frames, parses them via protocol.ParseFrame, and dispatches to the registered command handler or the built-in router.

func New

func New(opts ...Option) *Broker

New creates a Broker with the given options. Call Listen to start accepting.

func (*Broker) Addr

func (b *Broker) Addr() net.Addr

Addr returns the listener's network address. Returns nil if not listening.

func (*Broker) AuthzEngine

func (b *Broker) AuthzEngine() *authz.Engine

AuthzEngine returns the active authorization engine.

func (*Broker) Close

func (b *Broker) Close() error

Close initiates a graceful shutdown of the Broker.

func (*Broker) Handle

func (b *Broker) Handle(cmd protocol.Command, h Handler)

Handle registers a handler for the given command. Must be called before Listen.

func (*Broker) HandleStream

func (b *Broker) HandleStream(jig context.Context, conn *quic.Conn, stream *quic.Stream, clientIDStr string)

HandleStream is the public, transport-agnostic entry point for feeding a single bidirectional QUIC stream through the broker's frame parser, authorization engine, AAL log, and router. It is safe to call from any goroutine; concurrency control is handled inside transport.Broker.

Use this when integrating new transports (e.g. internal/webtransport) that ride on top of QUIC bi-di streams without spawning an extra transport.Broker.

func (*Broker) Listen

func (b *Broker) Listen(ctx context.Context, addr string, tlsConf *tls.Config) error

Listen starts the QUIC listener on the given UDP address with 0-RTT support. tlsConf must contain a valid certificate and set NextProtos.

func (*Broker) ReplayAAL

func (b *Broker) ReplayAAL(ctx context.Context, path string, key []byte) (int, error)

ReplayAAL replays all historical CmdPublish frames from AAL into the router before socket bind.

func (*Broker) Router

func (b *Broker) Router() *broker.Router

Router returns the pub/sub router.

func (*Broker) SetAuthzEngine

func (b *Broker) SetAuthzEngine(e *authz.Engine)

SetAuthzEngine updates the active authorization engine.

func (*Broker) Shutdown

func (b *Broker) Shutdown(ctx context.Context) error

Shutdown gracefully drains active connections and goroutines.

type Handler

type Handler func(ctx context.Context, frame protocol.Frame) ([]byte, error)

Handler processes a parsed frame and returns an optional response payload.

type Option

type Option func(*Broker)

Option configures the Broker.

func WithAAL

func WithAAL(l *aal.Log) Option

WithAAL enables Append-Only Logging for CmdPublish frames.

func WithAALReplay

func WithAALReplay(path string, key []byte) Option

WithAALReplay enables startup state restoration from an AAL log file.

func WithAuthz

func WithAuthz(e *authz.Engine) Option

WithAuthz enables the zero-allocation ACL authorization engine.

func WithCompression

func WithCompression(engine broker.CompressionEngine, minBatchSize int) Option

WithCompression enables batch payload decompression for received frames. engine provides ZSTD decompression. minBatchSize is ignored on the receive side (all compressed batches are decompressed regardless of size).

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets a structured logger for the broker.

func WithMaxBufSize

func WithMaxBufSize(n int) Option

WithMaxBufSize sets the maximum buffer size before the processor falls back to heap allocation. Default: 64KB.

func WithMaxDecompressedPerStream

func WithMaxDecompressedPerStream(n int) Option

WithMaxDecompressedPerStream sets the maximum total decompressed bytes per stream before the broker rejects further compressed frames. Default: 16 MB. Prevents a single malicious peer from exhausting memory via compression bombs.

func WithReadBufSize

func WithReadBufSize(n int) Option

WithReadBufSize sets the initial read buffer size per stream processor. Must be >= protocol.HeaderSize (10). Default: 1024.

func WithRouter

func WithRouter(r *broker.Router) Option

WithRouter sets the pub/sub router for CmdPublish and CmdSubscribe handling.

func WithTracer

func WithTracer(t *tracing.Tracer) Option

WithTracer sets the OpenTelemetry tracer for distributed tracing.