broker

package
v0.51.0 Latest Latest
Warning

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

 Go to latest Published: Aug 5, 2026 License: Apache-2.0 Imports: 10 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultIdentityCacheSize = 10000
	DefaultIdentityCacheTTL  = 24 * time.Hour
)

Default bounds for the identity cache. These are deliberately conservative; operators with large client populations should raise IdentityCacheSize.

View Source 
const (
	HookProtocolMQTT    = "mqtt"
	HookProtocolAMQP10  = "amqp"
	HookProtocolAMQP091 = "amqp091"
	HookProtocolHTTP    = "http"
	HookProtocolCoAP    = "coap"

	HookAuthOnRegister    = "auth_on_register"
	HookAuthOnPublish     = "auth_on_publish"
	HookAuthOnSubscribe   = "auth_on_subscribe"
	HookAuthOnUnsubscribe = "auth_on_unsubscribe"

	HookFailDeny  = "deny"
	HookFailAllow = "allow"
)
View Source 
const (
	AMQP091ClientPrefix = "amqp091:"
	AMQP1ClientPrefix   = "amqp:"
	HTTPClientPrefix    = "http:"
	CoAPClientPrefix    = "coap:"
	ClientIDProperty    = "client_id"
	ExternalIDProperty  = "external_id"
	ProtocolProperty    = "protocol"
)
View Source 
const (
	ProtocolMQTT    = "mqtt"
	ProtocolAMQP091 = "amqp"
	ProtocolAMQP1   = "amqp1"
	ProtocolHTTP    = "http"
	ProtocolCoAP    = "coap"
)

Origin protocol identifiers written into ProtocolProperty by ingress handlers so downstream consumers can tell where a message came from.

View Source 
const ReservedPropertyPrefix = "_flux."

ReservedPropertyPrefix marks message properties that carry broker-internal state between trusted services.

Trust is a property of the connection's listener policy, never of its protocol. A connection carries reserved properties only when its policy marks it trusted. Every other connection has them stripped on ingress, so it cannot forge one, and on egress, so it cannot observe one another service set. MQTT, HTTP, CoAP, and AMQP 1.0 have no trusted listener and are therefore stripped in both directions. The AMQP 0.9.1 local listener is the trusted one, under whichever configuration key names it.

Trust decides only whether reserved properties cross the boundary at all. What a session may then do with them — publish, consume, relay an origin — comes from the authenticated principal's role, so a listener grants no capability of its own.

Variables

View Source 
var ErrClientNotConnected = errors.New("client not connected")

ErrClientNotConnected is returned by protocol adapters when a queue delivery targets a client that no longer has a live connection.

Functions

func AddClientIDProperty 

func AddClientIDProperty(props map[string]string, clientID string) map[string]string

AddClientIDProperty writes the protocol-level client identity into the shared properties map used for cross-node and cross-protocol delivery. External identity must be set explicitly by the ingress handler.

func ClientIDFromProperties 

func ClientIDFromProperties(props map[string]string) string

ClientIDFromProperties returns the protocol-level client identity carried in the shared properties map.

func EffectiveConsumerGroupID added in v0.50.0 

func EffectiveConsumerGroupID(groupID, pattern string) string

EffectiveConsumerGroupID returns the group identifier a queue consumer is actually registered under. A pattern-scoped consumer is a distinct group from an unpatterned one on the same queue, so the pattern qualifies the ID.

This is the identifier the queue manager reports back — in stale-consumer cleanup, for instance — so a protocol handler matching its own consumers against a manager-supplied group has to build the same string rather than compare the raw group it was asked to subscribe with.

func ExternalIDFromProperties 

func ExternalIDFromProperties(props map[string]string) string

ExternalIDFromProperties returns the authenticated external identity of the client that originally published the message.

func IsAMQP1Client 

func IsAMQP1Client(clientID string) bool

IsAMQP1Client returns true when the client ID belongs to an AMQP 1.0 connection.

func IsAMQP091Client 

