Watermill（Golang 事件驱动库）Message Router 解析

// ...
type RouterConfig struct {
   // CloseTimeout 确定路由在关闭时应为处理程序工作多长时间。
   CloseTimeout time.Duration
}

func (c *RouterConfig) setDefaults() {
    if c.CloseTimeout == 0 {
        c.CloseTimeout = time.Second * 30
    }
}

func (c RouterConfig) Validate() error {
    return nil
}
// ...

// ...
// HandlerFunc 是在收到消息时调用的函数。
//
// msg.Ack() 会在 HandlerFunc 没有返回错误时自动调用。
// 当 HandlerFunc 返回错误时，msg.Nack() 被调用。
// 当 msg.Ack() 在 handler 中被调用并且 HandlerFunc 返回错误时，
// msg.Nack() 将不会被发送，因为 Ack 已经发送了。
//
// HandlerFunc 在接收到多条消息时并行执行
// (因为 msg.Ack() 是在 HandlerFunc 中发送的，或者订阅者支持多个消费者)
type HandlerFunc func(msg *Message) ([]*Message, error)

// ...

// ...
// AddHandler 添加一个新的处理程序。
//
// handlerName 必须唯一。目前，它仅用于调试。
//
// subscribeTopic 是一个处理程序将从其中接收消息的 topic。
//
// publishTopic 是一个 router 将生成 handlerFunc 返回的消息的 topic。
// 
// 当处理程序需要发布到多个主题时，
// 建议仅向处理程序注入 Publisher 或实现中间件，
// 该中间件将捕获消息并基于例如元数据发布到主题。
func (r *Router) AddHandler(
    handlerName string,
    subscribeTopic string,
    subscriber Subscriber,
    publishTopic string,
    publisher Publisher,
    handlerFunc HandlerFunc,
) *Handler {
    r.logger.Info("Adding handler", watermill.LogFields{
        "handler_name": handlerName,
        "topic":        subscribeTopic,
    })

    if _, ok := r.handlers[handlerName]; ok {
        panic(DuplicateHandlerNameError{handlerName})
    }

    publisherName, subscriberName := internal.StructName(publisher), internal.StructName(subscriber)

    newHandler := &handler{
        name:   handlerName,
        logger: r.logger,

        subscriber:     subscriber,
        subscribeTopic: subscribeTopic,
        subscriberName: subscriberName,

        publisher:     publisher,
        publishTopic:  publishTopic,
        publisherName: publisherName,

        handlerFunc:       handlerFunc,
        runningHandlersWg: r.runningHandlersWg,
        messagesCh:        nil,
        closeCh:           r.closeCh,
    }

    r.handlers[handlerName] = newHandler

    return &Handler{
        router:  r,
        handler: newHandler,
    }
}

// AddNoPublisherHandler 添加一个新的 handler。
// 该 handler 无法返回消息。
// 当消息返回时，它将发生一个错误，Nack 将被发送。
//
// handlerName 必须唯一。目前，它仅用于调试。
//
// subscribeTopic 是一个 handler 将从其中接收消息的 topic。
//
// subscriber 是将从其消费消息的 Subscriber。
func (r *Router) AddNoPublisherHandler(
    handlerName string,
    subscribeTopic string,
    subscriber Subscriber,
    handlerFunc NoPublishHandlerFunc,
) {
// ...

// ...
   handler := router.AddHandler(
        "struct_handler",          // handler 名称，必须是唯一的
       "incoming_messages_topic", // 我们将从中读取事件的 topic
       pubSub,
        "outgoing_messages_topic", // 我们将向其发布事件的 topic
       pubSub,
        structHandler{}.Handler,
    )
// ...

// ...
// AddNoPublisherHandler 添加一个新的 handler。
// 该 handler 无法返回消息。
// 当消息返回时，它将发生一个错误，Nack 将被发送。
//
// handlerName 必须唯一。目前，它仅用于调试。
//
// subscribeTopic 是一个 handler 将从其中接收消息的 topic。
//
// subscriber 是将从其消费消息的 Subscriber。
func (r *Router) AddNoPublisherHandler(
    handlerName string,
    subscribeTopic string,
    subscriber Subscriber,
    handlerFunc NoPublishHandlerFunc,
) {
// ...

// ...
// Run 运行所有插件和处理程序，并开始订阅所提供的 topic。
// 当 router 正在运行时，此调用被阻塞。
//
// 当所有处理程序都停止时(例如，因为订阅已关闭)，router 也将停止。
//
// 要停止 Run()，你应该在路由器上调用 Close()。
//
// ctx 将传播给所有订阅者。
//
// 当所有处理程序都停止时(例如:因为关闭连接)，Run() 也将停止。
func (r *Router) Run(ctx context.Context) (err error) {
// ...

// ...
// Running is closed when router is running.
// In other words: you can wait till router is running using
//        fmt.Println("Starting router")
//        go r.Run(ctx)
//        <- r.Running()
//        fmt.Println("Router is running")
func (r *Router) Running() chan struct{} {
// ...

// ...
// HandlerMiddleware 允许我们编写类似 HandlerFunc 的装饰器。
// 它可以在处理程序之前执行某些事情(例如:修改已消费的消息)
// 或之后(修改产生的消息，对被消费的消息进行 ack/nack，处理错误，记录日志，等等)执行一些事情。
//
// 它可以通过 `AddMiddleware` 方法附加到路由器上。
//
// Example:
//        func ExampleMiddleware(h message.HandlerFunc) message.HandlerFunc {
//            return func(message *message.Message) ([]*message.Message, error) {
//                fmt.Println("executed before handler")
//                producedMessages, err := h(message)
//                fmt.Println("executed after handler")
//
//                return producedMessages, err
//            }
//        }
type HandlerMiddleware func(h HandlerFunc) HandlerFunc

// ...

// ...
// RouterPlugin 是一个函数，在 Router 启动时执行。
type RouterPlugin func(*Router) error

// ...

// ...
// HandlerNameFromCtx 返回使用该消息的路由中的消息处理程序的名称。
func HandlerNameFromCtx(ctx context.Context) string {
    return valFromCtx(ctx, handlerNameKey)
}

// PublisherNameFromCtx 返回在路由中发布消息的消息发布者类型的名称。
// For example, for Kafka it will be `kafka.Publisher`.
func PublisherNameFromCtx(ctx context.Context) string {
    return valFromCtx(ctx, publisherNameKey)
}

// SubscriberNameFromCtx 返回在路由中订阅该消息的消息订阅者类型的名称。
// For example, for Kafka it will be `kafka.Subscriber`.
func SubscriberNameFromCtx(ctx context.Context) string {
    return valFromCtx(ctx, subscriberNameKey)
}

// SubscribeTopicFromCtx 返回从路由接收到消息的主题。
func SubscribeTopicFromCtx(ctx context.Context) string {
    return valFromCtx(ctx, subscribeTopicKey)
}

// PublishTopicFromCtx 返回路由将向其发布消息的主题。
func PublishTopicFromCtx(ctx context.Context) string {
    return valFromCtx(ctx, publishTopicKey)
}
// ...