Merge pull request #82 from reactive-commons/feature/spring-boot-3

Daniel Bustamante Ospina · web-flow · commit 8fd09f5add00 · 2023-03-02T10:38:00.000-05:00
Feature/spring boot 3
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
@@ -15,15 +15,16 @@ on:
 jobs:
   build:
     if: github.event_name != 'release'
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
         with:
           fetch-depth: 0
-      - name: Set up JDK 11
-        uses: actions/setup-java@v1
+      - name: Set up JDK 17
+        uses: actions/setup-java@v3
         with:
-          java-version: 11
+          distribution: temurin
+          java-version: 17
       - name: Grant execute permission for gradlew
         run: chmod +x gradlew
       - name: Execute build test jacocoTestReport and sonar analysis
@@ -41,13 +42,14 @@ jobs:
 
   release:
     if: github.event_name == 'release'
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
-      - name: Set up JDK 1.8
-        uses: actions/setup-java@v1
+      - uses: actions/checkout@v3
+      - name: Set up JDK 17
+        uses: actions/setup-java@v3
         with:
-          java-version: 1.8
+          distribution: temurin
+          java-version: 17
       - name: Grant execute permission for gradlew
         run: chmod +x gradlew
       #      - name: Execute jacocoTestReport
diff --git a/acceptance/async-tests/async-tests.gradle b/acceptance/async-tests/async-tests.gradle
@@ -1,6 +1,6 @@
 dependencies {
-    compile project(":async-commons-rabbit-starter")
-    compile('org.springframework.boot:spring-boot-starter')
+    implementation project(':async-commons-rabbit-starter')
+    implementation 'org.springframework.boot:spring-boot-starter'
 }
 
 test.onlyIf { false }
diff --git a/async/async-commons/async-commons.gradle b/async/async-commons/async-commons.gradle
@@ -8,7 +8,7 @@ dependencies {
     api project(':domain-events-api')
 
     compileOnly 'io.projectreactor:reactor-core'
-    api 'io.projectreactor.rabbitmq:reactor-rabbitmq'
+    api "io.projectreactor.rabbitmq:reactor-rabbitmq:${reactorRabbitVersion}"
     api 'com.fasterxml.jackson.core:jackson-databind'
     implementation 'commons-io:commons-io:2.11.0'
 
diff --git a/async/async-commons/src/main/java/org/reactivecommons/async/commons/reply/ReactiveReplyRouter.java b/async/async-commons/src/main/java/org/reactivecommons/async/commons/reply/ReactiveReplyRouter.java
@@ -2,36 +2,34 @@
 
 import org.reactivecommons.async.commons.communications.Message;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.UnicastProcessor;
-import reactor.util.concurrent.Queues;
+import reactor.core.publisher.Sinks;
 
 import java.util.concurrent.ConcurrentHashMap;
 
 public class ReactiveReplyRouter {
-    private final ConcurrentHashMap<String, UnicastProcessor<Message>> processors = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, Sinks.One<Message>> processors = new ConcurrentHashMap<>();
 
     public Mono<Message> register(String correlationID) {
-        final UnicastProcessor<Message> processor = UnicastProcessor.create(Queues.<Message>one().get());
+        final Sinks.One<Message> processor = Sinks.one();
         processors.put(correlationID, processor);
-        return processor.singleOrEmpty();
+        return processor.asMono();
     }
 
     public void routeReply(String correlationID, Message data) {
-        final UnicastProcessor<Message> processor = processors.remove(correlationID);
+        final Sinks.One<Message> processor = processors.remove(correlationID);
         if (processor != null) {
-            processor.onNext(data);
-            processor.onComplete();
+            processor.tryEmitValue(data);
         }
     }
 
-    public void deregister(String correlationID){
+    public void deregister(String correlationID) {
         processors.remove(correlationID);
     }
 
     public void routeEmpty(String correlationID) {
-        final UnicastProcessor<Message> processor = processors.remove(correlationID);
+        final Sinks.One<Message> processor = processors.remove(correlationID);
         if (processor != null) {
-            processor.onComplete();
+            processor.tryEmitEmpty();
         }
     }
 }
diff --git a/async/async-commons/src/test/java/org/reactivecommons/async/commons/reply/ReactiveReplyRouterTest.java b/async/async-commons/src/test/java/org/reactivecommons/async/commons/reply/ReactiveReplyRouterTest.java
@@ -11,7 +11,7 @@
 
 class ReactiveReplyRouterTest {
 
-    private ReactiveReplyRouter replyRouter = new ReactiveReplyRouter();
+    private final ReactiveReplyRouter replyRouter = new ReactiveReplyRouter();
 
     @Test
     void shouldRouteReply(){
@@ -50,4 +50,4 @@ void shouldDeRegisterProcessor(){
             .expectTimeout(Duration.ofSeconds(3)).verify();
     }
 
-}
+}
diff --git a/async/async-rabbit-standalone/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java b/async/async-rabbit-standalone/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java
@@ -1,5 +1,6 @@
 package org.reactivecommons.async.rabbit.config;
 
+import io.micrometer.core.instrument.MeterRegistry;
 import org.reactivecommons.async.rabbit.RabbitDirectAsyncGateway;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
@@ -22,8 +23,9 @@ public DirectAsyncGatewayConfig(String directMessagesExchangeName, String appNam
         this.appName = appName;
     }
 
-    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter) throws Exception {
-        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName, converter);
+    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter,
+                                                             MeterRegistry meterRegistry) throws Exception {
+        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName, converter, meterRegistry);
     }
 
     public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ReactiveMessageListener listener)  {
diff --git a/async/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java b/async/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java
@@ -1,14 +1,17 @@
 package org.reactivecommons.async.rabbit.config;
 
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import lombok.RequiredArgsConstructor;
+import org.reactivecommons.async.commons.config.BrokerConfig;
+import org.reactivecommons.async.commons.converters.MessageConverter;
+import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.rabbit.RabbitDirectAsyncGateway;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.config.props.BrokerConfigProps;
-import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.rabbit.listeners.ApplicationReplyListener;
-import org.reactivecommons.async.commons.config.BrokerConfig;
-import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
@@ -21,22 +24,25 @@ public class DirectAsyncGatewayConfig {
     private final BrokerConfigProps props;
 
     @Bean
-    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter) throws Exception {
-        return new RabbitDirectAsyncGateway(config, router, rSender, props.getDirectMessagesExchangeName(), converter);
+    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter, MeterRegistry meterRegistry) throws Exception {
+        return new RabbitDirectAsyncGateway(config, router, rSender, props.getDirectMessagesExchangeName(), converter, meterRegistry);
     }
 
     @Bean
-    public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ReactiveMessageListener listener)  {
+    public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ReactiveMessageListener listener) {
         final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, listener, props.getReplyQueue());
         replyListener.startListening(config.getRoutingKey());
         return replyListener;
     }
 