func IsAMQP091Client(clientID string) bool

IsAMQP091Client returns true when the client ID belongs to an AMQP 0.9.1 connection.

func IsErrClientNotConnected 

func IsErrClientNotConnected(err error) bool

IsErrClientNotConnected reports whether err means a queue delivery target is gone. Across the cluster RPC the signal is carried structurally (the client_not_connected proto field, re-wrapped into ErrClientNotConnected by the sender), so errors.Is matches it end to end.

func IsReservedProperty added in v0.50.0 

func IsReservedProperty(key string) bool

IsReservedProperty reports whether key names a broker-internal property.

func ParseQueueFilter 

func ParseQueueFilter(filter string) (queueName, pattern string)

ParseQueueFilter parses a $queue/ prefixed filter into queue name and pattern. Examples:

  • "$queue/tasks" -> queueName="tasks", pattern=""
  • "$queue/tasks/images" -> queueName="tasks", pattern="images"
  • "$queue/tasks/images/#" -> queueName="tasks", pattern="images/#"

func PrefixedAMQP1ClientID 

func PrefixedAMQP1ClientID(containerID string) string

PrefixedAMQP1ClientID returns the canonical AMQP 1.0 client ID.

func PrefixedAMQP091ClientID 

func PrefixedAMQP091ClientID(connID string) string

PrefixedAMQP091ClientID returns the canonical AMQP 0.9.1 client ID.

Types

type AckKind 

type AckKind int

AckKind identifies the type of queue acknowledgment. 

const (
	AckAccept AckKind = iota
	AckNack
	AckReject
)

type AsyncEventHook 

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

AsyncEventHook wraps an EventHook so that events are queued and dispatched by a small worker pool. The publish hot path is decoupled from hook latency. Lifecycle events (Connect/Disconnect/Subscribe/Unsubscribe) block on enqueue because they are infrequent and ordering matters; Publish events drop when the queue is full so a slow hook cannot stall the broker.

AsyncEventHook is safe for concurrent use.

func NewAsyncEventHook 

func NewAsyncEventHook(inner EventHook, cfg AsyncEventHookConfig) *AsyncEventHook

NewAsyncEventHook wraps inner so events dispatch on a worker pool. Returns inner unchanged when inner is nil.

func (*AsyncEventHook) Close 

func (h *AsyncEventHook) Close() error

Close stops dispatcher workers. Drains the queue up to ShutdownTimeout, then closes the inner hook. Safe to call multiple times.

func (*AsyncEventHook) OnConnect 

func (h *AsyncEventHook) OnConnect(_ context.Context, clientID, username, protocol string) error

func (*AsyncEventHook) OnDisconnect 

func (h *AsyncEventHook) OnDisconnect(_ context.Context, clientID, reason string) error

func (*AsyncEventHook) OnPublish 

func (h *AsyncEventHook) OnPublish(_ context.Context, clientID, topic string, qos byte, payload []byte) error

func (*AsyncEventHook) OnSubscribe 

func (h *AsyncEventHook) OnSubscribe(_ context.Context, clientID, topic string, qos byte) error

func (*AsyncEventHook) OnUnsubscribe 

func (h *AsyncEventHook) OnUnsubscribe(_ context.Context, clientID, topic string) error

func (*AsyncEventHook) Stats 

func (h *AsyncEventHook) Stats() AsyncEventHookStats

Stats returns a snapshot of dispatcher counters.

type AsyncEventHookConfig 

type AsyncEventHookConfig struct {
	// Workers is the number of goroutines draining the event queue.
	// Defaults to 1 when <= 0.
	Workers int
	// QueueSize is the bounded capacity of the event channel. Defaults to
	// 1024 when <= 0. Publish events that arrive when the queue is full
	// are dropped (newest is rejected) and counted in DroppedCount.
	QueueSize int
	// DispatchTimeout bounds the time a worker spends in the underlying
	// EventHook call. Zero means no timeout (not recommended).
	DispatchTimeout time.Duration
	// ShutdownTimeout bounds how long Close() waits for the queue to drain.
	ShutdownTimeout time.Duration
	// Logger is used for dropped-event warnings and dispatch errors.
	Logger *slog.Logger
}

