broker

package
v1.17.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func IsWildcardTopic added in v1.3.0 

func IsWildcardTopic(topic string) bool

IsWildcardTopic returns true if topic string contains '+' or '#'.

func MatchWildcard added in v1.3.0 

func MatchWildcard(pattern, topic []byte) bool

MatchWildcard checks if topic matches pattern (supporting '+' and '#') with 0 heap allocations. MQTT wildcard rules: 

'+' matches exactly one topic level (segment between '/').
'#' matches zero or more topic levels up to the end of topic.

Types

type BackpressurePolicy added in v1.2.0 

type BackpressurePolicy uint8

BackpressurePolicy defines how queue overflow is handled for slow consumers. 

const (
	PolicyDropOldest BackpressurePolicy = iota
	PolicyDropNewest
	PolicyDisconnect
)

func ParseBackpressurePolicy added in v1.2.0 

func ParseBackpressurePolicy(s string) BackpressurePolicy

func (BackpressurePolicy) String added in v1.2.0 

func (p BackpressurePolicy) String() string

type CompressionEngine added in v1.10.0 

type CompressionEngine interface {
	Compress(src []byte) ([]byte, error)
	Decompress(src []byte, uncompressedSize int) ([]byte, error)
	ReleaseBuf(buf []byte)
}

CompressionEngine is the interface for batch payload compression.

type ConsumerGroup added in v1.13.0 

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

ConsumerGroup manages lock-free atomic round-robin dispatch among competing consumers.

type MessageRef added in v1.2.0 

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

MessageRef wraps a pooled frame buffer (*[]byte) with an atomic reference counter, offset, and optional TTL expiry. It ensures that buffers recycled into sync.Pool are never reused while queued in another subscriber's pipeline or undergoing network I/O.

Nested Reference Counting: When a batch buffer is unpacked, each sub-message gets its own MessageRef with a parent pointer pointing to the batch MessageRef. When a child ref count reaches 0, it decrements the parent. The batch buffer is returned to sync.Pool only when the parent ref count reaches 0 (all sub-messages from the batch have been delivered).

For child refs, `frame` holds a zero-copy sub-slice of the parent's buffer. The child does not own the buffer; it delegates buffer lifecycle to the parent.

func AcquireChildMessageRef added in v1.6.0 

func AcquireChildMessageRef(parent *MessageRef, frame []byte, topicOffset uint64, expiresAt int64) *MessageRef

AcquireChildMessageRef creates a child MessageRef whose frame is a zero-copy sub-slice of the parent's buffer. The child's ref starts at 1. When released, it decrements the parent's ref count. The batch buffer is returned to sync.Pool only when the parent ref count reaches 0 (all sub-messages delivered).

frame must point into the parent's underlying buffer memory. No copy is made.

func AcquireMessageRef added in v1.2.0 

func AcquireMessageRef(buf *[]byte) *MessageRef

AcquireMessageRef pulls a MessageRef from pool, wraps the frame buffer, and sets ref count to 1. The returned MessageRef has no parent (top-level owner of the buffer).

func (*MessageRef) Buf added in v1.2.0 

func (m *MessageRef) Buf() []byte

Buf returns the underlying frame buffer byte slice. For top-level refs, returns the pooled buffer. For child refs (batch sub-messages), returns the zero-copy sub-slice pointing into the parent's buffer.

func (*MessageRef) IsExpired added in v1.3.0 

func (m *MessageRef) IsExpired(nowNano int64) bool

IsExpired checks if the message expiration time has passed.

func (*MessageRef) Offset added in v1.4.0 

func (m *MessageRef) Offset() uint64

Offset returns the message topic offset.

func (*MessageRef) Parent added in v1.6.0 

func (m *MessageRef) Parent() *MessageRef

Parent returns the parent MessageRef, or nil if this is a top-level ref.

func (*MessageRef) Release added in v1.2.0 

func (m *MessageRef) Release()

Release decrements the reference counter by 1.

If this is a child ref (has parent) and ref count drops to 0, it cascades: decrements the parent's ref count and recycles itself. The parent owns the real buffer and will return it to sync.Pool only when its own ref count reaches 0.

