Merge pull request #47 from juancgalvis/feature/unit-tests

Feature unit tests

Author: Daniel Bustamante Ospina

.github/workflows/main.yml

@@ -18,22 +18,26 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - name: Set up JDK 1.8
+        with:
+          fetch-depth: 0
+      - name: Set up JDK 11
         uses: actions/setup-java@v1
         with:
-          java-version: 1.8
+          java-version: 11
       - name: Grant execute permission for gradlew
         run: chmod +x gradlew
-      - name: Execute jacocoTestReport and Sonar
+      - name: Execute build test jacocoTestReport and sonar analysis
+        if: endsWith(github.REF, '/master') == true
         env:
-          SONAR_TOKEN: ${{secrets.SONAR_TOKEN}}
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: ./gradlew test jacocoTestReport sonarqube
-#        run: ./gradlew test jacocoTestReport && cp build/reports/jacoco/report.xml jacoco.xml || echo "Code coverage failed"
-      - name: Build with Gradle
-        run: ./gradlew build --refresh-dependencies --no-daemon --continue
-#      - name: Push codeCoverage to Codecov
-#        run: bash <(curl -s https://codecov.io/bash)
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+        run: ./gradlew build test jacocoTestReport sonarqube --refresh-dependencies --no-daemon --continue
+      - name: Execute build test jacocoTestReport pull request
+        if: endsWith(github.REF, '/merge') == true
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+        run: ./gradlew build test jacocoTestReport --refresh-dependencies --no-daemon --continue
 
   release:
     if: github.event_name == 'release'
@@ -54,6 +58,11 @@ jobs:
         uses: heinrichreimer/github-changelog-generator-action@v2.1.1
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
+          pullRequests: true
+          prWoLabels: true
+          issues: true
+          issuesWoLabels: true
+          stripGeneratorNotice: true
       - name: Push Docs and Changelog
         uses: github-actions-x/commit@v2.7
         with:

README.md