AsyncEventHookConfig configures an AsyncEventHook.

type AsyncEventHookStats 

type AsyncEventHookStats struct {
	Enqueued   uint64
	Dispatched uint64
	Dropped    uint64
	Errors     uint64
}

AsyncEventHookStats is a point-in-time snapshot of dispatcher counters.

type AuthEngine 

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

AuthEngine handles authentication and authorization. It transparently maps protocol-level client IDs to external identities returned by the Authenticator, so protocol handlers don't need to be aware of identity resolution.

The identity cache is bounded (TTL + LRU) so misbehaving disconnect paths cannot leak memory.

func NewAuthEngine 

func NewAuthEngine(auth Authenticator, authz Authorizer, opts ...AuthEngineOption) *AuthEngine

NewAuthEngine creates a new AuthEngine with the given authenticator and authorizer. Apply WithIdentityCache to override the default identity-cache bounds.

func (*AuthEngine) Authenticate 

func (e *AuthEngine) Authenticate(clientID, username, password string) (bool, string, error)

Authenticate validates client credentials. Returns true if authenticated or if no authenticator is configured. On success, also returns the resolved external identity (empty when the authenticator did not provide one) and caches it for subsequent authorization calls.

func (*AuthEngine) CanPublish 

func (e *AuthEngine) CanPublish(clientID, topic string) bool

CanPublish checks if a client is authorized to publish to a topic. Returns true if authorized or if no authorizer is configured.

func (*AuthEngine) CanSubscribe 

func (e *AuthEngine) CanSubscribe(clientID, filter string) bool

CanSubscribe checks if a client is authorized to subscribe to a topic filter. Returns true if authorized or if no authorizer is configured.

func (*AuthEngine) ExternalID 

func (e *AuthEngine) ExternalID(clientID string) string

ExternalID returns the authenticated external identity for a protocol client ID.

func (*AuthEngine) Forget 

func (e *AuthEngine) Forget(clientID string)

Forget removes the cached identity mapping for a client. Should be called when a client disconnects.

func (*AuthEngine) IdentityCacheLen 

func (e *AuthEngine) IdentityCacheLen() int

IdentityCacheLen returns the current number of cached identity mappings. Intended for monitoring; a steadily-growing value relative to live client count points to leaked Forget calls.

func (*AuthEngine) SetExternalID added in v0.50.0 

func (e *AuthEngine) SetExternalID(clientID, externalID string)

SetExternalID stores or replaces the resolved external identity for a client.

type AuthEngineOption 

type AuthEngineOption func(*authEngineOptions)

AuthEngineOption configures an AuthEngine.

func WithIdentityCache 

func WithIdentityCache(size int, ttl time.Duration) AuthEngineOption

WithIdentityCache sets the bounded identity-cache size and TTL. A non-positive size disables size-based eviction; a non-positive TTL disables expiry.

type Authenticator 

type Authenticator interface {
	Authenticate(clientID, username, secret string) (*AuthnResult, error)
}

Authenticator validates client credentials.

type AuthnResult 

type AuthnResult struct {
	Authenticated bool
	// ID is the external identity resolved by the auth provider (e.g. a UUID).
	// When non-empty, the AuthEngine stores this and passes it to the
	// Authorizer in place of the protocol-level client ID.
	ID string
}

AuthnResult holds the outcome of an authentication attempt.

type Authorizer 

type Authorizer interface {
	CanPublish(clientID string, topic string) bool
	CanSubscribe(clientID string, filter string) bool
}

Authorizer checks topic permissions. The clientID parameter receives the resolved external identity when available, otherwise the protocol-level client ID.

type BlockingHookEngine added in v0.50.0 

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

BlockingHookEngine wraps a hook provider with broker failure policy.

