Merge pull request #67 from juanpmarin/fix/propagate-empty-responses

fix: propagate empty response

Author: Juan Marín

async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/ApplicationQueryListener.java

@@ -24,6 +24,7 @@
 import java.util.function.Function;
 import java.util.logging.Level;
 
+import static java.lang.Boolean.TRUE;
 import static java.util.Optional.ofNullable;
 import static org.reactivecommons.async.commons.Headers.*;
 
@@ -138,15 +139,16 @@ protected Function<Mono<Object>, Mono<Object>> enrichPostProcess(Message msg) {
             if (signal.isOnError()) {
                 return Mono.error(ofNullable(signal.getThrowable()).orElseGet(RuntimeException::new));
             }
-            if (signal.isOnComplete()) {
-                return Mono.empty();
-            }
 
             final String replyID = msg.getProperties().getHeaders().get(REPLY_ID).toString();
             final String correlationID = msg.getProperties().getHeaders().get(CORRELATION_ID).toString();
             final HashMap<String, Object> headers = new HashMap<>();
             headers.put(CORRELATION_ID, correlationID);
 
+            if (!signal.hasValue()) {
+                headers.put(COMPLETION_ONLY_SIGNAL, TRUE.toString());
+            }
+
             return sender.sendNoConfirm(signal.get(), replyExchange, replyID, headers, false);
         });
     }

async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ApplicationQueryListenerTest.java

@@ -8,6 +8,7 @@
 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.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.reactivecommons.async.api.AsyncQuery;
@@ -29,6 +30,7 @@
 import reactor.rabbitmq.AcknowledgableDelivery;
 import reactor.rabbitmq.Receiver;
 import reactor.test.StepVerifier;
+import reactor.test.publisher.PublisherProbe;
 
 import java.time.Instant;
 import java.util.Date;
@@ -37,6 +39,7 @@
 import java.util.Optional;
 import java.util.function.Function;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.*;
 import static org.reactivecommons.async.commons.Headers.*;
@@ -80,7 +83,7 @@ public void setUp() {
         QueryHandler<String, SampleClass> handler = (message) -> just("OK");
         handlers.put("queryDirect", new RegisteredQueryHandler<>("queryDirect",
                 (from, message) -> handler.handle(message), SampleClass.class));
-        HandlerResolver resolver = new HandlerResolver(handlers, null,null, null, null);
+        HandlerResolver resolver = new HandlerResolver(handlers, null, null, null, null);
         applicationQueryListener = new ApplicationQueryListener(reactiveMessageListener, "queue", resolver, sender,
                 "directExchange", messageConverter, "replyExchange", false,
                 1, 100, maxLengthBytes, true, discardNotifier, errorReporter);
@@ -118,15 +121,26 @@ void shouldHandleErrorWhenNoQueryHandler() {
     }
 
     @Test
-    void shouldNotRespondQueryEnrichPostProcess() {
-        Message message = spy(TestStubs.mockMessage());
-        Function<Mono<Object>, Mono<Object>> handler = applicationQueryListener.enrichPostProcess(message);
-        Mono<Object> result = handler.apply(empty());
+    @SuppressWarnings("unchecked")
+    void enrichPostProcessShouldPropagateEmptyResponses() {
+        Message message = TestStubs.mockMessage();
+        PublisherProbe<Void> publishProbe = PublisherProbe.empty();
+
+        ArgumentCaptor<Map<String, Object>> headersCaptor = ArgumentCaptor.forClass(Map.class);
+
+        when(sender.sendNoConfirm(any(), anyString(), anyString(), headersCaptor.capture(), anyBoolean()))
+                .thenReturn(publishProbe.mono());
+
+        Function<Mono<Object>, Mono<Object>> transformer = applicationQueryListener.enrichPostProcess(message);
+        Mono<Object> result = transformer.apply(empty());
 
         StepVerifier.create(result)
                 .verifyComplete();
 
-        verify(message, times(0)).getProperties();
+        publishProbe.assertWasSubscribed();
+
+        assertThat(headersCaptor.getValue().get(COMPLETION_ONLY_SIGNAL))
+                .isEqualTo("true");
     }
 
     @Test