broker

package
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 15, 2026 License: MIT Imports: 23 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func IsInbox 

func IsInbox(subject string) bool

IsInbox reports whether subject is a reply inbox subject.

func IsWildcard 

func IsWildcard(pattern string) bool

IsWildcard reports whether pattern contains any wildcard token.

func Match 

func Match(subject, pattern string) bool

Match reports whether subject matches the subscription pattern. Patterns support "*" (single token) and ">" (tail wildcard, which matches zero or more trailing tokens, per design §6.4).

func ResolveMode 

func ResolveMode(subject string, rules protocol.GroupDeliveryRules) protocol.Mode

ResolveMode determines the queue-group delivery mode for a published subject (design §5.2.1). Precedence:

  1. longest matching rule in the rules table,
  2. the reserved fanout prefix,
  3. load-balance by default.

func ValidatePattern 

func ValidatePattern(pattern string) error

ValidatePattern validates a subscription pattern: wildcards allowed with the usual placement rules ("*" alone in a token, ">" as the final token).

func ValidateSubject 

func ValidateSubject(subject string) error

ValidateSubject validates a publish subject: no wildcards, no empty tokens, no reserved characters, bounded length.

Types

type Bridge added in v1.1.0 

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

Bridge owns the Node mbus compatible outbound bridge and replication machinery plus the URL-subscriber registry that powers inbound replication (other mbuses registering as subscribers of this node's topics).

func NewBridge added in v1.1.0 

func NewBridge(cfg BridgeConfig, log *slog.Logger) *Bridge

NewBridge creates a Bridge. The registry is always live; outbound loops only start for configured features.

func (*Bridge) AddSubscriber added in v1.1.0 

func (b *Bridge) AddSubscriber(topic, name, callback string, filter map[string]any) (*urlSub, error)

AddSubscriber registers a URL subscriber for a topic. A subscriber with the same name already exists yields errSubscriberExists (mbus 409).

func (*Bridge) BridgeREST added in v1.1.0 

func (b *Bridge) BridgeREST(topic string, msg any)

BridgeREST forwards a REST-published message to the upstream REST API, skipping topics that are replicated from an upstream (anti-loop).

func (*Bridge) Close added in v1.1.0 

func (b *Bridge) Close()

Close cancels all outbound loops and waits for them to drain.

func (*Bridge) GetSubscriber added in v1.1.0 

func (b *Bridge) GetSubscriber(topic, id string) (*urlSub, error)

GetSubscriber returns one subscriber by id.

func (*Bridge) HasSubscriber added in v1.1.0 

func (b *Bridge) HasSubscriber(topic, id string) bool

HasSubscriber reports whether a subscriber id exists (mbus HEAD).

func (*Bridge) ListSubscribers added in v1.1.0 

func (b *Bridge) ListSubscribers(topic string) []*urlSub

ListSubscribers returns the subscribers of a topic.

func (*Bridge) NotifySubscribers added in v1.1.0 

func (b *Bridge) NotifySubscribers(topic string, msg any)

NotifySubscribers pushes a REST-published message to every URL subscriber of the topic whose filter matches (mbus _process).

func (*Bridge) RemoveSubscriber added in v1.1.0 

func (b *Bridge) RemoveSubscriber(topic, id string) error

RemoveSubscriber deletes a URL subscriber by id.

func (*Bridge) Replicated added in v1.1.0 

func (b *Bridge) Replicated(topic string) bool

Replicated reports whether topic is replicated from an upstream.

func (*Bridge) SocketBridge added in v1.1.0 

func (b *Bridge) SocketBridge(name string, args []any)

SocketBridge forwards a compat socket event to the upstream socket bridge when connected. Called after local delivery in the compat plane.

func (*Bridge) Start added in v1.1.0 

func (b *Bridge) Start()

Start launches the configured outbound loops: socket bridge producer and one replication registration loop per upstream.

type BridgeConfig added in v1.1.0 

type BridgeConfig struct {
	// RESTBridgeTo is the upstream REST API base for the REST bridge,
	// e.g. "http://host:19092/v1". REST publishes are POSTed to
	// <RESTBridgeTo>/topics/<topic>/messages. Empty = REST bridge disabled.
	RESTBridgeTo string

	// SocketBridgeTo is the upstream socket.io URL for the socket bridge,
	// e.g. "http://host:19092/v1/socket". Compat socket events are forwarded
	// to that server's /v1/socket namespace. Empty = socket bridge disabled.
	SocketBridgeTo string

	// ReplicateFrom lists upstream topic URLs to replicate from, each like
	// "http://host:19092/v1/topics/<topic>". On success this node registers
	// itself as a subscriber of the upstream topic.
	ReplicateFrom []string

	// ReplicateAs is this node's externally reachable address used as the
	// callback target when registering with an upstream replicator, e.g.
	// "http://192.168.0.41:39192". Empty = "localhost:<Port>".
	ReplicateAs string

	// Port is this node's HTTP port, used for the default ReplicateAs.
	// Defaults to 19092 (the Node mbus default).
	Port int

	// BridgeTimeout bounds bridge / subscriber HTTP calls. Default 30s.
	BridgeTimeout time.Duration

	// VerifyCodes lists HTTP statuses treated as success. Default [200].
	VerifyCodes []int
}

BridgeConfig configures Node mbus compatible bridging and replication (mbus config: messaging.bridgeTo / socket.bridgeTo / messaging.replicateFrom / messaging.replicateAs).

type Broker 

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

Broker is a single gombus message broker node: it assembles the gosocketio server, the subscription registry, the dispatcher, the request/reply router and (optionally) a cluster control plane.

func New 

func New(nodeID string, opts Options) (*Broker, error)

New creates a Broker with single-node (LocalCluster) control plane.

func (*Broker) ApplyClusterState 

func (b *Broker) ApplyClusterState(st *protocol.ClusterState)

ApplyClusterState mirrors a cluster state image into the local registry and delivery rules. It is the FSM notify target in cluster mode.

func (*Broker) Close 

func (b *Broker) Close()

Close shuts down the gosocketio server and the bridge loops.

func (*Broker) Cluster 

func (b *Broker) Cluster() Cluster

Cluster returns the control plane.

func (*Broker) Dispatcher 

func (b *Broker) Dispatcher() *Dispatcher

Dispatcher exposes the dispatcher (used by the stream manager to deliver and by tests).

func (*Broker) Handler 

func (b *Broker) Handler() http.Handler

Handler returns the broker as an http.Handler: the Node mbus REST surface on /v1/topics/... and the socket.io server everywhere else.

func (*Broker) NodeID 

func (b *Broker) NodeID() string

NodeID returns the broker node id.

func (*Broker) Registry 

func (b *Broker) Registry() *Registry

Registry exposes the subscription registry (used by the cluster FSM to apply replicated subscriptions).

func (*Broker) Server 

func (b *Broker) Server() *socketio.Server

Server returns the underlying gosocketio server.

func (*Broker) SetCluster 

func (b *Broker) SetCluster(c Cluster)

SetCluster swaps in a clustered control plane (called by cmd/broker when starting with raft peers).

func (*Broker) SetForwarder 

func (b *Broker) SetForwarder(f Forwarder)

SetForwarder installs the cross-node forwarder (cluster mode).

func (*Broker) SetStream 

func (b *Broker) SetStream(sm StreamManager)

SetStream installs the stream manager and wires its control events.

func (*Broker) Stats 

func (b *Broker) Stats() protocol.StatsSnapshot

Stats assembles a stats snapshot for the admin surface.

type Cluster 

type Cluster interface {
	// ProposeSubscribe / ProposeUnsubscribe replicate a subscription change
	// so every node's registry mirrors it. No-op when already applied.
	// Compat marks a Node mbus compatible subscription; Filter narrows its
	// delivery by message content.
	ProposeSubscribe(node, socketID, pattern, queue string, compat bool, filter map[string]any) error
	ProposeUnsubscribe(node, socketID, pattern, queue string) error

	// RegisterNode / RemoveNode manage cluster membership.
	RegisterNode(n protocol.Node) error
	RemoveNode(nodeID string) error

	// ProposeRules replicates the group delivery rules table.
	ProposeRules(rules protocol.GroupDeliveryRules) error

	// ProposeStream / DeleteStream / ProposeConsumer / DeleteConsumer
	// replicate stream and consumer metadata.
	ProposeStream(meta *protocol.StreamMeta) error
	DeleteStream(name string) error
	ProposeConsumer(meta *protocol.ConsumerMeta) error
	DeleteConsumer(stream, name string) error

	// ProposeCursor replicates a consumer's delivery cursor (acked seq and
	// inflight state) so it survives failover.
	ProposeCursor(stream, consumer string, cursor *protocol.Cursor) error

	// ApplyCommand applies a control mutation received from a peer node
	// (leader-forwarding path). Single-node clusters never receive commands.
	ApplyCommand(cmd protocol.ClusterCommand) error

	// State returns the local mirror of the control-plane state.
	State() *protocol.ClusterState

	// NodeID / Role / IsLeader expose the cluster view of this node.
	NodeID() string
	Role() string
	IsLeader() bool
}

Cluster is the control-plane contract consumed by the broker. Single-node brokers use LocalCluster; clustered brokers implement it over a Raft group (internal/cluster).

type Dispatcher 

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

Dispatcher resolves delivery targets for published messages and performs the emission, locally or via the forwarder.

func NewDispatcher 

func NewDispatcher(nodeID string, registry *Registry) *Dispatcher

NewDispatcher creates a dispatcher bound to a registry.

func (*Dispatcher) DeliverTo 

func (d *Dispatcher) DeliverTo(matchSubject string, msg protocol.Message)

DeliverTo dispatches msg against matchSubject while keeping the original subject carried by the message. It is used by the stream layer to deliver push-consumer messages on a consumer's deliver subject.

func (*Dispatcher) Publish 

func (d *Dispatcher) Publish(subject string, header map[string]string, data []byte) error

Publish validates and dispatches a message (core NATS semantics). It never blocks on the network: local sockets receive an async emit, remote targets are handed to the forwarder.

func (*Dispatcher) PublishCompat added in v1.1.0 

func (d *Dispatcher) PublishCompat(event string, args []any) error

PublishCompat publishes a Node mbus compatible message: the event name is the subject and the raw emit arguments are preserved per-argument so compatible consumers receive them unmodified.

func (*Dispatcher) PublishReply 

func (d *Dispatcher) PublishReply(subject string, data []byte) error

PublishReply publishes a reply to an inbox subject (request/reply).

func (*Dispatcher) Rules 

func (d *Dispatcher) Rules() protocol.GroupDeliveryRules

Rules returns the current group delivery rules.

func (*Dispatcher) SetForwarder 

func (d *Dispatcher) SetForwarder(f Forwarder)

SetForwarder installs the cross-node forwarder (cluster mode).

func (*Dispatcher) SetInboxHandler 

func (d *Dispatcher) SetInboxHandler(h InboxHandler)

SetInboxHandler installs the inbox router for request/reply.

func (*Dispatcher) SetRules 

func (d *Dispatcher) SetRules(rules protocol.GroupDeliveryRules)

SetRules atomically updates the group delivery rules (from the control raft FSM).

func (*Dispatcher) SocketAt 

func (d *Dispatcher) SocketAt(socketID string) *socketio.Socket

SocketAt returns the local socket with the given id, if any.

type Forwarder 

type Forwarder interface {
	// Forward sends msg to the given socket id on the remote node. emitEvent
	// is the original event name for Node mbus compatible delivery ("" means
	// the regular "message" event with the Message envelope).
	Forward(node string, msg protocol.Message, targetSocket string, emitEvent string) error
}

Forwarder delivers a message to a socket living on a remote node. The cluster implementation (peer.go) supplies it; single-node brokers leave it nil so every target is local.

type InboxHandler 

type InboxHandler interface {
	// Deliver hands a reply message to the inbox router. It returns true
	// when the message was consumed by a pending request.
	Deliver(msg protocol.Message) bool
}

InboxHandler consumes replies addressed to _INBOX.* subjects (request/ reply). A broker with request support installs it so that reply publishes never leak into the subscription registry.

type LocalCluster 

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

LocalCluster is the single-node control plane: every proposal is applied directly to the local state image and the local registry.

func NewLocalCluster 

func NewLocalCluster(nodeID string, onSubscribe, onUnsubscribe func(node, socketID, pattern, queue string)) *LocalCluster

NewLocalCluster creates a single-node control plane.

func (*LocalCluster) ApplyCommand 

func (c *LocalCluster) ApplyCommand(cmd protocol.ClusterCommand) error

ApplyCommand routes a peer-issued command through the proposal methods. LocalCluster is single-node, so no peer ever sends one; the routing keeps the Cluster contract uniform for the broker's cluster.command handler.

func (*LocalCluster) DeleteConsumer 

func (c *LocalCluster) DeleteConsumer(stream, name string) error

func (*LocalCluster) DeleteStream 

func (c *LocalCluster) DeleteStream(name string) error

func (*LocalCluster) IsLeader 

func (c *LocalCluster) IsLeader() bool

func (*LocalCluster) NodeID 

func (c *LocalCluster) NodeID() string

func (*LocalCluster) ProposeConsumer 

func (c *LocalCluster) ProposeConsumer(meta *protocol.ConsumerMeta) error

func (*LocalCluster) ProposeCursor 

func (c *LocalCluster) ProposeCursor(stream, consumer string, cursor *protocol.Cursor) error

func (*LocalCluster) ProposeRules 

func (c *LocalCluster) ProposeRules(rules protocol.GroupDeliveryRules) error

func (*LocalCluster) ProposeStream 

func (c *LocalCluster) ProposeStream(meta *protocol.StreamMeta) error

func (*LocalCluster) ProposeSubscribe 

func (c *LocalCluster) ProposeSubscribe(node, socketID, pattern, queue string, compat bool, filter map[string]any) error

func (*LocalCluster) ProposeUnsubscribe 

func (c *LocalCluster) ProposeUnsubscribe(node, socketID, pattern, queue string) error

func (*LocalCluster) RegisterNode 

func (c *LocalCluster) RegisterNode(n protocol.Node) error

func (*LocalCluster) RemoveNode 

func (c *LocalCluster) RemoveNode(nodeID string) error

func (*LocalCluster) Role 

func (c *LocalCluster) Role() string

func (*LocalCluster) State 

func (c *LocalCluster) State() *protocol.ClusterState

type Options 

type Options struct {
	NodeID       string
	ControlToken string // token required on the /control namespace ("" = open)
	MaxPayload   int64  // per-message payload cap (0 = 1 MiB default)
	Logger       *slog.Logger
	Bridge       BridgeConfig // Node mbus compatible bridge/replication (optional)
}

Options configures a Broker.

type Registry 

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

Registry is the local mirror of the cluster subscription table. In single node mode it is authoritative; in cluster mode it is fed by the control raft group (remote subscriptions) plus the local socket connections.

The registry drives delivery: for a published subject it resolves the set of matching subscriptions, aggregates queue groups, applies the group delivery mode (loadbalance -> pick one member, broadcast -> all members) and returns the target list.

func NewRegistry 

func NewRegistry(localNode string) *Registry

NewRegistry creates an empty registry for the given local node id.

func (*Registry) Add 

func (r *Registry) Add(s *socketio.Socket, pattern, queue string, compat bool, filter map[string]any) error

Add registers a local subscription. The socket is stored so the dispatcher can deliver to it directly.

func (*Registry) AddRemote 

func (r *Registry) AddRemote(node, socketID, pattern, queue string, compat bool, filter map[string]any) error

AddRemote registers a subscription owned by another node (applied from the control raft FSM image).

func (*Registry) ClientCount 

func (r *Registry) ClientCount() int

ClientCount returns the number of connected local sockets.

func (*Registry) Count 

func (r *Registry) Count() int

Count returns the number of distinct subscription patterns.

func (*Registry) Dispatch 

func (r *Registry) Dispatch(subject string, rules protocol.GroupDeliveryRules) []Target

Dispatch resolves delivery targets for a published subject, applying queue group semantics and the group delivery mode. It returns zero or more targets; the caller performs the actual emission (possibly cross-node).

Dedup rule: one delivery per (node, socket, queue) even when multiple patterns match the subject. Plain (non-queue) subscriptions always receive a copy; each queue group receives either one copy (loadbalance) or one per member (broadcast) according to ResolveMode.

func (*Registry) Remove 

func (r *Registry) Remove(socketID, pattern, queue string)

Remove unregisters a local subscription.

func (*Registry) RemoveRemote 

func (r *Registry) RemoveRemote(node, socketID, pattern, queue string)

RemoveRemote removes a subscription owned by another node.

func (*Registry) RemoveSocketAll 

func (r *Registry) RemoveSocketAll(socketID string) []Subscription

RemoveSocketAll removes every subscription of a local socket (called on disconnect). It returns the removed (node, pattern, queue) triples so the caller can replicate the cleanup to the cluster.

func (*Registry) ReplaceAll 

func (r *Registry) ReplaceAll(subs map[string][]protocol.Subscriber)

ReplaceAll rebuilds the registry from a cluster state image (subscriptions map). Remote entries are re-added; local socket references are preserved. Called when a snapshot is applied.

func (*Registry) Socket 

func (r *Registry) Socket(socketID string) *socketio.Socket

Socket returns the local socket with the given id, if connected.

func (*Registry) Subscribers 

func (r *Registry) Subscribers(subject string) []*Subscription

Subscribers returns the matching subscriptions for a subject (excluding the sender itself when sender is non-empty). Used for stats and debugging.

func (*Registry) SubscriptionsOf 

func (r *Registry) SubscriptionsOf(socketID string) []*Subscription

SubscriptionsOf lists the subscriptions of a socket (any node).

type RequestManager 

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

RequestManager implements request/reply: each request registers a temporary _INBOX.<node>.<n> subject, publishes a message carrying the reply subject to responders, and waits for the reply or the timeout.

Replies are routed by the Dispatcher's inbox hook, never through the subscription registry, so they cannot collide with user subjects.

func NewRequestManager 

func NewRequestManager(nodeID string, disp *Dispatcher, registry *Registry) *RequestManager

NewRequestManager creates the request/reply router for a dispatcher.

func (*RequestManager) Deliver 

func (rm *RequestManager) Deliver(msg protocol.Message) bool

Deliver routes a reply message to the matching pending request.

func (*RequestManager) Request 

func (rm *RequestManager) Request(subject string, data []byte, timeout time.Duration) protocol.ReplyMsg

Request issues a request and blocks until a responder replies or the timeout elapses. When no subscriber matches the subject at all the call fails fast with no_responders.

type StreamManager 

type StreamManager interface {
	StreamPublish(req protocol.StreamPublishReq) (protocol.StreamPublishAck, error)
	StreamAck(req protocol.StreamAckReq) error
	StreamNak(req protocol.StreamNakReq) error
	StreamPull(req protocol.StreamPullReq) ([]protocol.Message, error)
	CreateStream(req protocol.CreateStreamReq) (*protocol.StreamMeta, error)
	DeleteStream(name string) error
	PurgeStream(name string) error
	AddConsumer(req protocol.AddConsumerReq) error
	DeleteConsumer(req protocol.DeleteConsumerReq) error
	StreamInfo(name string) (protocol.StreamInfoAck, error)
}

StreamManager is implemented by internal/stream.Manager.

type Subscription 

type Subscription struct {
	Node     string // owning broker node id
	SocketID string // socket id on Node
	Pattern  string // subject pattern subscribed on
	Queue    string // queue group name ("" = plain pub/sub)
	Compat   bool
	Filter   map[string]any
	Socket   *socketio.Socket
}

Subscription is a single registration of a socket on a subject pattern, optionally within a queue group. Compat marks a Node mbus compatible subscription: delivery re-emits the original event name with the raw arguments (optionally narrowed by Filter) instead of the "message" event.

type Target 

type Target struct {
	Node      string // "" = local delivery via Socket
	SocketID  string
	Queue     string
	EmitEvent string // original event name for compat delivery ("" = "message" event)
	Filter    map[string]any
	Socket    *socketio.Socket // non-nil for local delivery
}

Target is one delivery destination resolved by Dispatch.

func (*Target) TargetKey 

func (t *Target) TargetKey() string

TargetKey deduplicates deliveries to the same (node, socket, queue).

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.