From 740bcd2691451461f47f8cfa26eb2b0d0471a2f3 Mon Sep 17 00:00:00 2001 From: Davi Henrique Date: Mon, 20 Jun 2022 07:19:45 -0300 Subject: [PATCH] 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 --- message/router.go | 11 +++++++++++ message/router_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/message/router.go b/message/router.go index c8d223f7..320a5a22 100644 --- a/message/router.go +++ b/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 diff --git a/message/router_test.go b/message/router_test.go index 207c4582..56319028 100644 --- a/message/router_test.go +++ b/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") +}