-
     @Bean
     public ReactiveReplyRouter router() {
         return new ReactiveReplyRouter();
     }
 
-
+    @Bean
+    @ConditionalOnMissingBean(MeterRegistry.class)
+    public MeterRegistry defaultRabbitMeterRegistry() {
+        return new SimpleMeterRegistry();
+    }
 }
diff --git a/async/async-rabbit/async-rabbit.gradle b/async/async-rabbit/async-rabbit.gradle
@@ -4,11 +4,12 @@ ext {
 }
 
 dependencies {
-    api project(":async-commons-api")
-    api project(":domain-events-api")
-    api project(":async-commons")
+    api project(':async-commons-api')
+    api project(':domain-events-api')
+    api project(':async-commons')
 
     api 'io.projectreactor:reactor-core'
+    api 'io.projectreactor:reactor-core-micrometer'
     api 'io.projectreactor.rabbitmq:reactor-rabbitmq'
     api 'com.rabbitmq:amqp-client'
     api 'com.fasterxml.jackson.core:jackson-databind'
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGateway.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGateway.java
@@ -1,5 +1,6 @@
 package org.reactivecommons.async.rabbit;
 
+import io.micrometer.core.instrument.MeterRegistry;
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.api.DirectAsyncGateway;
@@ -8,6 +9,7 @@
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
+import reactor.core.observability.micrometer.Micrometer;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.rabbitmq.OutboundMessageResult;
@@ -33,10 +35,11 @@ public class RabbitDirectAsyncGateway implements DirectAsyncGateway {
     private final boolean persistentCommands;
     private final boolean persistentQueries;
     private final Duration replyTimeout;
+    private final MeterRegistry meterRegistry;
 
 
     public RabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender sender,
-                                    String exchange, MessageConverter converter) {
+                                    String exchange, MessageConverter converter, MeterRegistry meterRegistry) {
         this.config = config;
         this.router = router;
         this.sender = sender;
@@ -45,6 +48,7 @@ public RabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router,
         this.persistentCommands = config.isPersistentCommands();
         this.persistentQueries = config.isPersistentQueries();
         this.replyTimeout = config.getReplyTimeout();
+        this.meterRegistry = meterRegistry;
     }
 
     @Override
@@ -77,7 +81,7 @@ public <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class
                 .name("async_query")
                 .tag("operation", query.getResource())
                 .tag("target", targetName)
-                .metrics();
+                .tap(Micrometer.metrics(meterRegistry));
     }
 
     @Override
