package cqrs import ( stdErrors "errors" "fmt" "github.com/ThreeDotsLabs/watermill" "github.com/ThreeDotsLabs/watermill/message" "github.com/pkg/errors" ) type EventGroupProcessorConfig struct { // GenerateSubscribeTopic is used to generate topic for subscribing to events for handler groups. // This option is required for EventProcessor if handler groups are used. GenerateSubscribeTopic EventGroupProcessorGenerateSubscribeTopicFn // SubscriberConstructor is used to create subscriber for GroupEventHandler. // This function is called for every events group once - thanks to that it's possible to have one subscription per group. // It's useful, when we are processing events from one stream and we want to do it in order. SubscriberConstructor EventGroupProcessorSubscriberConstructorFn // OnHandle is called before handling event. // OnHandle works in a similar way to middlewares: you can inject additional logic before and after handling an event. // // Because of that, you need to explicitly call params.Handler.Handle() to handle the event. // // func(params EventGroupProcessorOnHandleParams) (err error) { // // logic before handle // // (...) // // err := params.Handler.Handle(params.Message.Context(), params.Event) // // // logic after handle // // (...) // // return err // } // // This option is not required. OnHandle EventGroupProcessorOnHandleFn // AckOnUnknownEvent is used to decide if message should be acked if event has no handler defined. AckOnUnknownEvent bool // Marshaler is used to marshal and unmarshal events. // It is required. Marshaler CommandEventMarshaler // Logger instance used to log. // If not provided, watermill.NopLogger is used. Logger watermill.LoggerAdapter } func (c *EventGroupProcessorConfig) setDefaults() { if c.Logger == nil { c.Logger = watermill.NopLogger{} } } func (c EventGroupProcessorConfig) Validate() error { var err error if c.Marshaler == nil { err = stdErrors.Join(err, errors.New("missing Marshaler")) } if c.GenerateSubscribeTopic == nil { err = stdErrors.Join(err, errors.New("missing GenerateHandlerGroupTopic")) } if c.SubscriberConstructor == nil { err = stdErrors.Join(err, errors.New("missing SubscriberConstructor")) } return err } type EventGroupProcessorGenerateSubscribeTopicFn func(EventGroupProcessorGenerateSubscribeTopicParams) (string, error) type EventGroupProcessorGenerateSubscribeTopicParams struct { EventGroupName string EventGroupHandlers []GroupEventHandler } type EventGroupProcessorSubscriberConstructorFn func(EventGroupProcessorSubscriberConstructorParams) (message.Subscriber, error) type EventGroupProcessorSubscriberConstructorParams struct { EventGroupName string EventGroupHandlers []GroupEventHandler } type EventGroupProcessorOnHandleFn func(params EventGroupProcessorOnHandleParams) error type EventGroupProcessorOnHandleParams struct { GroupName string Handler GroupEventHandler Event any EventName string // Message is never nil and can be modified. Message *message.Message } // EventGroupProcessor determines which EventHandler should handle event received from event bus. // Compared to EventProcessor, EventGroupProcessor allows to have multiple handlers that share the same subscriber instance. type EventGroupProcessor struct { router *message.Router groupEventHandlers map[string][]GroupEventHandler config EventGroupProcessorConfig } // NewEventGroupProcessorWithConfig creates a new EventGroupProcessor. func NewEventGroupProcessorWithConfig(router *message.Router, config EventGroupProcessorConfig) (*EventGroupProcessor, error) { config.setDefaults() if err := config.Validate(); err != nil { return nil, errors.Wrap(err, "invalid config EventProcessor") } if router == nil { return nil, errors.New("missing router") } return &EventGroupProcessor{ router: router, groupEventHandlers: map[string][]GroupEventHandler{}, config: config, }, nil } // AddHandlersGroup adds a new list of GroupEventHandler to the EventGroupProcessor and adds it to the router. // // Compared to AddHandlers, AddHandlersGroup allows to have multiple handlers that share the same subscriber instance. // // It's allowed to have multiple handlers for the same event type in one group, but we recommend to not do that. // Please keep in mind that those handlers will be processed within the same message. // If first handler succeeds and the second fails, the message will be re-delivered and the first will be re-executed. // // Handlers group needs to be unique within the EventProcessor instance. // // Handler group name is used as handler's name in router. func (p *EventGroupProcessor) AddHandlersGroup(groupName string, handlers ...GroupEventHandler) error { if len(handlers) == 0 { return errors.New("no handlers provided") } if _, ok := p.groupEventHandlers[groupName]; ok { return fmt.Errorf("event handler group '%s' already exists", groupName) } if err := p.addHandlerToRouter(p.router, groupName, handlers); err != nil { return err } p.groupEventHandlers[groupName] = handlers return nil } func (p EventGroupProcessor) addHandlerToRouter(r *message.Router, groupName string, handlersGroup []GroupEventHandler) error { for i, handler := range handlersGroup { if err := validateEvent(handler.NewEvent()); err != nil { return errors.Wrapf( err, "invalid event for handler %T (num %d) in group %s", handler, i, groupName, ) } } topicName, err := p.config.GenerateSubscribeTopic(EventGroupProcessorGenerateSubscribeTopicParams{ EventGroupName: groupName, EventGroupHandlers: handlersGroup, }) if err != nil { return errors.Wrapf(err, "cannot generate topic name for handler group %s", groupName) } logger := p.config.Logger.With(watermill.LogFields{ "event_handler_group_name": groupName, "topic": topicName, }) handlerFunc, err := p.routerHandlerGroupFunc(handlersGroup, groupName, logger) if err != nil { return err } subscriber, err := p.config.SubscriberConstructor(EventGroupProcessorSubscriberConstructorParams{ EventGroupName: groupName, EventGroupHandlers: handlersGroup, }) if err != nil { return errors.Wrap(err, "cannot create subscriber for event processor") } _ = addHandlerToRouter(p.config.Logger, r, groupName, topicName, handlerFunc, subscriber) return nil } func (p EventGroupProcessor) routerHandlerGroupFunc(handlers []GroupEventHandler, groupName string, logger watermill.LoggerAdapter) (message.NoPublishHandlerFunc, error) { return func(msg *message.Message) error { messageEventName := p.config.Marshaler.NameFromMessage(msg) handledAnyEvent := false for _, handler := range handlers { initEvent := handler.NewEvent() expectedEventName := p.config.Marshaler.Name(initEvent) event := handler.NewEvent() if messageEventName != expectedEventName { logger.Trace("Received different event type than expected, ignoring", watermill.LogFields{ "message_uuid": msg.UUID, "expected_event_type": expectedEventName, "received_event_type": messageEventName, }) continue } logger.Debug("Handling event", watermill.LogFields{ "message_uuid": msg.UUID, "received_event_type": messageEventName, }) ctx := CtxWithOriginalMessage(msg.Context(), msg) msg.SetContext(ctx) if err := p.config.Marshaler.Unmarshal(msg, event); err != nil { return err } handle := func(params EventGroupProcessorOnHandleParams) error { return params.Handler.Handle(ctx, params.Event) } if p.config.OnHandle != nil { handle = p.config.OnHandle } err := handle(EventGroupProcessorOnHandleParams{ GroupName: groupName, Handler: handler, EventName: messageEventName, Event: event, Message: msg, }) if err != nil { logger.Debug("Error when handling event", watermill.LogFields{"err": err}) return err } handledAnyEvent = true } if handledAnyEvent { return nil } if !p.config.AckOnUnknownEvent { return fmt.Errorf("no handler found for event %s", p.config.Marshaler.NameFromMessage(msg)) } else { logger.Trace("Received event can't be handled by any handler in handler group", watermill.LogFields{ "message_uuid": msg.UUID, "received_event_type": messageEventName, }) return nil } }, nil }