Add AddConsumerHandler (rename AddNoPublisherHandler) (#611)

* AddConsumerHandle

The current AddNoPublisherHandler name can be confusing, especially since it's the one most often used.

Author: m110

components/cqrs/command_processor.go

@@ -299,7 +299,7 @@ func (p CommandProcessor) addHandlerToRouter(r *message.Router, handler CommandH
 		return nil, errors.Wrap(err, "cannot create subscriber for command processor")
 	}
 
-	return r.AddNoPublisherHandler(
+	return r.AddConsumerHandler(
 		handlerName,
 		topicName,
 		subscriber,

components/cqrs/event_processor.go

@@ -303,7 +303,7 @@ func addHandlerToRouter(logger watermill.LoggerAdapter, r *message.Router, handl
 
 	logger.Debug("Adding CQRS event handler to router", nil)
 
-	return r.AddNoPublisherHandler(
+	return r.AddConsumerHandler(
 		handlerName,
 		topicName,
 		subscriber,

components/fanin/fanin_test.go

@@ -56,7 +56,7 @@ func TestFanIn(t *testing.T) {
 	receivedMessages := make(chan string, expectedNumberOfMessages)
 
 	for i := 0; i < workersCount; i++ {
-		router.AddNoPublisherHandler(
+		router.AddConsumerHandler(
 			fmt.Sprintf("worker-%v", i),
 			downstreamTopic,
 			pubsub,

components/forwarder/forwarder.go

@@ -85,7 +85,7 @@ func NewForwarder(subscriberIn message.Subscriber, publisherOut message.Publishe
 
 	f := &Forwarder{router, publisherOut, logger, config}
 
-	handler := router.AddNoPublisherHandler(
+	handler := router.AddConsumerHandler(
 		"events_forwarder",
 		config.ForwarderTopic,
 		subscriberIn,

components/requeuer/requeuer.go

@@ -110,7 +110,7 @@ func NewRequeuer(
 		config: config,
 	}
 
-	config.Router.AddNoPublisherHandler(
+	config.Router.AddConsumerHandler(
 		"requeuer",
 		config.SubscribeTopic,
 		config.Subscriber,

components/requeuer/requeuer_test.go

@@ -52,7 +52,7 @@ func TestRequeue(t *testing.T) {
 	lock := sync.Mutex{}
 	counter := 0
 
-	router.AddNoPublisherHandler(
+	router.AddConsumerHandler(
 		"test",
 		"test",
 		pubSub,

docs/content/docs/getting-started.md

@@ -322,7 +322,7 @@ Any messages returned from the handler function will be published to the given p
 *Note: the example above uses one `pubSub` argument for both the subscriber and publisher.
 It's because we use the `GoChannel` implementation, which is a simple in-memory Pub/Sub.*
 
-Alternatively, if you don't plan to publish messages from within the handler, you can use the simpler `AddNoPublisherHandler` method.
+Alternatively, if you don't plan to publish messages from within the handler, you can use the simpler `AddConsumerHandler` method.
 
 {{% load-snippet-partial file="src-link/_examples/basic/3-router/main.go" first_line_contains="AddNoPublisherHandler" last_line_contains=")" padding_after="0" %}}
 

docs/content/docs/messages-router.md

@@ -34,9 +34,9 @@ Next, you have to add a new handler with `Router.AddHandler`:
 See an example usage from [Getting Started]({{< ref "/docs/getting-started#using-messages-router" >}}):
 {{% load-snippet-partial file="src-link/_examples/basic/3-router/main.go" first_line_contains="// AddHandler returns a handler" last_line_contains="return h(message)" padding_after="3" %}}
 
-## No publisher handler
+## Consumer handler
 
-Not every handler will produce new messages. You can add this kind of handler by using `Router.AddNoPublisherHandler`:
+Not every handler will produce new messages. You can add this kind of handler by using `Router.AddConsumerHandler`:
 
 {{% load-snippet-partial file="src-link/message/router.go" first_line_contains="// AddNoPublisherHandler" last_line_contains=") {" padding_after="0" %}}
 
@@ -81,7 +81,7 @@ If the timeout is reached, `Close()` will return an error.
 ## Adding handler after the router has started
 
 You can add a new handler while the router is already running.
-To do that, you need to call `AddNoPublisherHandler` or `AddHandler` and call `RunHandlers`.
+To do that, you need to call `AddConsumerHandler` or `AddHandler` and call `RunHandlers`.
 
 {{% load-snippet-partial file="src-link/message/router.go" first_line_contains="// RunHandlers" last_line_contains="func (r *Router) RunHandlers" padding_after="0" %}}
 

message/router.go

@@ -265,7 +265,7 @@ func (d DuplicateHandlerNameError) Error() string {
 //
 // publishTopic is a topic to which router will produce messages returned by handlerFunc.
 // When handler needs to publish to multiple topics,
-// it is recommended to use AddNoPublisherHandler and inject a Publisher or implement middleware
+// it is recommended to use AddConsumerHandler and inject a Publisher or implement middleware
 // which will catch messages and publish to topic based on metadata for example.
 //
 // If handler is added while router is already running, you need to explicitly call RunHandlers().
@@ -329,9 +329,8 @@ func (r *Router) AddHandler(
 	}
 }
 
-// AddNoPublisherHandler adds a new handler.
-// This handler cannot return messages.
-// When message is returned it will occur an error and Nack will be sent.
+// AddConsumerHandler adds a new handler that does not return any messages.
+// It can publish messages by directly using a Publisher.
 //
 // handlerName must be unique. For now, it is used only for debugging.
 //
@@ -340,7 +339,7 @@ func (r *Router) AddHandler(
 // subscriber is Subscriber from which messages will be consumed.
 //
 // If handler is added while router is already running, you need to explicitly call RunHandlers().
-func (r *Router) AddNoPublisherHandler(
+func (r *Router) AddConsumerHandler(
 	handlerName string,
 	subscribeTopic string,
 	subscriber Subscriber,
@@ -353,6 +352,27 @@ func (r *Router) AddNoPublisherHandler(
 	return r.AddHandler(handlerName, subscribeTopic, subscriber, "", disabledPublisher{}, handlerFuncAdapter)
 }
 
+// AddNoPublisherHandler adds a new handler.
+// This handler cannot return messages.
+//
+// handlerName must be unique. For now, it is used only for debugging.
+//
+// subscribeTopic is a topic from which handler will receive messages.
+//
+// subscriber is Subscriber from which messages will be consumed.
+//
+// If handler is added while router is already running, you need to explicitly call RunHandlers().
+//
+// Deprecated: use AddConsumerHandler instead.
+func (r *Router) AddNoPublisherHandler(
+	handlerName string,
+	subscribeTopic string,
+	subscriber Subscriber,
+	handlerFunc NoPublishHandlerFunc,
+) *Handler {
+	return r.AddConsumerHandler(handlerName, subscribeTopic, subscriber, handlerFunc)
+}
+
 // Run runs all plugins and handlers and starts subscribing to provided topics.
 // This call is blocking while the router is running.
 //

message/router/middleware/poison_test.go

@@ -125,7 +125,7 @@ func TestPoisonQueue_context_values(t *testing.T) {
 	require.NoError(t, err)
 	router.AddMiddleware(pq)
 
-	router.AddNoPublisherHandler("handler_name", "test", pubSub, func(msg *message.Message) error {
+	router.AddConsumerHandler("handler_name", "test", pubSub, func(msg *message.Message) error {
 		return errors.New("error")
 	})