diff --git a/async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGatewayTest.java b/async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGatewayTest.java
@@ -2,6 +2,8 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import lombok.Data;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -12,22 +14,20 @@
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.api.From;
 import org.reactivecommons.async.commons.communications.Message;
-import org.reactivecommons.async.rabbit.RabbitDirectAsyncGateway;
-import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.commons.config.BrokerConfig;
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.converters.json.DefaultObjectMapperSupplier;
-import org.reactivecommons.async.rabbit.converters.json.JacksonMessageConverter;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
+import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
+import org.reactivecommons.async.rabbit.converters.json.JacksonMessageConverter;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.UnicastProcessor;
+import reactor.core.publisher.Sinks;
 import reactor.rabbitmq.OutboundMessage;
 import reactor.rabbitmq.OutboundMessageResult;
 import reactor.rabbitmq.Sender;
 import reactor.test.StepVerifier;
-import reactor.util.concurrent.Queues;
 
 import java.time.Duration;
 import java.util.List;
@@ -40,9 +40,18 @@
 import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.*;
-import static org.mockito.Mockito.*;
-import static org.reactivecommons.async.commons.Headers.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.reactivecommons.async.commons.Headers.COMPLETION_ONLY_SIGNAL;
+import static org.reactivecommons.async.commons.Headers.CORRELATION_ID;
+import static org.reactivecommons.async.commons.Headers.REPLY_ID;
 
 
 @ExtendWith(MockitoExtension.class)
