core

README

core 目录说明

目录职责

• Broker 运行时总编排层：启动监听、接受连接、创建 Client、关闭清理。
• 串联发布路由、消息落库、投递唤醒、共享订阅与 will delay 扫描。

关键代码

• broker.go：Broker 结构体与依赖注入、ACL 插件挂载。
• broker_runtime.go：Start/Close、连接 accept loop、周期清理任务。
• broker_delivery.go：发布完成后的路由、任务写入、客户端唤醒。
• option.go：外部注入 store/center/plugin 的配置入口。

你会看到的行为

• 支持优雅关停广播 DISCONNECT。
• QoS0 在配置下可走“直发不落库”快路径。

Documentation

Index

• type Broker
  • func NewBroker(cfg config.Broker, pluginCfg config.Plugins, clusterCfg config.Cluster, ...) (*Broker, error)
  • func (b *Broker) ACLManager() *acl.Manager
  • func (b *Broker) CallRetry(task *retry.Task) error
  • func (b *Broker) CallTimeout(task *retry.Task) error
  • func (b *Broker) Close() error
  • func (b *Broker) Name() string
  • func (b *Broker) RoutePublish(ctx context.Context, m *brokerpublish.Message) error
  • func (b *Broker) SetPublishRetry(schedule facade.RetrySchedule)
  • func (b *Broker) SharedSubscriptionManager() *shared_manager.SharedSubscriptionManager
  • func (b *Broker) Start(ctx context.Context) error
• type Handlers
• type Observer
• type Option
  • func WithClientCenter(center session.Center) Option
  • func WithClientDeliveryEvent(ev delivery_notify.ClientDeliveryEvent) Option
  • func WithClientManager(manager *client.Manager) Option
  • func WithClusterController(controller cluster.State) Option
  • func WithDeliveryCursorStore(cursorStore delivery.CursorStore) Option
  • func WithDeliveryTaskStore(taskStore delivery.TaskStore) Option
  • func WithEvent(driver events.EventEmmiter) Option
  • func WithGlobalNotify(notify cluster.NodeController) Option
  • func WithHandlers(handlers *Handlers) Option
  • func WithKeyStore(store store.KVStore) Option
  • func WithNodeMeta(meta *cluster.NodeMeta) Option
  • func WithPlugins(plugins *plugin.Plugins) Option
  • func WithPublishRetry(schedule facade.RetrySchedule) Option
  • func WithRetainStore(retain *retain.Store) Option
  • func WithSharedSubscriptionStore(ss store.SharedSubscriptionStore) Option
  • func WithStateRouter(router staterouter.Client) Option
  • func WithSubCenter(tree subscription.Center) Option
  • func WithWillDelayCenter(center willdelay.Center, sessionCenter session.Center, cluster *raft.Cluster, ...) Option
• type Options

Types

type Broker

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

func NewBroker

func NewBroker(
	cfg config.Broker,
	pluginCfg config.Plugins,
	clusterCfg config.Cluster,
	deliveryCfg config.DeliveryRunner,
	option ...Option,
) (*Broker, error)

NewBroker 创建 Broker 核心实例，并完成 listener、基础缓存和插件依赖的初始化。

func (*Broker) ACLManager

func (b *Broker) ACLManager() *acl.Manager

func (*Broker) CallRetry

func (b *Broker) CallRetry(task *retry.Task) error

CallRetry 在发布重试任务触发时定位在线 client，并将消息交回 client 侧重发。

func (*Broker) CallTimeout

func (b *Broker) CallTimeout(task *retry.Task) error

CallTimeout handles publish retry timeout for the current connection while preserving persistent session state.

func (*Broker) Close

func (b *Broker) Close() error

Close 执行 Broker 的优雅关闭流程，先通知在线客户端，再停止后台组件并等待 client 退出。

func (*Broker) Name

func (b *Broker) Name() string

func (*Broker) RoutePublish

func (b *Broker) RoutePublish(ctx context.Context, m *brokerpublish.Message) error

RoutePublish 将 client 侧 publish 事件接入 Broker 的下行投递流水线。

func (*Broker) SetPublishRetry

func (b *Broker) SetPublishRetry(schedule facade.RetrySchedule)

func (*Broker) SharedSubscriptionManager

func (b *Broker) SharedSubscriptionManager() *shared_manager.SharedSubscriptionManager

func (*Broker) Start

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

Start 启动 MQTT Broker，并按顺序初始化集群注册、后台清理、共享订阅和遗嘱延迟扫描。

type Handlers

type Handlers struct {
	Connect     brokerHandler
	Publish     brokerHandler
	PublishAck  brokerHandler
	PublishRec  brokerHandler
	PublishRel  brokerHandler
	PublishComp brokerHandler
	Ping        brokerHandler
	Sub         brokerHandler
	UnSub       brokerHandler
	Auth        brokerHandler
	Disconnect  brokerHandler
}

type Observer

type Observer interface {
	OnClientClose(b Broker, c *brokerclient.Client)
}

type Option

type Option func(*Broker)

func WithClientCenter

func WithClientCenter(center session.Center) Option

func WithClientDeliveryEvent

func WithClientDeliveryEvent(ev delivery_notify.ClientDeliveryEvent) Option

func WithClientManager

func WithClientManager(manager *client.Manager) Option

func WithClusterController

func WithClusterController(controller cluster.State) Option

func WithDeliveryCursorStore

func WithDeliveryCursorStore(cursorStore delivery.CursorStore) Option

func WithDeliveryTaskStore

func WithDeliveryTaskStore(taskStore delivery.TaskStore) Option

func WithEvent

func WithEvent(driver events.EventEmmiter) Option

func WithGlobalNotify

func WithGlobalNotify(notify cluster.NodeController) Option

func WithHandlers

func WithHandlers(handlers *Handlers) Option

func WithKeyStore

func WithKeyStore(store store.KVStore) Option

func WithNodeMeta

func WithNodeMeta(meta *cluster.NodeMeta) Option

func WithPlugins

func WithPlugins(plugins *plugin.Plugins) Option

func WithPublishRetry

func WithPublishRetry(schedule facade.RetrySchedule) Option

func WithRetainStore

func WithRetainStore(retain *retain.Store) Option

func WithSharedSubscriptionStore

func WithSharedSubscriptionStore(ss store.SharedSubscriptionStore) Option

func WithStateRouter

func WithStateRouter(router staterouter.Client) Option

func WithSubCenter

func WithSubCenter(tree subscription.Center) Option

func WithWillDelayCenter

func WithWillDelayCenter(center willdelay.Center, sessionCenter session.Center, cluster *raft.Cluster, clusterID uint64, localNodeID uint64) Option

Scanner will be created in Start() when ctx is available

type Options

type Options struct {
}