break circular dependency

Author: juancgalvis

async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/communications/ReactiveMessageSender.java

@@ -38,7 +38,7 @@ public class ReactiveMessageSender {
     @Getter
     private final TopologyCreator topologyCreator;
     private final boolean isMandatory;
-    private final UnroutableMessageHandler unroutableMessageHandler;
+    private final UnroutableMessageNotifier unroutableMessageNotifier;
 
     private static final int NUMBER_OF_SENDER_SUBSCRIPTIONS = 4;
     private final CopyOnWriteArrayList<FluxSink<MyOutboundMessage>> fluxSinkConfirm = new CopyOnWriteArrayList<>();
@@ -55,16 +55,16 @@ public class ReactiveMessageSender {
 
     public ReactiveMessageSender(Sender sender, String sourceApplication,
                                  MessageConverter messageConverter, TopologyCreator topologyCreator,
-                                 boolean isMandatory, UnroutableMessageHandler unroutableMessageHandler) {
+                                 boolean isMandatory, UnroutableMessageNotifier unroutableMessageNotifier) {
         this.sender = sender;
         this.sourceApplication = sourceApplication;
         this.messageConverter = messageConverter;
         this.topologyCreator = topologyCreator;
-        this.isMandatory = isMandatory && unroutableMessageHandler != null;
-        this.unroutableMessageHandler = unroutableMessageHandler;
+        this.isMandatory = isMandatory;
+        this.unroutableMessageNotifier = unroutableMessageNotifier;
 
         System.out.println("ReactiveMessageSender initialized with mandatory: " + isMandatory);
-        System.out.println("onReturnedCallback: " + unroutableMessageHandler);
+        System.out.println("onReturnedCallback: " + unroutableMessageNotifier);
 
         initializeSenders();
     }
@@ -76,10 +76,11 @@ private void initializeSenders() {
                     .doOnNext((OutboundMessageResult<MyOutboundMessage> outboundMessageResult) -> {
                         System.out.println("MANDATORY: " + isMandatory);
                         if (outboundMessageResult.isReturned()) {
-                            System.out.println("CALLBACK: " + unroutableMessageHandler);
-                            this.unroutableMessageHandler.processMessage(outboundMessageResult);
+                            System.out.println("CALLBACK: " + unroutableMessageNotifier);
+                            this.unroutableMessageNotifier.notifyUnroutableMessage(outboundMessageResult);
                         }
-                        final Consumer<Boolean> ackNotifier = outboundMessageResult.getOutboundMessage().getAckNotifier();
+                        final Consumer<Boolean> ackNotifier =
+                                outboundMessageResult.getOutboundMessage().getAckNotifier();
                         executorService.submit(() -> ackNotifier.accept(outboundMessageResult.isAck()));
                     }).subscribe();
         }

async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/communications/UnroutableMessageNotifier.java

@@ -0,0 +1,32 @@
+package org.reactivecommons.async.rabbit.communications;
+
+import lombok.extern.java.Log;
+import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Schedulers;
+import reactor.rabbitmq.OutboundMessageResult;
+
+@Log
+public class UnroutableMessageNotifier {
+    private final Sinks.Many<OutboundMessageResult<MyOutboundMessage>> sink;
+
+    public UnroutableMessageNotifier() {
+        this.sink = Sinks.many().multicast().onBackpressureBuffer();
+    }
+
+    public void notifyUnroutableMessage(OutboundMessageResult<MyOutboundMessage> message) {
+        if (sink.tryEmitNext(message).isFailure()) {
+            log.warning("Failed to emit unroutable message: " + message);
+        }
+    }
+
+    public void listenToUnroutableMessages(UnroutableMessageHandler handler) {
+        sink.asFlux()
+                .flatMap(handler::processMessage)
+                .subscribeOn(Schedulers.boundedElastic())
+                .onErrorContinue((throwable, o) ->
+                        log.severe("Error processing unroutable message: " + throwable.getMessage())
+                )
+                .subscribe();
+
+    }
+}

async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGatewayTest.java

@@ -19,7 +19,7 @@
 import org.reactivecommons.async.commons.converters.json.DefaultObjectMapperSupplier;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
-import org.reactivecommons.async.rabbit.communications.UnroutableMessageHandler;
+import org.reactivecommons.async.rabbit.communications.UnroutableMessageNotifier;
 import org.reactivecommons.async.rabbit.converters.json.RabbitJacksonMessageConverter;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -68,7 +68,7 @@ class RabbitDirectAsyncGatewayTest {
     @Mock
     private ReactiveMessageSender senderMock;
     @Mock
-    private UnroutableMessageHandler unroutableMessageHandler;
+    private UnroutableMessageNotifier unroutableMessageNotifier;
 
     private final MeterRegistry meterRegistry = new SimpleMeterRegistry();
     private RabbitDirectAsyncGateway asyncGateway;
@@ -208,7 +208,7 @@ private void mockReply() throws JsonProcessingException {
 
     private ReactiveMessageSender getReactiveMessageSender() {
         Sender sender = new StubSender();
-        return new ReactiveMessageSender(sender, "sourceApplication", converter, null, true, unroutableMessageHandler);
+        return new ReactiveMessageSender(sender, "sourceApplication", converter, null, true, unroutableMessageNotifier);
     }
 
     private Flux<Command<DummyMessage>> createMessagesHot(int count) {

async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/communications/ReactiveMessageSenderTest.java

@@ -45,7 +45,7 @@ class ReactiveMessageSenderTest {
     private final SendOptions sendOptions = new SendOptions();
 
     @Mock
-    private UnroutableMessageHandler unroutableMessageHandler;
+    private UnroutableMessageNotifier unroutableMessageNotifier;
 
     @BeforeEach
     void init() {
@@ -61,7 +61,8 @@ void init() {
         });
         String sourceApplication = "TestApp";
 
-        messageSender = new ReactiveMessageSender(sender, sourceApplication, messageConverter, null, false, unroutableMessageHandler);
+        messageSender = new ReactiveMessageSender(sender, sourceApplication, messageConverter, null, false,
+                unroutableMessageNotifier);
     }
 
     @Test
@@ -73,15 +74,15 @@ void shouldCallUnroutableMessageHandlerWhenMessageIsReturned() {
         });
 
         messageSender = new ReactiveMessageSender(
-                sender, sourceApplication, messageConverter, null, true, unroutableMessageHandler
+                sender, sourceApplication, messageConverter, null, true, unroutableMessageNotifier
         );
         SomeClass messageContent = new SomeClass("id", "name", new Date());
 
         messageSender.sendWithConfirm(messageContent, "exchange", "rkey", new HashMap<>(), true)
                 .subscribe();
 
-        verify(unroutableMessageHandler, timeout(1000).times(1))
-                .processMessage(any(OutboundMessageResult.class));
+        verify(unroutableMessageNotifier, timeout(1000).times(1))
+                .notifyUnroutableMessage(any(OutboundMessageResult.class));
     }
 
 
@@ -113,7 +114,8 @@ void shouldSendBatchWithConfirmSuccessfully() {
         when(sender.sendWithPublishConfirms(any(Publisher.class)))
                 .thenReturn(Flux.just(result1, result2));
 
-        Flux<OutboundMessageResult> resultFlux = messageSender.sendWithConfirmBatch(messages, exchange, routingKey, headers, true);
+        Flux<OutboundMessageResult> resultFlux = messageSender.sendWithConfirmBatch(messages, exchange, routingKey,
+                headers, true);
 
         StepVerifier.create(resultFlux).verifyComplete();
     }

starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/RabbitMQBrokerProviderFactory.java

@@ -8,7 +8,7 @@
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
-import org.reactivecommons.async.rabbit.communications.UnroutableMessageHandler;
+import org.reactivecommons.async.rabbit.communications.UnroutableMessageNotifier;
 import org.reactivecommons.async.rabbit.config.ConnectionFactoryProvider;
 import org.reactivecommons.async.rabbit.config.RabbitProperties;
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
@@ -29,7 +29,7 @@ public class RabbitMQBrokerProviderFactory implements BrokerProviderFactory<Asyn
     private final MeterRegistry meterRegistry;
     private final CustomReporter errorReporter;
     private final RabbitMQDiscardProviderFactory discardProvider;
-    private final UnroutableMessageHandler unroutableMessageHandler;
+    private final UnroutableMessageNotifier unroutableMessageNotifier;
 
     @Override
     public String getBrokerType() {
@@ -47,7 +47,8 @@ public BrokerProvider<AsyncProps> getProvider(String domain, AsyncProps props, D
         ConnectionFactoryProvider provider = RabbitMQSetupUtils.connectionFactoryProvider(properties);
         RabbitReactiveHealthIndicator healthIndicator =
                 new RabbitReactiveHealthIndicator(domain, provider.getConnectionFactory());
-        ReactiveMessageSender sender = RabbitMQSetupUtils.createMessageSender(provider, props, converter, unroutableMessageHandler);
+        ReactiveMessageSender sender = RabbitMQSetupUtils.createMessageSender(provider, props, converter,
+                unroutableMessageNotifier);
         ReactiveMessageListener listener = RabbitMQSetupUtils.createMessageListener(provider, props);
         DiscardNotifier discardNotifier;
         if (props.isUseDiscardNotifierPerDomain()) {

starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/RabbitMQSetupUtils.java

@@ -15,6 +15,7 @@
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import org.reactivecommons.async.rabbit.communications.UnroutableMessageHandler;
+import org.reactivecommons.async.rabbit.communications.UnroutableMessageNotifier;
 import org.reactivecommons.async.rabbit.config.ConnectionFactoryProvider;
 import org.reactivecommons.async.rabbit.config.RabbitProperties;
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
@@ -95,11 +96,11 @@ public static ConnectionFactoryProvider connectionFactoryProvider(RabbitProperti
     public static ReactiveMessageSender createMessageSender(ConnectionFactoryProvider provider,
                                                             AsyncProps props,
                                                             MessageConverter converter,
-                                                            UnroutableMessageHandler unroutableMessageHandler) {
+                                                            UnroutableMessageNotifier unroutableMessageNotifier) {
         final Sender sender = RabbitFlux.createSender(reactiveCommonsSenderOptions(props.getAppName(), provider,
                 props.getConnectionProperties()));
         return new ReactiveMessageSender(sender, props.getAppName(), converter, new TopologyCreator(sender),
-                props.getMandatory(), unroutableMessageHandler
+                props.getMandatory(), unroutableMessageNotifier
         );
     }
 

starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/starter/impl/common/rabbit/RabbitMQConfig.java

@@ -6,6 +6,7 @@
 import org.reactivecommons.async.rabbit.RabbitMQBrokerProviderFactory;
 import org.reactivecommons.async.rabbit.RabbitMQFactory;
 import org.reactivecommons.async.rabbit.communications.UnroutableMessageHandler;
+import org.reactivecommons.async.rabbit.communications.UnroutableMessageNotifier;
 import org.reactivecommons.async.rabbit.config.RabbitProperties;
 import org.reactivecommons.async.rabbit.config.RabbitPropertiesAutoConfig;
 import org.reactivecommons.async.rabbit.config.props.AsyncPropsDomain;
@@ -51,8 +52,8 @@ public RabbitProperties defaultRabbitProperties(RabbitPropertiesAutoConfig prope
 
     @Bean
     @ConditionalOnMissingBean(UnroutableMessageHandler.class)
-    public UnroutableMessageHandler defaultUnroutableMessageHandler() {
-        return (result -> {
+    public UnroutableMessageHandler defaultUnroutableMessageHandler(UnroutableMessageNotifier notifier) {
+        UnroutableMessageHandler handler = (result -> {
             System.out.println("MENSAJE DEVUELTO: " + result.isReturned());
             OutboundMessage returned = result.getOutboundMessage();
             log.severe("Unroutable message: exchange=" + returned.getExchange()
@@ -62,6 +63,14 @@ public UnroutableMessageHandler defaultUnroutableMessageHandler() {
             );
             return Mono.empty();
         });
+        notifier.listenToUnroutableMessages(handler);
+        return handler;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean(UnroutableMessageNotifier.class)
+    public UnroutableMessageNotifier defaultUnroutableMessageNotifier() {
+        return new UnroutableMessageNotifier();
     }
 
     @Bean

starters/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/RabbitMQBrokerProviderFactoryTest.java

@@ -10,7 +10,7 @@
 import org.reactivecommons.async.commons.config.IBrokerConfigProps;
 import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
-import org.reactivecommons.async.rabbit.communications.UnroutableMessageHandler;
+import org.reactivecommons.async.rabbit.communications.UnroutableMessageNotifier;
 import org.reactivecommons.async.rabbit.config.RabbitProperties;
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.rabbit.config.props.BrokerConfigProps;
@@ -34,14 +34,14 @@ class RabbitMQBrokerProviderFactoryTest {
     @Mock
     private CustomReporter errorReporter;
     @Mock
-    private UnroutableMessageHandler unroutableMessageHandler;
+    private UnroutableMessageNotifier unroutableMessageNotifier;
 
     private BrokerProviderFactory<AsyncProps> providerFactory;
 
     @BeforeEach
     void setUp() {
         providerFactory = new RabbitMQBrokerProviderFactory(config, router, converter, meterRegistry, errorReporter,
-                RabbitMQDiscardProviderImpl::new, unroutableMessageHandler);
+                RabbitMQDiscardProviderImpl::new, unroutableMessageNotifier);
     }
 
     @Test

starters/async-rabbit-starter/src/test/java/org/reactivecommons/async/starter/impl/common/rabbit/RabbitMQConfigTest.java

@@ -8,6 +8,7 @@
 import org.reactivecommons.async.rabbit.RabbitMQFactory;
 import org.reactivecommons.async.rabbit.communications.MyOutboundMessage;
 import org.reactivecommons.async.rabbit.communications.UnroutableMessageHandler;
+import org.reactivecommons.async.rabbit.communications.UnroutableMessageNotifier;
 import org.reactivecommons.async.rabbit.config.props.AsyncPropsDomain;
 import org.reactivecommons.async.rabbit.converters.json.RabbitJacksonMessageConverter;
 import org.reactivecommons.async.starter.config.ConnectionManager;
@@ -43,6 +44,8 @@ class RabbitMQConfigTest {
 
     @Mock
     private MyOutboundMessage outboundMessageMock;
+    @Mock
+    private UnroutableMessageNotifier unroutableMessageNotifier;
 
     private RabbitMQConfig rabbitMQConfig;
 
@@ -66,7 +69,7 @@ void shouldHasManager() {
     @Test
     void shouldProcessAndLogWhenMessageIsReturned() {
         // Arrange
-        UnroutableMessageHandler handler = rabbitMQConfig.defaultUnroutableMessageHandler();
+        UnroutableMessageHandler handler = rabbitMQConfig.defaultUnroutableMessageHandler(unroutableMessageNotifier);
         when(resultMock.getOutboundMessage()).thenReturn(outboundMessageMock);
         when(outboundMessageMock.getExchange()).thenReturn("test.exchange");
         when(outboundMessageMock.getRoutingKey()).thenReturn("test.key");