diff --git a/components/forwarder/forwarder.go b/components/forwarder/forwarder.go index f4bdeaac2..826c536dc 100644 --- a/components/forwarder/forwarder.go +++ b/components/forwarder/forwarder.go @@ -9,13 +9,23 @@ import ( "github.com/pkg/errors" ) -const defaultForwarderTopic = "forwarder_topic" +const ( + defaultForwarderTopic = "forwarder_topic" + defaultHandlerName = "events_forwarder" +) type Config struct { // ForwarderTopic is a topic on which the forwarder will be listening to enveloped messages to forward. // Defaults to `forwarder_topic`. ForwarderTopic string + // HandlerName is the name of the forwarder's handler registered in the router. + // Defaults to `events_forwarder`. + // + // Handler names must be unique within a router, so it needs to be set explicitly + // if you want to run multiple forwarders on the same router. + HandlerName string + // Middlewares are used to decorate forwarder's handler function. Middlewares []message.HandlerMiddleware @@ -44,6 +54,9 @@ func (c *Config) setDefaults() { if c.ForwarderTopic == "" { c.ForwarderTopic = defaultForwarderTopic } + if c.HandlerName == "" { + c.HandlerName = defaultHandlerName + } if c.Marshaler == nil { c.Marshaler = DefaultMarshaler{} } @@ -53,6 +66,9 @@ func (c *Config) Validate() error { if c.ForwarderTopic == "" { return errors.New("empty forwarder topic") } + if c.HandlerName == "" { + return errors.New("empty handler name") + } return nil } @@ -99,7 +115,7 @@ func NewForwarder( f := &Forwarder{router, publisherOut, logger, config} handler := router.AddConsumerHandler( - "events_forwarder", + config.HandlerName, config.ForwarderTopic, subscriberIn, f.forwardMessage, diff --git a/components/forwarder/forwarder_test.go b/components/forwarder/forwarder_test.go index 0c592e829..33cd9ab44 100644 --- a/components/forwarder/forwarder_test.go +++ b/components/forwarder/forwarder_test.go @@ -9,6 +9,8 @@ import ( "github.com/ThreeDotsLabs/watermill/components/forwarder" "github.com/ThreeDotsLabs/watermill/message" "github.com/ThreeDotsLabs/watermill/pubsub/gochannel" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" ) @@ -109,6 +111,69 @@ func (s *ForwarderSuite) TestForwarder_publish_using_non_decorated_publisher_ack s.requireFirstAckingResult(msgAckedCh, true) } +// TestForwarder_multiple_forwarders_on_one_router ensures that forwarders sharing a router +// can coexist as long as they have unique handler names. +func TestForwarder_multiple_forwarders_on_one_router(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() + + pubSubIn := gochannel.NewGoChannel(gochannel.Config{}, logger) + pubSubOut := gochannel.NewGoChannel(gochannel.Config{}, logger) + + router, err := message.NewRouter(message.RouterConfig{}, logger) + require.NoError(t, err) + + handlerNames := []string{"first_forwarder", "second_forwarder"} + + for _, handlerName := range handlerNames { + _, err := forwarder.NewForwarder(pubSubIn, pubSubOut, logger, forwarder.Config{ + ForwarderTopic: handlerName + "_topic", + HandlerName: handlerName, + Router: router, + }) + require.NoError(t, err) + } + + go func() { + assert.NoError(t, router.Run(ctx)) + }() + defer func() { + assert.NoError(t, router.Close()) + }() + + select { + case <-router.Running(): + case <-ctx.Done(): + t.Fatal("router not running") + } + + outMessagesCh, err := pubSubOut.Subscribe(ctx, outTopic) + require.NoError(t, err) + + for _, handlerName := range handlerNames { + publisher := forwarder.NewPublisher(pubSubIn, forwarder.PublisherConfig{ + ForwarderTopic: handlerName + "_topic", + }) + + msg := message.NewMessage(watermill.NewUUID(), message.Payload(handlerName)) + require.NoError(t, publisher.Publish(outTopic, msg)) + } + + // Both forwarders publish to the same out topic, so messages can arrive in any order. + receivedPayloads := map[string]bool{} + for range handlerNames { + select { + case receivedMsg := <-outMessagesCh: + receivedPayloads[string(receivedMsg.Payload)] = true + receivedMsg.Ack() + case <-time.After(time.Second * 3): + t.Fatal("didn't receive all forwarded messages") + } + } + + assert.Equal(t, map[string]bool{"first_forwarder": true, "second_forwarder": true}, receivedPayloads) +} + type PubSubInPublisher struct { message.Publisher }