feat: implement UnroutableMessageProcessor and refactor message handling

Author: lugomez29

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

@@ -10,7 +10,6 @@
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
-import reactor.core.scheduler.Schedulers;
 import reactor.rabbitmq.OutboundMessage;
 import reactor.rabbitmq.OutboundMessageResult;
 import reactor.rabbitmq.SendOptions;
@@ -62,10 +61,6 @@ public ReactiveMessageSender(Sender sender, String sourceApplication,
         this.topologyCreator = topologyCreator;
         this.isMandatory = isMandatory;
         this.unroutableMessageNotifier = unroutableMessageNotifier;
-
-        System.out.println("ReactiveMessageSender initialized with mandatory: " + isMandatory);
-        System.out.println("onReturnedCallback: " + unroutableMessageNotifier);
-
         initializeSenders();
     }
 
@@ -74,9 +69,7 @@ private void initializeSenders() {
             final Flux<MyOutboundMessage> messageSource = Flux.create(fluxSinkConfirm::add);
             sender.sendWithTypedPublishConfirms(messageSource, new SendOptions().trackReturned(isMandatory))
                     .doOnNext((OutboundMessageResult<MyOutboundMessage> outboundMessageResult) -> {
-                        System.out.println("MANDATORY: " + isMandatory);
                         if (outboundMessageResult.isReturned()) {
-                            System.out.println("CALLBACK: " + unroutableMessageNotifier);
                             this.unroutableMessageNotifier.notifyUnroutableMessage(outboundMessageResult);
                         }
                         final Consumer<Boolean> ackNotifier =
@@ -121,12 +114,6 @@ public <T> Flux<OutboundMessageResult> sendWithConfirmBatch(Flux<T> messages, St
                 );
     }
 
-    public Mono<Void> sendMessage(Object message, String exchange, String routingKey, Map<String, Object> headers) {
-        return sendNoConfirm(message, exchange, routingKey, headers, true)
-                .subscribeOn(Schedulers.boundedElastic())
-                .doOnError(e -> log.severe("Failed to send unroutable message: " + e.getMessage()));
-    }
-
     private record AckNotifier(MonoSink<Void> monoSink) implements Consumer<Boolean> {
 
         @Override

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

@@ -1,13 +1,15 @@
 package org.reactivecommons.async.rabbit.communications;
 
 import lombok.extern.java.Log;
+import reactor.core.Disposable;
 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;
+    private volatile Disposable currentSubscription;
 
     public UnroutableMessageNotifier() {
         this.sink = Sinks.many().multicast().onBackpressureBuffer();
@@ -20,13 +22,15 @@ public void notifyUnroutableMessage(OutboundMessageResult<MyOutboundMessage> mes
     }
 
     public void listenToUnroutableMessages(UnroutableMessageHandler handler) {
-        sink.asFlux()
-                .flatMap(handler::processMessage)
+        if (currentSubscription != null && !currentSubscription.isDisposed()) {
+            currentSubscription.dispose();
+        }
+        currentSubscription = sink.asFlux()
                 .subscribeOn(Schedulers.boundedElastic())
+                .flatMap(handler::processMessage)
                 .onErrorContinue((throwable, o) ->
                         log.severe("Error processing unroutable message: " + throwable.getMessage())
                 )
                 .subscribe();
-
     }
 }

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

@@ -0,0 +1,25 @@
+package org.reactivecommons.async.rabbit.communications;
+
+import lombok.NoArgsConstructor;
+import lombok.extern.java.Log;
+import reactor.core.publisher.Mono;
+import reactor.rabbitmq.OutboundMessageResult;
+
+import java.nio.charset.StandardCharsets;
+
+@Log
+@NoArgsConstructor
+public class UnroutableMessageProcessor implements UnroutableMessageHandler {
+
+
+    @Override
+    public Mono<Void> processMessage(OutboundMessageResult<MyOutboundMessage> result) {
+        var outboundMessage = result.getOutboundMessage();
+        log.severe("Unroutable message: exchange=" + outboundMessage.getExchange()
+                + ", routingKey=" + outboundMessage.getRoutingKey()
+                + ", body=" + new String(outboundMessage.getBody(), StandardCharsets.UTF_8)
+                + ", properties=" + outboundMessage.getProperties()
+        );
+        return Mono.empty();
+    }
+}

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

@@ -27,8 +27,8 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-@ExtendWith(MockitoExtension.class)
 @SuppressWarnings("unchecked")
+@ExtendWith(MockitoExtension.class)
 class ReactiveMessageSenderTest {
 
     private ReactiveMessageSender messageSender;
@@ -85,19 +85,6 @@ void shouldCallUnroutableMessageHandlerWhenMessageIsReturned() {
                 .notifyUnroutableMessage(any(OutboundMessageResult.class));
     }
 
-
-    @Test
-    void shouldSendMessageSuccessfully() {
-        Object message = new SomeClass("id", "name", new Date());
-        String exchange = "test.exchange";
-        String routingKey = "test.routingKey";
-        Map<String, Object> headers = new HashMap<>();
-
-        Mono<Void> result = messageSender.sendMessage(message, exchange, routingKey, headers);
-
-        StepVerifier.create(result).verifyComplete();
-    }
-
     @Test
     void shouldSendBatchWithConfirmSuccessfully() {
         Flux<SomeClass> messages = Flux.just(

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

@@ -0,0 +1,117 @@
+package org.reactivecommons.async.rabbit.communications;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.rabbitmq.OutboundMessageResult;
+
+import java.lang.reflect.Field;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@SuppressWarnings("unchecked")
+class UnroutableMessageNotifierTest {
+
+    @Spy
+    private UnroutableMessageNotifier unroutableMessageNotifier;
+
+    @Mock
+    private Sinks.Many<OutboundMessageResult<MyOutboundMessage>> sink;
+
+    @Mock
+    private OutboundMessageResult<MyOutboundMessage> messageResult;
+
+    @Mock
+    private UnroutableMessageHandler handler;
+
+    @Captor
+    private ArgumentCaptor<OutboundMessageResult<MyOutboundMessage>> messageCaptor;
+
+    @BeforeEach
+    void setUp() {
+        // Usar el constructor por defecto y espiar el sink interno
+        unroutableMessageNotifier = new UnroutableMessageNotifier();
+        // Inyectar el mock del sink usando un spy para poder verificarlo
+        try {
+            java.lang.reflect.Field sinkField = UnroutableMessageNotifier.class.getDeclaredField("sink");
+            sinkField.setAccessible(true);
+            sinkField.set(unroutableMessageNotifier, sink);
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Test
+    void shouldEmitUnroutableMessageSuccessfully() {
+        when(sink.tryEmitNext(messageResult)).thenReturn(Sinks.EmitResult.OK);
+
+        unroutableMessageNotifier.notifyUnroutableMessage(messageResult);
+
+        verify(sink).tryEmitNext(messageResult);
+    }
+
+    @Test
+    void shouldNotThrowWhenEmissionFails() {
+        when(sink.tryEmitNext(messageResult)).thenReturn(Sinks.EmitResult.FAIL_NON_SERIALIZED);
+
+        unroutableMessageNotifier.notifyUnroutableMessage(messageResult);
+
+        verify(sink).tryEmitNext(messageResult);
+    }
+
+    @Test
+    void shouldSubscribeHandlerToFluxOfMessages() {
+        final Flux<OutboundMessageResult<MyOutboundMessage>> messageResultFlux = Flux.just(messageResult);
+        when(sink.asFlux()).thenReturn(messageResultFlux);
+        when(handler.processMessage(any(OutboundMessageResult.class))).thenReturn(Mono.empty());
+
+        unroutableMessageNotifier.listenToUnroutableMessages(handler);
+
+        verify(handler, timeout(1000)).processMessage(messageCaptor.capture());
+        assertThat(messageCaptor.getValue()).isEqualTo(messageResult);
+    }
+
+    @Test
+    void shouldDisposePreviousSubscriptionWhenNewHandlerIsSubscribed() throws Exception {
+        Disposable firstSubscription = mock(Disposable.class);
+        when(firstSubscription.isDisposed()).thenReturn(false);
+
+        Field field = UnroutableMessageNotifier.class.getDeclaredField("currentSubscription");
+        field.setAccessible(true);
+        field.set(unroutableMessageNotifier, firstSubscription);
+
+        when(sink.asFlux()).thenReturn(Flux.never());
+
+        unroutableMessageNotifier.listenToUnroutableMessages(handler);
+
+        verify(firstSubscription).dispose();
+    }
+
+    @Test
+    void shouldContinueProcessingWhenHandlerFails() {
+        OutboundMessageResult<MyOutboundMessage> messageResult2 = mock(OutboundMessageResult.class);
+        when(sink.asFlux()).thenReturn(Flux.just(messageResult, messageResult2));
+        when(handler.processMessage(messageResult)).thenReturn(Mono.error(new RuntimeException("Processing Error")));
+        when(handler.processMessage(messageResult2)).thenReturn(Mono.empty());
+
+        unroutableMessageNotifier.listenToUnroutableMessages(handler);
+
+        verify(handler, timeout(1000).times(1)).processMessage(messageResult);
+        verify(handler, timeout(1000).times(1)).processMessage(messageResult2);
+    }
+}

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

@@ -0,0 +1,45 @@
+package org.reactivecommons.async.rabbit.communications;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.rabbitmq.OutboundMessageResult;
+import reactor.test.StepVerifier;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class UnroutableMessageProcessorTest {
+
+    @Mock
+    private OutboundMessageResult<MyOutboundMessage> messageResult;
+
+    @Mock
+    private MyOutboundMessage myOutboundMessage;
+
+    @InjectMocks
+    private UnroutableMessageProcessor unroutableMessageProcessor;
+
+
+    @Test
+    void logsUnroutableMessageDetails() {
+        when(messageResult.getOutboundMessage()).thenReturn(myOutboundMessage);
+        when(myOutboundMessage.getExchange()).thenReturn("test-exchange");
+        when(myOutboundMessage.getRoutingKey()).thenReturn("test-routingKey");
+        when(myOutboundMessage.getBody()).thenReturn("test-body".getBytes(StandardCharsets.UTF_8));
+        when(myOutboundMessage.getProperties()).thenReturn(null);
+
+        StepVerifier.create(unroutableMessageProcessor.processMessage(messageResult))
+                .verifyComplete();
+
+        verify(messageResult).getOutboundMessage();
+        verify(myOutboundMessage).getExchange();
+        verify(myOutboundMessage).getRoutingKey();
+        verify(myOutboundMessage).getBody();
+    }
+}

starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/broker/BrokerProvider.java

@@ -9,7 +9,7 @@
 
 @SuppressWarnings("rawtypes")
 public interface BrokerProvider<T extends GenericAsyncProps> {
-    T getProps();
+    T props();
 
     DomainEventBus getDomainBus();
 

starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/mybroker/MyBrokerProvider.java

@@ -17,7 +17,7 @@ public class MyBrokerProvider implements BrokerProvider<MyBrokerAsyncProps> {
     private final DiscardProvider discardProvider;
 
     @Override
-    public MyBrokerAsyncProps getProps() {
+    public MyBrokerAsyncProps props() {
         return null;
     }
 

starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/KafkaBrokerProvider.java

@@ -1,8 +1,6 @@
 package org.reactivecommons.async.kafka;
 
 import io.micrometer.core.instrument.MeterRegistry;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 import org.reactivecommons.api.domain.DomainEventBus;
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.commons.DiscardNotifier;
@@ -23,23 +21,19 @@
 import org.springframework.boot.ssl.SslBundles;
 import reactor.core.publisher.Mono;
 
-@Getter
-@RequiredArgsConstructor
-public class KafkaBrokerProvider implements BrokerProvider<AsyncKafkaProps> {
-    private final String domain;
-    private final AsyncKafkaProps props;
-    private final ReactiveReplyRouter router;
-    private final KafkaJacksonMessageConverter converter;
-    private final MeterRegistry meterRegistry;
-    private final CustomReporter errorReporter;
-    private final KafkaReactiveHealthIndicator healthIndicator;
-    private final ReactiveMessageListener receiver;
-    private final ReactiveMessageSender sender;
-    private final DiscardNotifier discardNotifier;
-    private final TopologyCreator topologyCreator;
-    private final KafkaCustomizations customizations;
-    private final SslBundles sslBundles;
-
+public record KafkaBrokerProvider(String domain,
+                                  AsyncKafkaProps props,
+                                  ReactiveReplyRouter router,
+                                  KafkaJacksonMessageConverter converter,
+                                  MeterRegistry meterRegistry,
+                                  CustomReporter errorReporter,
+                                  KafkaReactiveHealthIndicator healthIndicator,
+                                  ReactiveMessageListener receiver,
+                                  ReactiveMessageSender sender,
+                                  DiscardNotifier discardNotifier,
+                                  TopologyCreator topologyCreator,
+                                  KafkaCustomizations customizations,
+                                  SslBundles sslBundles) implements BrokerProvider<AsyncKafkaProps> {
     @Override
     public DomainEventBus getDomainBus() {
         return new KafkaDomainEventBus(sender);

starters/async-kafka-starter/src/test/java/org/reactivecommons/async/starter/impl/common/kafka/KafkaConfigTest.java

@@ -41,6 +41,6 @@ void shouldHasManager() {
         // Assert
         assertThat(manager).isNotNull();
         assertThat(manager.getProviders()).isNotEmpty();
-        assertThat(manager.getProviders().get("app").getProps().getAppName()).isEqualTo("async-kafka-starter");
+        assertThat(manager.getProviders().get("app").props().getAppName()).isEqualTo("async-kafka-starter");
     }
 }