func NewBlockingHookEngine added in v0.50.0 

func NewBlockingHookEngine(provider BlockingHookProvider, failMode string, logger *slog.Logger, protocols, hooks map[string]bool) *BlockingHookEngine

NewBlockingHookEngine creates a blocking hook engine.

func (*BlockingHookEngine) Handle added in v0.50.0 

func (e *BlockingHookEngine) Handle(ctx context.Context, req BlockingHookRequest) (BlockingHookRequest, bool)

Handle runs a blocking hook and returns the possibly-mutated request plus whether processing may continue.

type BlockingHookProvider added in v0.50.0 

type BlockingHookProvider interface {
	HandleHook(ctx context.Context, req BlockingHookRequest) (BlockingHookResult, error)
}

BlockingHookProvider executes synchronous broker hooks.

type BlockingHookRequest added in v0.50.0 

type BlockingHookRequest struct {
	Hook       string
	ClientID   string
	ExternalID string
	Protocol   string
	Topic      string
	Payload    []byte
	QoS        byte
	Retain     bool
	Properties map[string]string
	Username   string
	Password   string
}

BlockingHookRequest describes a synchronous broker hook request.

type BlockingHookResult added in v0.50.0 

type BlockingHookResult struct {
	Allowed    bool
	Topic      string
	Payload    []byte
	PayloadSet bool
	QoS        byte
	QoSSet     bool
	Retain     bool
	RetainSet  bool
	Properties map[string]string
	ExternalID string
	Reason     string
	ReasonCode uint32
}

BlockingHookResult holds a hook decision and supported mutations.

type ChainedEventHook 

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

ChainedEventHook composes multiple EventHook implementations. Hooks are called in order; first error short-circuits.

func NewChainedEventHook 

func NewChainedEventHook(hooks ...EventHook) *ChainedEventHook

func (*ChainedEventHook) Close 

func (c *ChainedEventHook) Close() error

func (*ChainedEventHook) OnConnect 

func (c *ChainedEventHook) OnConnect(ctx context.Context, clientID, username, protocol string) error

func (*ChainedEventHook) OnDisconnect 

func (c *ChainedEventHook) OnDisconnect(ctx context.Context, clientID, reason string) error

func (*ChainedEventHook) OnPublish 

func (c *ChainedEventHook) OnPublish(ctx context.Context, clientID, topic string, qos byte, payload []byte) error

func (*ChainedEventHook) OnSubscribe 

func (c *ChainedEventHook) OnSubscribe(ctx context.Context, clientID, topic string, qos byte) error

func (*ChainedEventHook) OnUnsubscribe 

func (c *ChainedEventHook) OnUnsubscribe(ctx context.Context, clientID, topic string) error

type ClientRateLimiter 

type ClientRateLimiter interface {
	RateLimiter
	// OnClientDisconnect cleans up rate limiters for a disconnected client.
	OnClientDisconnect(clientID string)
}

ClientRateLimiter defines the interface for per-client rate limiting.

type CrossDeliverFunc 

type CrossDeliverFunc func(ctx context.Context, clientID string, topic string, payload []byte, qos byte, props map[string]string)

CrossDeliverFunc delivers a pub/sub message to a client in another protocol broker.

type EventHook 

type EventHook interface {
	OnConnect(ctx context.Context, clientID, username, protocol string) error
	OnDisconnect(ctx context.Context, clientID, reason string) error
	OnSubscribe(ctx context.Context, clientID, topic string, qos byte) error
	OnUnsubscribe(ctx context.Context, clientID, topic string) error
	OnPublish(ctx context.Context, clientID, topic string, qos byte, payload []byte) error
	Close() error
}

EventHook receives lifecycle and messaging events from the broker. Implementations should be non-blocking; slow hooks degrade broker throughput.

type ExistingQueueSubscriber added in v0.50.0 

