Implement custom metrics reporter for message execution and errors

Author: Daniel Bustamante Ospina

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/CommandListenersConfig.java

@@ -6,7 +6,7 @@
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationCommandListener;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -27,7 +27,7 @@ public class CommandListenersConfig {
     public ApplicationCommandListener applicationCommandListener(ReactiveMessageListener listener,
                                                                  HandlerResolver resolver, MessageConverter converter,
                                                                  DiscardNotifier discardNotifier,
-                                                                 CustomErrorReporter errorReporter) {
+                                                                 CustomReporter errorReporter) {
         ApplicationCommandListener commandListener = new ApplicationCommandListener(listener, appName, resolver,
                 asyncProps.getDirect().getExchange(), converter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
                 asyncProps.getRetryDelay(), asyncProps.getDirect().getMaxLengthBytes(), discardNotifier, errorReporter);

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/EventListenersConfig.java

@@ -6,7 +6,7 @@
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -25,7 +25,7 @@ public class EventListenersConfig {
 
     @Bean
     public ApplicationEventListener eventListener(HandlerResolver resolver, MessageConverter messageConverter,
-                                                  ReactiveMessageListener receiver, DiscardNotifier discardNotifier, CustomErrorReporter errorReporter) {
+                                                  ReactiveMessageListener receiver, DiscardNotifier discardNotifier, CustomReporter errorReporter) {
 
         final ApplicationEventListener listener = new ApplicationEventListener(receiver,
                 appName + ".subsEvents", resolver, asyncProps.getDomain().getEvents().getExchange(),

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/NotificacionListenersConfig.java

@@ -6,7 +6,7 @@
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationNotificationListener;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -25,7 +25,7 @@ public class NotificacionListenersConfig {
 
     @Bean
     public ApplicationNotificationListener eventNotificationListener(HandlerResolver resolver, MessageConverter messageConverter,
-                                                                     ReactiveMessageListener receiver, DiscardNotifier discardNotifier, CustomErrorReporter errorReporter) {
+                                                                     ReactiveMessageListener receiver, DiscardNotifier discardNotifier, CustomReporter errorReporter) {
         final ApplicationNotificationListener listener = new ApplicationNotificationListener(
                 receiver,
                 asyncProps.getDomain().getEvents().getExchange(),

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/QueryListenerConfig.java

@@ -7,7 +7,7 @@
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationQueryListener;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -28,7 +28,7 @@ public class QueryListenerConfig {
     public ApplicationQueryListener queryListener(MessageConverter converter, HandlerResolver resolver,
                                                   ReactiveMessageSender sender, ReactiveMessageListener rlistener,
                                                   DiscardNotifier discardNotifier,
-                                                  CustomErrorReporter errorReporter) {
+                                                  CustomReporter errorReporter) {
         final ApplicationQueryListener listener = new ApplicationQueryListener(rlistener,
                 appName + ".query", resolver, sender, asyncProps.getDirect().getExchange(), converter,
                 asyncProps.getGlobal().getExchange(), asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/RabbitMqConfig.java

@@ -22,7 +22,7 @@
 import org.reactivecommons.async.impl.converters.json.DefaultObjectMapperSupplier;
 import org.reactivecommons.async.impl.converters.json.JacksonMessageConverter;
 import org.reactivecommons.async.impl.converters.json.ObjectMapperSupplier;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -141,8 +141,8 @@ public DiscardNotifier rabbitDiscardNotifier(ObjectMapperSupplier objectMapperSu
 
     @Bean
     @ConditionalOnMissingBean
-    public CustomErrorReporter reactiveCommonsCustomErrorReporter() {
-        return new CustomErrorReporter() {
+    public CustomReporter reactiveCommonsCustomErrorReporter() {
+        return new CustomReporter() {
             @Override
             public Mono<Void> reportError(Throwable ex, Message rawMessage, Command<?> message, boolean redelivered) {
                 return Mono.empty();

async/async-commons-starter/src/test/java/org/reactivecommons/async/impl/config/CommandListenersConfigTest.java

@@ -12,7 +12,7 @@
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationCommandListener;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -38,7 +38,7 @@ public class CommandListenersConfigTest {
     private final HandlerResolver handlerResolver = mock(HandlerResolver.class);
     private final MessageConverter messageConverter = mock(MessageConverter.class);
     private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
-    private final CustomErrorReporter customErrorReporter = mock(CustomErrorReporter.class);
+    private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
 
     @BeforeEach
@@ -62,7 +62,7 @@ public void applicationCommandListener() {
                 handlerResolver,
                 messageConverter,
                 discardNotifier,
-                customErrorReporter
+            customReporter
         );
         Assertions.assertThat(commandListener).isNotNull();
     }

async/async-commons-starter/src/test/java/org/reactivecommons/async/impl/config/EventListenersConfigTest.java

@@ -10,7 +10,7 @@
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -31,7 +31,7 @@ public class EventListenersConfigTest {
     private final HandlerResolver handlerResolver = mock(HandlerResolver.class);
     private final MessageConverter messageConverter = mock(MessageConverter.class);
     private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
-    private final CustomErrorReporter customErrorReporter = mock(CustomErrorReporter.class);
+    private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
 
     @BeforeEach
@@ -56,7 +56,7 @@ public void eventListener() {
                 messageConverter,
                 listener,
                 discardNotifier,
-                customErrorReporter
+            customReporter
         );
 
         Assertions.assertThat(eventListener).isNotNull();

async/async-commons-starter/src/test/java/org/reactivecommons/async/impl/config/NotificacionListenersConfigTest.java

@@ -10,7 +10,7 @@
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationNotificationListener;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -31,7 +31,7 @@ public class NotificacionListenersConfigTest {
     private final HandlerResolver handlerResolver = mock(HandlerResolver.class);
     private final MessageConverter messageConverter = mock(MessageConverter.class);
     private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
-    private final CustomErrorReporter customErrorReporter = mock(CustomErrorReporter.class);
+    private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
 
     @BeforeEach
@@ -52,7 +52,7 @@ public void init() {
     @Test
     public void eventNotificationListener() {
         final ApplicationNotificationListener applicationEventListener = config.
-                eventNotificationListener(handlerResolver, messageConverter, listener, discardNotifier, customErrorReporter);
+                eventNotificationListener(handlerResolver, messageConverter, listener, discardNotifier, customReporter);
         Assertions.assertThat(applicationEventListener).isNotNull();
     }
 }

async/async-commons-starter/src/test/java/org/reactivecommons/async/impl/config/QueryListenerConfigTest.java

@@ -11,7 +11,7 @@
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import org.reactivecommons.async.impl.listeners.ApplicationQueryListener;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -32,7 +32,7 @@ public class QueryListenerConfigTest {
     private final HandlerResolver handlerResolver = mock(HandlerResolver.class);
     private final MessageConverter messageConverter = mock(MessageConverter.class);
     private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
-    private final CustomErrorReporter customErrorReporter = mock(CustomErrorReporter.class);
+    private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
     private final ReactiveMessageSender sender = mock(ReactiveMessageSender.class);
 
@@ -53,7 +53,7 @@ public void init() {
 
     @Test
     public void queryListener() {
-        final ApplicationQueryListener queryListener = config.queryListener(messageConverter, handlerResolver, sender, listener, discardNotifier, customErrorReporter);
+        final ApplicationQueryListener queryListener = config.queryListener(messageConverter, handlerResolver, sender, listener, discardNotifier, customReporter);
         Assertions.assertThat(queryListener).isNotNull();
     }
 }

async/async-commons-starter/src/test/java/org/reactivecommons/async/impl/config/RabbitMqConfigTest.java

@@ -7,7 +7,7 @@
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.impl.communications.Message;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -46,18 +46,18 @@ public void retryInitialConnection() throws IOException, TimeoutException {
 
     @Test
     public void shouldCreateDefaultErrorReporter() {
-        final CustomErrorReporter errorReporter = config.reactiveCommonsCustomErrorReporter();
+        final CustomReporter errorReporter = config.reactiveCommonsCustomErrorReporter();
         assertThat(errorReporter.reportError(mock(Throwable.class), mock(Message.class), mock(Command.class), true)).isNotNull();
         assertThat(errorReporter.reportError(mock(Throwable.class), mock(Message.class), mock(DomainEvent.class), true)).isNotNull();
         assertThat(errorReporter.reportError(mock(Throwable.class), mock(Message.class), mock(AsyncQuery.class), true)).isNotNull();
     }
 
     @Test
     public void shouldGenerateDefaultReeporter() {
-        final CustomErrorReporter customErrorReporter = config.reactiveCommonsCustomErrorReporter();
-        final Mono<Void> r1 = customErrorReporter.reportError(mock(Throwable.class), mock(Message.class), mock(Command.class), true);
-        final Mono<Void> r2 = customErrorReporter.reportError(mock(Throwable.class), mock(Message.class), mock(DomainEvent.class), true);
-        final Mono<Void> r3 = customErrorReporter.reportError(mock(Throwable.class), mock(Message.class), mock(AsyncQuery.class), true);
+        final CustomReporter customReporter = config.reactiveCommonsCustomErrorReporter();
+        final Mono<Void> r1 = customReporter.reportError(mock(Throwable.class), mock(Message.class), mock(Command.class), true);
+        final Mono<Void> r2 = customReporter.reportError(mock(Throwable.class), mock(Message.class), mock(DomainEvent.class), true);
+        final Mono<Void> r3 = customReporter.reportError(mock(Throwable.class), mock(Message.class), mock(AsyncQuery.class), true);
 
         assertThat(r1).isNotNull();
         assertThat(r2).isNotNull();

async/async-commons/src/main/java/org/reactivecommons/async/impl/ext/CustomErrorReporter.java renamed to async/async-commons/src/main/java/org/reactivecommons/async/impl/ext/CustomReporter.java

@@ -6,7 +6,7 @@
 import org.reactivecommons.async.impl.communications.Message;
 import reactor.core.publisher.Mono;
 
-public interface CustomErrorReporter {
+public interface CustomReporter {
 
     String COMMAND_CLASS = "org.reactivecommons.api.domain.Command";
     String EVENT_CLASS = "org.reactivecommons.api.domain.DomainEvent";
@@ -25,6 +25,9 @@ default Mono<Void> reportError(Throwable ex, Message rawMessage, Object message,
         }
     }
 
+    default void reportMetric(String type, String handlerPath, Long duration, boolean success) {
+    }
+
     Mono<Void> reportError(Throwable ex, Message rawMessage, Command<?> message, boolean redelivered);
     Mono<Void> reportError(Throwable ex, Message rawMessage, DomainEvent<?> message, boolean redelivered);
     Mono<Void> reportError(Throwable ex, Message rawMessage, AsyncQuery<?> message, boolean redelivered);

async/async-commons/src/main/java/org/reactivecommons/async/impl/listeners/ApplicationCommandListener.java

@@ -12,12 +12,11 @@
 import org.reactivecommons.async.impl.RabbitMessage;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.TopologyCreator;
-import org.reactivecommons.async.impl.ext.CustomErrorReporter;
+import org.reactivecommons.async.impl.ext.CustomReporter;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.AcknowledgableDelivery;
 import reactor.rabbitmq.BindingSpecification;
 import reactor.rabbitmq.ExchangeSpecification;
-import reactor.rabbitmq.QueueSpecification;
 
 import java.util.Optional;
 import java.util.function.Function;
@@ -33,7 +32,7 @@ public class ApplicationCommandListener extends GenericMessageListener {
     private final Optional<Integer> maxLengthBytes;
 
     //TODO: change large constructor parameters number
-    public ApplicationCommandListener(ReactiveMessageListener listener, String queueName, HandlerResolver resolver, String directExchange, MessageConverter messageConverter, boolean withDLQRetry, long maxRetries, int retryDelay, Optional<Integer> maxLengthBytes, DiscardNotifier discardNotifier, CustomErrorReporter errorReporter) {
+    public ApplicationCommandListener(ReactiveMessageListener listener, String queueName, HandlerResolver resolver, String directExchange, MessageConverter messageConverter, boolean withDLQRetry, long maxRetries, int retryDelay, Optional<Integer> maxLengthBytes, DiscardNotifier discardNotifier, CustomReporter errorReporter) {
         super(queueName, listener, withDLQRetry, maxRetries, discardNotifier, "command", errorReporter);
         this.retryDelay = retryDelay;
         this.withDLQRetry = withDLQRetry;
@@ -44,16 +43,15 @@ public ApplicationCommandListener(ReactiveMessageListener listener, String queue
     }
 
     protected Mono<Void> setUpBindings(TopologyCreator creator) {
+        final Mono<AMQP.Exchange.DeclareOk> declareExchange = creator.declare(ExchangeSpecification.exchange(directExchange).durable(true).type("direct"));
         if (withDLQRetry) {
-            final Mono<AMQP.Exchange.DeclareOk> declareExchange = creator.declare(ExchangeSpecification.exchange(directExchange).durable(true).type("direct"));
             final Mono<AMQP.Exchange.DeclareOk> declareExchangeDLQ = creator.declare(ExchangeSpecification.exchange(directExchange + ".DLQ").durable(true).type("direct"));
             final Mono<AMQP.Queue.DeclareOk> declareQueue = creator.declareQueue(queueName, directExchange + ".DLQ", maxLengthBytes);
             final Mono<AMQP.Queue.DeclareOk> declareDLQ = creator.declareDLQ(queueName, directExchange, retryDelay, maxLengthBytes);
             final Mono<AMQP.Queue.BindOk> binding = creator.bind(BindingSpecification.binding(directExchange, queueName, queueName));
             final Mono<AMQP.Queue.BindOk> bindingDLQ = creator.bind(BindingSpecification.binding(directExchange + ".DLQ", queueName, queueName + ".DLQ"));
             return declareExchange.then(declareExchangeDLQ).then(declareDLQ).then(declareQueue).then(bindingDLQ).then(binding).then();
         } else {
-            final Mono<AMQP.Exchange.DeclareOk> declareExchange = creator.declare(ExchangeSpecification.exchange(directExchange).durable(true).type("direct"));
             final Mono<AMQP.Queue.DeclareOk> declareQueue = creator.declareQueue(queueName, maxLengthBytes);
             final Mono<AMQP.Queue.BindOk> binding = creator.bind(BindingSpecification.binding(directExchange, queueName, queueName));
             return declareExchange.then(declareQueue).then(binding).then();