Router handlers method for router plugins (#289)

Exposed the registered router handlers to be used by a router plugin in order to allow use cases such as message reprocessing described in issue #288

Co-authored-by: Davi Henrique <davi_henrique@sicredi.com.br>

message/router.go

@@ -208,6 +208,17 @@ func (r *Router) AddSubscriberDecorators(dec ...SubscriberDecorator) {
 	r.subscriberDecorators = append(r.subscriberDecorators, dec...)
 }
 
+// Handlers returns all registered handlers.
+func (r *Router) Handlers() map[string]HandlerFunc {
+	handlers := map[string]HandlerFunc{}
+
+	for handlerName, handler := range r.handlers {
+		handlers[handlerName] = handler.handlerFunc
+	}
+
+	return handlers
+}
+
 // DuplicateHandlerNameError is sent in a panic when you try to add a second handler with the same name.
 type DuplicateHandlerNameError struct {
 	HandlerName string

message/router_test.go

@@ -1177,3 +1177,47 @@ func readMessages(messagesCh <-chan *message.Message, limit int, timeout time.Du
 
 	return receivedMessages, len(receivedMessages) == limit
 }
+
+func TestRouter_Handlers(t *testing.T) {
+	pub, sub := createPubSub()
+	defer func() {
+		assert.NoError(t, pub.Close())
+		assert.NoError(t, sub.Close())
+	}()
+
+	logger := watermill.NewCaptureLogger()
+
+	r, err := message.NewRouter(
+		message.RouterConfig{},
+		logger,
+	)
+	require.NoError(t, err)
+
+	handlerCalled := false
+
+	handlerName := "test_get_handler"
+
+	r.AddNoPublisherHandler(
+		handlerName,
+		"subscribe_topic",
+		sub,
+		func(msg *message.Message) error {
+			handlerCalled = true
+			return nil
+		},
+	)
+
+	actual := r.Handlers()
+
+	assert.Len(t, actual, 1)
+
+	actualHandler := actual[handlerName]
+
+	assert.NotNil(t, actualHandler)
+
+	messages, err := actualHandler(nil)
+
+	assert.Empty(t, messages)
+	assert.NoError(t, err)
+	assert.True(t, handlerCalled, "Handler function should be the same")
+}