@@ -18,7 +18,7 @@ To include all (API and implementation) (Spring boot Starter):
 ```groovy
 
     dependencies {
-      compile 'org.reactivecommons:async-commons-starter:0.6.2'
+      compile 'org.reactivecommons:async-commons-starter:<version-here>'
     }
 
     //IMPORTANT! if you use the version 0.6.x
@@ -54,15 +54,15 @@ To include only domain events API:
 
 ```groovy
     dependencies {
-      compile 'org.reactivecommons:domain-events-api:0.6.2'
+      compile 'org.reactivecommons:domain-events-api:<version-here>'
     }
 ```
 
 To include only async commons API:
 
 ```groovy
     dependencies {
-      compile 'org.reactivecommons:async-commons-api:0.6.2'
+      compile 'org.reactivecommons:async-commons-api:<version-here>'
     }
 ```
 
@@ -150,7 +150,7 @@ Don't forget to add the starter bundle to the main spring boot module (applicati
 
 ```groovy
     dependencies {
-      compile 'org.reactivecommons:async-commons-starter:0.6.2'
+      compile 'org.reactivecommons:async-commons-starter:<version-here>'
     }
 ```
 
@@ -159,7 +159,7 @@ Or add the implementation dependency if for any reason you don't want to use the
 
 ```groovy
     dependencies {
-      compile 'org.reactivecommons:async-commons:0.6.2'
+      compile 'org.reactivecommons:async-commons:<version-here>'
     }
 ```
 

async/async-commons/src/test/java/org/reactivecommons/async/impl/RabbitDirectAsyncGatewayTest.java

@@ -1,11 +1,17 @@
 package org.reactivecommons.async.impl;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.Data;
-import org.assertj.core.api.Assertions;
-import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
 import org.reactivecommons.api.domain.Command;
+import org.reactivecommons.async.api.AsyncQuery;
+import org.reactivecommons.async.api.From;
+import org.reactivecommons.async.impl.communications.Message;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
 import org.reactivecommons.async.impl.config.BrokerConfig;
 import org.reactivecommons.async.impl.converters.MessageConverter;
@@ -14,64 +20,156 @@
 import org.reactivecommons.async.impl.reply.ReactiveReplyRouter;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.UnicastProcessor;
 import reactor.rabbitmq.OutboundMessage;
 import reactor.rabbitmq.OutboundMessageResult;
 import reactor.rabbitmq.Sender;
+import reactor.test.StepVerifier;
+import reactor.util.concurrent.Queues;
 
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+import static org.reactivecommons.async.impl.Headers.*;
 
+
+@RunWith(MockitoJUnitRunner.class)
 public class RabbitDirectAsyncGatewayTest {
 
     private final BrokerConfig config = new BrokerConfig();
-
+    private final Semaphore semaphore = new Semaphore(0);
+    private final MessageConverter converter = new JacksonMessageConverter(new DefaultObjectMapperSupplier().get());
     @Mock
     private ReactiveReplyRouter router;
-
     @Mock
-    private MessageConverter converter;
-
+    private ReactiveMessageSender senderMock;
     private RabbitDirectAsyncGateway asyncGateway;
-    private final Semaphore semaphore = new Semaphore(0);
 
-    @Before
-    public void init() {
-        ReactiveMessageSender sender = getReactiveMessageSender();
+    public void init(ReactiveMessageSender sender) {
         asyncGateway = new RabbitDirectAsyncGateway(config, router, sender, "exchange", converter);
     }
 
     @Test
     public void shouldSendInOptimalTime() throws InterruptedException {
+        init(getReactiveMessageSender());
+
         int messageCount = 40000;
         final Flux<Command<DummyMessage>> messages = createMessagesHot(messageCount);
-        final Flux<Void> target = messages.flatMap(dummyMessageCommand -> asyncGateway.sendCommand(dummyMessageCommand, "testTarget")
-            .doOnSuccess(aVoid -> semaphore.release()));
-
+        final Flux<Void> target =
+                messages.flatMap(dummyMessageCommand -> asyncGateway.sendCommand(dummyMessageCommand, "testTarget")
+                        .doOnSuccess(aVoid -> semaphore.release()));
 
         final long init = System.currentTimeMillis();
         target.subscribe();
         semaphore.acquire(messageCount);
         final long end = System.currentTimeMillis();
 
         final long total = end - init;
-        final double microsPerMessage =  ((total+0.0)/messageCount)*1000;
+        final double microsPerMessage = ((total + 0.0) / messageCount) * 1000;
         System.out.println("Message count: " + messageCount);
         System.out.println("Total Execution Time: " + total + "ms");
         System.out.println("Microseconds per message: " + microsPerMessage + "us");
-        Assertions.assertThat(microsPerMessage).isLessThan(150);
+        assertThat(microsPerMessage).isLessThan(150);
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldReplyQuery() {
+        // Arrange
+        senderMock();
+
+        From from = new From();
+        from.setReplyID("replyId");
+        from.setCorrelationID("correlationId");
+        DummyMessage response = new DummyMessage();
+        // Act
+        Mono<Void> result = asyncGateway.reply(response, from);
+        // Assert
+        StepVerifier.create(result).verifyComplete();
+        ArgumentCaptor<Map<String, Object>> headersCaptor = ArgumentCaptor.forClass(Map.class);
+        verify(senderMock, times(1))
+                .sendNoConfirm(eq(response), eq("globalReply"), eq("replyId"), headersCaptor.capture(), anyBoolean());
+        assertThat(headersCaptor.getValue().get(CORRELATION_ID)).isEqualTo("correlationId");
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldReplyQueryWithout() {
+        // Arrange
+        senderMock();
+
+        From from = new From();
+        from.setReplyID("replyId");
+        from.setCorrelationID("correlationId");
+        // Act
+        Mono<Void> result = asyncGateway.reply(null, from);
+        // Assert
+        StepVerifier.create(result).verifyComplete();
+        ArgumentCaptor<Map<String, Object>> headersCaptor = ArgumentCaptor.forClass(Map.class);
+        verify(senderMock, times(1))
+                .sendNoConfirm(eq(null), eq("globalReply"), eq("replyId"), headersCaptor.capture(), anyBoolean());
+        assertThat(headersCaptor.getValue().get(CORRELATION_ID)).isEqualTo("correlationId");
+        assertThat(headersCaptor.getValue().get(COMPLETION_ONLY_SIGNAL)).isEqualTo(Boolean.TRUE.toString());
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void shouldHandleRequestReply() throws JsonProcessingException {
+        // Arrange
+        senderMock();
+        mockReply();
+
+        String queryName = "my.query";
+        String targetName = "app-target";
+        AsyncQuery<DummyMessage> query = new AsyncQuery<>(queryName, new DummyMessage());
+        // Act
+        Mono<DummyMessage> result = asyncGateway.requestReply(query, targetName, DummyMessage.class);
+        // Assert
+        StepVerifier.create(result)
+                .assertNext(res -> assertThat(res.getName()).startsWith("Daniel"))
+                .verifyComplete();
+        ArgumentCaptor<Map<String, Object>> headersCaptor = ArgumentCaptor.forClass(Map.class);
+        verify(senderMock, times(1))
+                .sendNoConfirm(eq(query), eq("exchange"), eq("app-target.query"), headersCaptor.capture(),
+                        anyBoolean());
+        assertThat(headersCaptor.getValue().get(REPLY_ID).toString().length()).isEqualTo(32);
+        assertThat(headersCaptor.getValue().get(CORRELATION_ID).toString().length()).isEqualTo(32);
+    }
+
+    private void senderMock() {
+        init(senderMock);
+        when(senderMock.sendNoConfirm(any(), anyString(), anyString(), anyMap(), anyBoolean()))
+                .thenReturn(Mono.empty());
+    }
+
+    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());
     }
 
     private ReactiveMessageSender getReactiveMessageSender() {
-        MessageConverter messageConverter = new JacksonMessageConverter(new DefaultObjectMapperSupplier().get());
         Sender sender = new StubSender();
-        ReactiveMessageSender reactiveSender = new ReactiveMessageSender(sender, "sourceApplication", messageConverter, null);
-        return reactiveSender;
+        return new ReactiveMessageSender(sender, "sourceApplication", converter, null);
+    }
+
+    private Flux<Command<DummyMessage>> createMessagesHot(int count) {
+        final List<Command<DummyMessage>> commands = IntStream.range(0, count).mapToObj(value -> new Command<>("app" +
+                ".command.test", UUID.randomUUID().toString(), new DummyMessage())).collect(Collectors.toList());
+        return Flux.fromIterable(commands);
     }
 
     static class StubSender extends Sender {
@@ -80,15 +178,14 @@ static class StubSender extends Sender {
         public <OMSG extends OutboundMessage> Flux<OutboundMessageResult<OMSG>> sendWithTypedPublishConfirms(Publisher<OMSG> messages) {
             return Flux.from(messages).map(omsg -> new OutboundMessageResult<>(omsg, true));
         }
-    }
-
 
-    private Flux<Command<DummyMessage>> createMessagesHot(int count) {
-        final List<Command<DummyMessage>> commands = IntStream.range(0, count).mapToObj(value -> new Command<>("app.command.test", UUID.randomUUID().toString(), new DummyMessage())).collect(Collectors.toList());
-        return Flux.fromIterable(commands);
+        @Override
+        @SuppressWarnings("rawtypes")
+        public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMessage> messages) {
+            return Flux.from(messages).map(omsg -> new OutboundMessageResult<>(omsg, true));
+        }
     }
 
-
     @Data
     static class DummyMessage {
         private String name = "Daniel" + ThreadLocalRandom.current().nextLong();

build.gradle

@@ -19,6 +19,7 @@ plugins {
 sonarqube {
     properties {
         property "sonar.projectKey", "reactive-commons_reactive-commons-java"
+        property "sonar.coverage.exclusions", "samples/**/*"
         property "sonar.organization", "reactive-commons"
         property "sonar.host.url", "https://sonarcloud.io"
     }