@@ -55,16 +64,18 @@ class RabbitDirectAsyncGatewayTest {
     private ReactiveReplyRouter router;
     @Mock
     private ReactiveMessageSender senderMock;
+
+    private final MeterRegistry meterRegistry = new SimpleMeterRegistry();
     private RabbitDirectAsyncGateway asyncGateway;
 
     public void init(ReactiveMessageSender sender) {
-        asyncGateway = new RabbitDirectAsyncGateway(config, router, sender, "exchange", converter);
+        asyncGateway = new RabbitDirectAsyncGateway(config, router, sender, "exchange", converter, meterRegistry);
     }
 
     @Test
     void shouldReleaseRouterResourcesOnTimeout() {
         BrokerConfig config = new BrokerConfig(false, false, false, Duration.ofSeconds(1));
-        asyncGateway = new RabbitDirectAsyncGateway(config, router, senderMock, "ex", converter);
+        asyncGateway = new RabbitDirectAsyncGateway(config, router, senderMock, "ex", converter, meterRegistry);
         when(router.register(anyString())).thenReturn(Mono.never());
         when(senderMock.sendNoConfirm(any(), anyString(), anyString(), anyMap(), anyBoolean()))
                 .thenReturn(Mono.empty());
@@ -179,10 +190,9 @@ private void mockReply() throws JsonProcessingException {
         Message message = mock(Message.class);
         ObjectMapper mapper = new ObjectMapper();
         when(message.getBody()).thenReturn(mapper.writeValueAsString(new DummyMessage()).getBytes());
-        final UnicastProcessor<Message> processor = UnicastProcessor.create(Queues.<Message>one().get());
-        processor.onNext(message);
-        processor.onComplete();
-        when(router.register(anyString())).thenReturn(processor.singleOrEmpty());
+        final Sinks.One<Message> processor = Sinks.one();
+        processor.tryEmitValue(message);
+        when(router.register(anyString())).thenReturn(processor.asMono());
     }
 
     private ReactiveMessageSender getReactiveMessageSender() {
diff --git a/build.gradle b/build.gradle
@@ -6,7 +6,6 @@ buildscript {
     }
 
     dependencies {
-        classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
         classpath("com.github.ben-manes:gradle-versions-plugin:$gradleVersionsVersion")
     }
 }
@@ -15,6 +14,7 @@ plugins {
     id 'jacoco'
     id 'org.sonarqube' version '3.0'
     id 'org.asciidoctor.jvm.convert' version '3.1.0'
+    id 'org.springframework.boot' version "${springBootVersion}" apply false
 }
 
 sonarqube {
diff --git a/domain/domain-events/domain-events-api.gradle b/domain/domain-events/domain-events-api.gradle
@@ -4,5 +4,5 @@ ext {
 }
 
 dependencies {
-    api 'org.reactivestreams:reactive-streams:1.0.3'
+    api 'org.reactivestreams:reactive-streams:1.0.4'
 }
diff --git a/gradle.properties b/gradle.properties
@@ -1,4 +1,5 @@
-version=1.0.5
-springBootVersion=2.6.6
+version=2.0.0
+springBootVersion=3.0.2
+reactorRabbitVersion=1.5.5
 gradleVersionsVersion=0.36.0
 toPublish=async-commons,async-commons-api,async-commons-rabbit-standalone,async-commons-rabbit-starter,domain-events-api,async-rabbit
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.8.2-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.6-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/main.gradle b/main.gradle
diff --git a/samples/async/no-springboot-client/no-springboot-client.gradle b/samples/async/no-springboot-client/no-springboot-client.gradle
diff --git a/samples/async/receiver-responder/async-receiver-sample.gradle b/samples/async/receiver-responder/async-receiver-sample.gradle
diff --git a/samples/async/sender-client/async-sender-client.gradle b/samples/async/sender-client/async-sender-client.gradle
diff --git a/samples/async/sender-client/src/main/resources/application.properties b/samples/async/sender-client/src/main/resources/application.properties
diff --git a/samples/async/simpleConsumer/simple-consumer.gradle b/samples/async/simpleConsumer/simple-consumer.gradle

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`dependencies {`
`2`		`- compile project(":async-commons-rabbit-starter")`
`3`		`- compile('org.springframework.boot:spring-boot-starter')`
	`2`	`+ implementation project(':async-commons-rabbit-starter')`
	`3`	`+ implementation 'org.springframework.boot:spring-boot-starter'`
`4`	`4`	`}`
`5`	`5`
`6`	`6`	`test.onlyIf { false }`
Original file line number	Diff line number	Diff line change
`@@ -2,36 +2,34 @@`
`2`	`2`
`3`	`3`	`import org.reactivecommons.async.commons.communications.Message;`
`4`	`4`	`import reactor.core.publisher.Mono;`
`5`		`-import reactor.core.publisher.UnicastProcessor;`
`6`		`-import reactor.util.concurrent.Queues;`
	`5`	`+import reactor.core.publisher.Sinks;`
`7`	`6`
`8`	`7`	`import java.util.concurrent.ConcurrentHashMap;`
`9`	`8`
`10`	`9`	`public class ReactiveReplyRouter {`
`11`		`- private final ConcurrentHashMap<String, UnicastProcessor<Message>> processors = new ConcurrentHashMap<>();`
	`10`	`+ private final ConcurrentHashMap<String, Sinks.One<Message>> processors = new ConcurrentHashMap<>();`
`12`	`11`
`13`	`12`	`public Mono<Message> register(String correlationID) {`
`14`		`- final UnicastProcessor<Message> processor = UnicastProcessor.create(Queues.<Message>one().get());`
	`13`	`+ final Sinks.One<Message> processor = Sinks.one();`
`15`	`14`	`processors.put(correlationID, processor);`
`16`		`- return processor.singleOrEmpty();`
	`15`	`+ return processor.asMono();`
`17`	`16`	`}`
`18`	`17`
`19`	`18`	`public void routeReply(String correlationID, Message data) {`
`20`		`- final UnicastProcessor<Message> processor = processors.remove(correlationID);`
	`19`	`+ final Sinks.One<Message> processor = processors.remove(correlationID);`
`21`	`20`	`if (processor != null) {`
`22`		`- processor.onNext(data);`
`23`		`- processor.onComplete();`
	`21`	`+ processor.tryEmitValue(data);`
`24`	`22`	`}`
`25`	`23`	`}`
`26`	`24`
`27`		`- public void deregister(String correlationID){`
	`25`	`+ public void deregister(String correlationID) {`
`28`	`26`	`processors.remove(correlationID);`
`29`	`27`	`}`
`30`	`28`
`31`	`29`	`public void routeEmpty(String correlationID) {`
`32`		`- final UnicastProcessor<Message> processor = processors.remove(correlationID);`
	`30`	`+ final Sinks.One<Message> processor = processors.remove(correlationID);`
`33`	`31`	`if (processor != null) {`
`34`		`- processor.onComplete();`
	`32`	`+ processor.tryEmitEmpty();`
`35`	`33`	`}`
`36`	`34`	`}`
`37`	`35`	`}`