type ExistingQueueSubscriber interface {
	SubscribeExisting(ctx context.Context, queueName, pattern, clientID, groupID, proxyNodeID string) error
	SubscribeExistingWithCursor(ctx context.Context, queueName, pattern, clientID, groupID, proxyNodeID string, cursor *types.CursorOption) error
}

ExistingQueueSubscriber manages subscriptions without creating queues or changing their configured type. Protocol authorization paths that grant read access only use this interface so consuming cannot become an administrative queue mutation.

type Notifier 

type Notifier interface {
	Notify(ctx context.Context, event events.Event) error
	Close() error
}

Notifier defines the interface for webhook notifications.

type QueueAcknowledger 

type QueueAcknowledger interface {
	// Ack acknowledges successful processing of a message by a consumer group.
	// groupID is required for fan-out support - each group acknowledges independently.
	Ack(ctx context.Context, queueName, messageID, groupID string) error
	// Nack negatively acknowledges a message for a consumer group (triggers retry).
	Nack(ctx context.Context, queueName, messageID, groupID string) error
	// Reject permanently rejects a message by a consumer group (move to DLQ).
	Reject(ctx context.Context, queueName, messageID, groupID, reason string) error
}

QueueAcknowledger handles queue delivery acknowledgments.

type QueueAdmin 

type QueueAdmin interface {
	QueueAdminRead
	QueueAdminWrite
}

QueueAdmin combines queue read and write management operations.

type QueueAdminRead 

type QueueAdminRead interface {
	// GetQueue returns the configuration for a queue.
	GetQueue(ctx context.Context, queueName string) (*types.QueueConfig, error)
	// ListQueues returns all queue configurations.
	ListQueues(ctx context.Context) ([]types.QueueConfig, error)
}

QueueAdminRead provides queue configuration read operations.

type QueueAdminWrite 

type QueueAdminWrite interface {
	// CreateQueue creates a new queue with the given configuration.
	CreateQueue(ctx context.Context, config types.QueueConfig) error
	// DeleteQueue deletes a queue by name.
	DeleteQueue(ctx context.Context, queueName string) error
}

QueueAdminWrite provides queue configuration mutation operations.

type QueueLifecycle 

type QueueLifecycle interface {
	Start(ctx context.Context) error
	Stop() error
	// UpdateHeartbeat updates the heartbeat timestamp for a consumer across all queues/groups.
	// This should be called when a PINGREQ is received from a client.
	UpdateHeartbeat(ctx context.Context, clientID string) error
}

QueueLifecycle controls queue manager startup/shutdown and heartbeats.

type QueueManager 

type QueueManager interface {
	QueuePublisher
	QueueSubscriber
	QueueAcknowledger
	QueueLifecycle
	QueueAdmin
}

QueueManager defines the interface for durable queue-based queue management.

type QueuePublisher 

type QueuePublisher interface {
	// Publish adds a message to all queues whose topic patterns match the topic.
	Publish(ctx context.Context, publish types.PublishRequest) error
}

QueuePublisher publishes queue-targeted messages.

type QueueStreamOps 

type QueueStreamOps interface {
	// UpdateQueue updates queue settings such as retention policy and queue type.
	UpdateQueue(ctx context.Context, config types.QueueConfig) error
	// CommitOffset commits a stream group offset when auto-commit is disabled.
	CommitOffset(ctx context.Context, queueName, groupID string, offset uint64) error
}

QueueStreamOps provides stream-specific queue operations.

type QueueSubscriber 

type QueueSubscriber interface {
	// Subscribe adds a consumer to a queue with optional pattern matching.
	Subscribe(ctx context.Context, queueName, pattern, clientID, groupID, proxyNodeID string) error
	// SubscribeWithCursor adds a consumer with explicit cursor positioning.
	SubscribeWithCursor(ctx context.Context, queueName, pattern, clientID, groupID, proxyNodeID string, cursor *types.CursorOption) error
	// Unsubscribe removes a consumer from a queue.
	Unsubscribe(ctx context.Context, queueName, pattern, clientID, groupID string) error
}

QueueSubscriber manages queue subscriptions.

