Merge pull request #68 from reactive-commons/feature/upgrade-spring

Feature/upgrade spring

Author: Daniel Bustamante Ospina

async/async-rabbit/async-rabbit.gradle

@@ -8,8 +8,9 @@ dependencies {
     api project(":domain-events-api")
     api project(":async-commons")
 
-    compileOnly 'io.projectreactor:reactor-core'
-    api ("io.projectreactor.rabbitmq:reactor-rabbitmq")
+    api 'io.projectreactor:reactor-core'
+    api 'io.projectreactor.rabbitmq:reactor-rabbitmq'
+    api 'com.rabbitmq:amqp-client'
     api 'com.fasterxml.jackson.core:jackson-databind'
     testImplementation 'io.projectreactor:reactor-test'
 }

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

@@ -4,14 +4,14 @@
 import lombok.extern.java.Log;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
 import org.reactivecommons.async.commons.DiscardNotifier;
-import org.reactivecommons.async.rabbit.HandlerResolver;
-import org.reactivecommons.async.commons.communications.Message;
 import org.reactivecommons.async.commons.QueryExecutor;
+import org.reactivecommons.async.commons.communications.Message;
+import org.reactivecommons.async.commons.converters.MessageConverter;
+import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.rabbit.HandlerResolver;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
-import org.reactivecommons.async.commons.converters.MessageConverter;
-import org.reactivecommons.async.commons.ext.CustomReporter;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.AcknowledgableDelivery;
 import reactor.rabbitmq.BindingSpecification;
@@ -24,9 +24,11 @@
 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.*;
+import static org.reactivecommons.async.commons.Headers.CORRELATION_ID;
+import static org.reactivecommons.async.commons.Headers.REPLY_ID;
+import static org.reactivecommons.async.commons.Headers.REPLY_TIMEOUT_MILLIS;
+import static org.reactivecommons.async.commons.Headers.SERVED_QUERY_ID;
 
 @Log
 //TODO: Organizar inferencia de tipos de la misma forma que en comandos y eventos
@@ -139,16 +141,15 @@ 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

@@ -121,26 +121,15 @@ void shouldHandleErrorWhenNoQueryHandler() {
     }
 
     @Test
-    @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());
+    void shouldNotRespondQueryEnrichPostProcess() {
+        Message message = spy(TestStubs.mockMessage());
+        Function<Mono<Object>, Mono<Object>> handler = applicationQueryListener.enrichPostProcess(message);
+        Mono<Object> result = handler.apply(empty());
 
         StepVerifier.create(result)
                 .verifyComplete();
 
-        publishProbe.assertWasSubscribed();
-
-        assertThat(headersCaptor.getValue().get(COMPLETION_ONLY_SIGNAL))
-                .isEqualTo("true");
+        verify(message, times(0)).getProperties();
     }
 
     @Test

gradle.properties

@@ -1,4 +1,4 @@
 version=1.0.5
-springBootVersion=2.4.2
+springBootVersion=2.6.6
 gradleVersionsVersion=0.36.0
 toPublish=async-commons,async-commons-api,async-commons-rabbit-standalone,async-commons-rabbit-starter,domain-events-api,async-rabbit

samples/async/receiver-responder/src/main/java/sample/Config.java

@@ -0,0 +1,14 @@
+package sample;
+
+
+import com.rabbitmq.client.ConnectionFactory;
+import org.reactivecommons.async.rabbit.config.ConnectionFactoryProvider;
+import org.springframework.context.annotation.Bean;
+
+public class Config {
+
+    @Bean
+    public ConnectionFactoryProvider sampleBeanWhenProgrammaticConfiguration() {
+        return ConnectionFactory::new;
+    }
+}