If this is a top-level ref (no parent) and ref count drops to 0, the underlying buffer is returned to protocol.ReleaseBuffer and the MessageRef is recycled.

func (*MessageRef) Retain added in v1.2.0 

func (m *MessageRef) Retain()

Retain increments the reference counter by 1.

func (*MessageRef) SetExpiresAt added in v1.3.0 

func (m *MessageRef) SetExpiresAt(exp int64)

SetExpiresAt sets the unix nanosecond expiration timestamp.

func (*MessageRef) SetOffset added in v1.4.0 

func (m *MessageRef) SetOffset(offset uint64)

SetOffset sets the 64-bit monotonic topic offset.

type PeerForwarder added in v1.5.0 

type PeerForwarder interface {
	// Forward sends rawBuf zero-copy to all connected peers.
	// addForwardedBit=true sets the MeshForwardedBit in the wire frame.
	Forward(rawBuf []byte, addForwardedBit bool)
	ActivePeers() int
}

PeerForwarder is the interface satisfied by cluster.PeerManager. It is abstracted here to avoid an import cycle.

type Router 

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

Router implements In-Memory Direct Mesh Routing using Structure of Arrays (SoA). Each subscriber has a dedicated non-blocking bounded queue and Writer goroutine.

func NewRouter 

func NewRouter(m RouterMetrics, opts ...RouterOption) *Router

func (*Router) AckOffset added in v1.4.0 

func (r *Router) AckOffset(consumerID, topic string, offset uint64)

AckOffset updates the acknowledged consumer offset for a durable subscriber or consumer group.

func (*Router) Close added in v1.2.0 

func (r *Router) Close()

Close drains all active subscriber queues and waits for Writer goroutines to exit.

func (*Router) GetConsumerOffset added in v1.4.0 

func (r *Router) GetConsumerOffset(consumerID, topic string) uint64

GetConsumerOffset returns the acknowledged offset for consumerID on topic.

func (*Router) GetGroupOffset added in v1.13.0 

func (r *Router) GetGroupOffset(groupID, topic string) uint64

GetGroupOffset returns the acknowledged offset for a consumer group on topic.

func (*Router) GetTopicOffset added in v1.4.0 

func (r *Router) GetTopicOffset(topic string) uint64

GetTopicOffset returns the current monotonic offset for topic.

func (*Router) NackByStream added in v1.7.0 

func (r *Router) NackByStream(streamID uint32, offset uint64)

NackByStream routes a NACK offset to the subscriber's writer goroutine. Uses non-blocking send — drops if the nack channel is full (should not happen in practice).

func (*Router) Publish 

func (r *Router) Publish(ctx context.Context, frame protocol.Frame) error

Publish non-blockingly dispatches a message to all matching subscriber queues (exact & wildcard). Operates with 0 heap allocations and nano-second publisher latency.

func (*Router) PublishBatch added in v1.6.0 

func (r *Router) PublishBatch(ctx context.Context, frame protocol.Frame) error

PublishBatch unpacks a CmdPublishBatch frame and publishes each sub-message individually. Uses nested reference counting: creates a parent MessageRef for the batch buffer, then child MessageRefs for each sub-frame pointing into the parent's buffer (zero-copy).

Each sub-frame is a standard frame (Magic, Cmd, StreamID, Len, Payload) whose Payload contains the topic for routing. Sub-frames are NOT copied — they are zero-copy sub-slices of the parent batch buffer.

If compression is enabled (via WithCompression) and the batch payload exceeds the minimum size threshold, the peer-forwarded copy is compressed with ZSTD and tagged with a Compression TLV extension. Local subscribers always receive the uncompressed payload.

func (*Router) PublishFromPeer added in v1.5.0 

func (r *Router) PublishFromPeer(ctx context.Context, frame protocol.Frame) error

PublishFromPeer routes a frame received from a peer node to local subscribers only. It NEVER re-forwards to peers, preventing mesh broadcast storms.