type RateLimiter 

type RateLimiter interface {
	// AllowPublish checks if a publish from the given client is allowed.
	AllowPublish(clientID string) bool
	// AllowSubscribe checks if a subscription from the given client is allowed.
	AllowSubscribe(clientID string) bool
}

RateLimiter defines hot-path publish/subscribe checks.

type RouteKind 

type RouteKind int

RouteKind indicates how a publish or subscribe should be routed. 

const (
	// RoutePubSub routes through the topic-based pub/sub path.
	RoutePubSub RouteKind = iota
	// RouteQueue routes through the queue manager.
	RouteQueue
	// RouteQueueAck routes a queue acknowledgment (ack/nack/reject).
	RouteQueueAck
	// RouteQueueCommit routes a stream offset commit.
	RouteQueueCommit
)

type RouteResult 

type RouteResult struct {
	Kind RouteKind

	// QueueName is the resolved queue name (populated for RouteQueue, RouteQueueAck, RouteQueueCommit).
	QueueName string

	// Pattern is the topic pattern after the queue name (e.g., "images/#").
	Pattern string

	// PublishTopic is the full topic to use when publishing to the queue manager
	// (e.g., "$queue/tasks/images"). For RouteQueueAck, this is the base queue topic
	// without the /$ack suffix.
	PublishTopic string

	// AckKind identifies ack/nack/reject for RouteQueueAck results.
	AckKind AckKind
}

RouteResult holds the resolved routing decision for a topic or address.

type RoutingResolver 

type RoutingResolver struct{}

RoutingResolver encapsulates all routing decisions so protocol handlers don't need to know about the $queue/ prefix convention.

func NewRoutingResolver 

func NewRoutingResolver() *RoutingResolver

NewRoutingResolver creates a new RoutingResolver.

func (*RoutingResolver) IsQueueTopic 

func (r *RoutingResolver) IsQueueTopic(topic string) bool

IsQueueTopic returns true if the topic targets the queue system.

func (*RoutingResolver) QueueTopic 

func (r *RoutingResolver) QueueTopic(queueName string, parts ...string) string

QueueTopic constructs a full queue topic from a queue name and optional sub-path.

func (*RoutingResolver) Resolve 

func (r *RoutingResolver) Resolve(topic string) RouteResult

Resolve determines the routing for a given topic or address.

type StreamQueueManager 

type StreamQueueManager interface {
	QueueManager
	QueueStreamOps
}

StreamQueueManager extends QueueManager with stream-specific controls. Used by protocol implementations that support stream retention updates and manual commits.

type TopicQueuePublisher added in v0.51.0 

type TopicQueuePublisher interface {
	PublishToMatchingQueues(ctx context.Context, publish types.PublishRequest) error
}

TopicQueuePublisher captures an ordinary pub/sub publish in every existing queue whose configured topic pattern matches it. It never creates a queue. Implementations must copy payload and properties they retain before returning because protocol brokers may release or reuse them afterwards.

A returned error reports that capture failed; it must not fail the publish. Capture is a broker-side policy the publisher never asked for and gets no signal about, and it carries no durability barrier, so failing the publish would deny every subscriber a message without buying any guarantee — one queue's storage error would silence pub/sub across every topic its pattern covers. Callers log and continue delivering. A publisher that needs a persistence guarantee uses an exact publish target, whose durable append is synced before the confirm and does fail the publish.

Implementations must not perform the storage work on the caller's goroutine. The append honours no cancellation, so running it inline let a queue whose store stalls delay every subscriber of a matching topic and the publisher's acknowledgement with it. The queue manager resolves the matching queues synchronously — an in-memory lookup — and dispatches the appends, so a returned error reports only that resolution failed. An append that fails or is dropped afterwards is reported through the capture counters instead.

Source Files

View all Source files

Directories

Path Synopsis
Package localauth authenticates and authorizes principals configured locally in FluxMQ.
 Package localauth authenticates and authorizes principals configured locally in FluxMQ.

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.