core

Documentation

Overview

Package eventx 是进程内事件总线基座：主题发布-订阅（同步/异步）、 通配符匹配、过滤器与优先级，与 errx / logx / tracex 家族打通。

典型用法：

bus := eventx.New()
sub, _ := bus.Subscribe("orders.created", func(ctx context.Context, e eventx.Event) error {
    return nil
})
defer sub.Unsubscribe()
_ = bus.Publish(context.Background(), "orders.created", "10086")

Index

• Constants
• type Bus
  • func New(opts ...Option) (*Bus, error)
  • func (b *Bus) Close() error
  • func (b *Bus) Metrics() Metrics
  • func (b *Bus) Publish(ctx context.Context, topic string, payload any) error
  • func (b *Bus) PublishAsync(ctx context.Context, topic string, payload any) error
  • func (b *Bus) Subscribe(topic string, handler Handler) (Subscription, error)
  • func (b *Bus) SubscribeFiltered(topic string, filter Filter, handler Handler) (Subscription, error)
  • func (b *Bus) SubscribeWithOptions(topic string, handler Handler, opts ...SubscribeOption) (Subscription, error)
• type Event
• type Filter
• type Handler
• type MetricProvider
• type Metrics
• type Option
  • func WithErrorHandler(fn func(error)) Option
  • func WithLogger(logger logx.Logger) Option
  • func WithQueueSize(n int) Option
  • func WithWorkers(n int) Option
• type SubscribeOption
  • func WithPriority(p int) SubscribeOption
• type Subscription
  • func SubscribeTyped[T any](b *Bus, topic string, handler TypedHandler[T]) (Subscription, error)
• type TypedHandler

Constants

const (
	// CodeBusClosed 总线已关闭。
	CodeBusClosed errx.Code = "EVENTX_BUS_CLOSED"
	// CodeInvalidTopic 主题非法。
	CodeInvalidTopic errx.Code = "EVENTX_INVALID_TOPIC"
	// CodeInvalidHandler 订阅处理函数非法。
	CodeInvalidHandler errx.Code = "EVENTX_INVALID_HANDLER"
	// CodeSubscriptionNotFound 订阅不存在。
	CodeSubscriptionNotFound errx.Code = "EVENTX_SUBSCRIPTION_NOT_FOUND"
	// CodeHandlerPanic 订阅处理函数发生未捕获异常。
	CodeHandlerPanic errx.Code = "EVENTX_HANDLER_PANIC"
	// CodeQueueFull 异步队列已满。
	CodeQueueFull errx.Code = "EVENTX_QUEUE_FULL"
	// CodeInvalidOption 选项参数非法。
	CodeInvalidOption errx.Code = "EVENTX_INVALID_OPTION"
	// CodeCancelled 发布或分发被上下文取消。
	CodeCancelled errx.Code = "EVENTX_CANCELLED"
	// CodeTypeMismatch 类型化订阅载荷类型不匹配。
	CodeTypeMismatch errx.Code = "EVENTX_TYPE_MISMATCH"
)

eventx 错误码全集：所有失败场景统一使用 errx 结构化错误。

Types

type Bus

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

Bus 是并发安全的事件总线，无全局单例。

func New

func New(opts ...Option) (*Bus, error)

New 创建事件总线。选项非法时返回 errx 错误。

func (*Bus) Close

func (b *Bus) Close() error

Close 关闭总线：拒绝新发布/订阅，排空异步在途任务后清空订阅。 重复关闭幂等。

func (*Bus) Metrics

func (b *Bus) Metrics() Metrics

Metrics 返回总线运行指标快照。

func (*Bus) Publish

func (b *Bus) Publish(ctx context.Context, topic string, payload any) error

Publish 同步发布事件：按优先级与注册顺序调用全部匹配 handler， 聚合所有返回的错误（errx.Join）后返回。

func (*Bus) PublishAsync

func (b *Bus) PublishAsync(ctx context.Context, topic string, payload any) error

PublishAsync 异步发布事件：投递到队列后立即返回。 队列满时返回 EVENTX_QUEUE_FULL；handler 错误通过 WithErrorHandler 上报。

func (*Bus) Subscribe

func (b *Bus) Subscribe(topic string, handler Handler) (Subscription, error)

Subscribe 按主题注册订阅（支持 `*` / `**` 通配符）。

func (*Bus) SubscribeFiltered

func (b *Bus) SubscribeFiltered(topic string, filter Filter, handler Handler) (Subscription, error)

SubscribeFiltered 按主题注册带过滤器的订阅；filter 为 nil 时不过滤。

func (*Bus) SubscribeWithOptions

func (b *Bus) SubscribeWithOptions(topic string, handler Handler, opts ...SubscribeOption) (Subscription, error)

SubscribeWithOptions 按主题注册订阅并应用订阅选项（优先级等）。

type Event

type Event struct {
	// Topic 发布主题。
	Topic string
	// Payload 事件载荷。
	Payload any
}

Event 是总线传递的事件。

type Filter

type Filter func(ctx context.Context, e Event) bool

Filter 是订阅过滤器；返回 false 时跳过该订阅者。

type Handler

type Handler func(ctx context.Context, e Event) error

Handler 是订阅处理函数；返回错误会被聚合。

type MetricProvider

type MetricProvider interface {
	Metrics() Metrics
}

MetricProvider 是总线指标接口，供 metricsx 等外部适配。

type Metrics

type Metrics struct {
	// Publishes 发布次数（同步分发开始/异步入队成功）。
	Publishes uint64
	// Deliveries 投递到订阅处理函数的次数。
	Deliveries uint64
	// Failures 订阅处理/过滤器失败次数。
	Failures uint64
	// Subscriptions 当前订阅数（精确 + 通配符）。
	Subscriptions uint64
}

Metrics 是总线运行指标快照。

type Option

type Option func(*config)

Option 修改总线构造配置。

func WithErrorHandler

func WithErrorHandler(fn func(error)) Option

WithErrorHandler 设置异步发布错误回调； 未设置时异步 handler 错误被忽略（建议生产环境设置）。

func WithLogger

func WithLogger(logger logx.Logger) Option

WithLogger 注入 logx.Logger，记录事件分发审计（载荷不记录）。

func WithQueueSize

func WithQueueSize(n int) Option

WithQueueSize 设置异步队列容量（默认 1024）。

func WithWorkers

func WithWorkers(n int) Option

WithWorkers 设置异步 worker 数（默认 1）。

type SubscribeOption

type SubscribeOption func(*subscribeConfig)

SubscribeOption 修改订阅配置。

func WithPriority

func WithPriority(p int) SubscribeOption

WithPriority 设置订阅优先级：数值越小越先执行（默认 0，稳定排序）。

type Subscription

type Subscription interface {
	// ID 返回订阅唯一标识。
	ID() uint64
	// Topic 返回订阅主题（可能含通配符）。
	Topic() string
	// Unsubscribe 取消订阅；重复取消幂等返回 nil。
	Unsubscribe() error
}

Subscription 是订阅句柄，可查询与取消。

func SubscribeTyped

func SubscribeTyped[T any](b *Bus, topic string, handler TypedHandler[T]) (Subscription, error)

SubscribeTyped 订阅类型化事件：发布载荷类型不匹配时返回 EVENTX_TYPE_MISMATCH（该订阅者失败，不影响其他订阅者）。

type TypedHandler

type TypedHandler[T any] func(ctx context.Context, topic string, payload T) error

TypedHandler 是类型化订阅处理函数：载荷已断言为 T。