func (*Router) PublishWithClientID added in v1.12.0 

func (r *Router) PublishWithClientID(ctx context.Context, frame protocol.Frame, clientID string) error

PublishWithClientID routes a published frame and checks rate limits for the given clientID.

func (*Router) QuotaManager added in v1.12.0 

func (r *Router) QuotaManager() *quotas.Manager

QuotaManager returns the quota manager associated with the router.

func (*Router) SetGroupOffset added in v1.13.0 

func (r *Router) SetGroupOffset(groupID, topic string, offset uint64)

SetGroupOffset explicitly sets the group offset for groupID on topic.

func (*Router) Subscribe 

func (r *Router) Subscribe(ctx context.Context, stream *quic.Stream, frame protocol.Frame) error

Subscribe registers a QUIC stream as a subscriber for the topic parsed from frame.Payload and spawns a dedicated Writer goroutine. Expected payload format: "topic:<name>[:durable:<consumerID>:<offset>]".

func (*Router) TopicOfStream added in v1.7.0 

func (r *Router) TopicOfStream(streamID uint32) (string, bool)

TopicOfStream returns the topic name for a subscriber stream, or false if not found.

func (*Router) Unsubscribe 

func (r *Router) Unsubscribe(streamID uint32)

Unsubscribe removes all active subscriber registrations for streamID.

type RouterMetrics 

type RouterMetrics interface {
	OnPublish(topic string)
	OnDeliver(topic string)
	SetActiveSubscribers(n float64)
	OnRateLimited(clientID string)
}

RouterMetrics is the interface for publishing metrics from the router.

type RouterOption added in v1.2.0 

type RouterOption func(*Router)

RouterOption configures the Router.

func WithAALPath added in v1.4.0 

func WithAALPath(path string, key []byte) RouterOption

WithAALPath provides path and key for AAL backfill replay workers.

func WithBackpressurePolicy added in v1.2.0 

func WithBackpressurePolicy(p BackpressurePolicy) RouterOption

WithBackpressurePolicy sets the slow consumer isolation policy.

func WithBatchSize added in v1.6.0 

func WithBatchSize(n int) RouterOption

WithBatchSize sets the coalesced write batch size in bytes for subscriber writers. Must be > 0. Default: 64 KB.

func WithCompression added in v1.10.0 

func WithCompression(engine CompressionEngine, minBatchSize int) RouterOption

WithCompression enables batch payload compression for peer forwarding. engine provides ZSTD compression, minBatchSize is the minimum payload size in bytes before compression is applied (default 1024).

func WithFlushInterval added in v1.6.0 

func WithFlushInterval(d time.Duration) RouterOption

WithFlushInterval sets the maximum time to wait before flushing accumulated messages. Must be > 0. Default: 50 µs.

func WithMaxRetries added in v1.7.0 

func WithMaxRetries(n int) RouterOption

NewRouter creates a Router with optional metrics collector and configuration options. WithMaxRetries sets the maximum NACK retry count before a message is moved to DLQ.

func WithPeerForwarder added in v1.5.0 

func WithPeerForwarder(f PeerForwarder) RouterOption

WithPeerForwarder plugs a cluster PeerManager into the Router for inter-node forwarding.

func WithPriorityTTLs added in v1.11.0 

func WithPriorityTTLs(ttls [4]time.Duration) RouterOption

WithPriorityTTLs sets the per-priority TTL thresholds (array of 4 durations).

func WithQueueSize added in v1.2.0 

func WithQueueSize(n int) RouterOption

WithQueueSize sets the per-subscriber bounded channel queue size.

func WithQuotas added in v1.8.0 

func WithQuotas(qm *quotas.Manager) RouterOption

type SubscriptionSpec added in v1.4.0 

type SubscriptionSpec struct {
	Topic           string
	IsDurable       bool
	ConsumerID      string
	RequestedOffset uint64
	GroupID         string
}

SubscriptionSpec holds parsed parameters from a Subscribe command.

type WildcardSub added in v1.3.0 

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

WildcardSub stores a wildcard pattern registration and subscriber index.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.