Merge pull request #32 from juanpmarin/master

test: connection factory tests

Author: dbuos

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/RabbitMqConfig.java

@@ -32,6 +32,11 @@
 @Import(BrokerConfigProps.class)
 public class RabbitMqConfig {
 
+    private static final String LISTENER_TYPE = "listener";
+
+    private static final String SENDER_TYPE = "sender";
+
+
     @Value("${app.async.flux.maxConcurrency:250}")
     private Integer maxConcurrency;
 
@@ -41,14 +46,15 @@ public class RabbitMqConfig {
     @Bean
     public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter,
                                                BrokerConfigProps brokerConfigProps, RabbitProperties rabbitProperties) {
-        Mono<Connection> senderConnection = createSenderConnectionMono(provider.getConnectionFactory(), "sender");
-        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions();
+        final Mono<Connection> senderConnection =
+                createConnectionMono(provider.getConnectionFactory(), appName, SENDER_TYPE);
+        final ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions();
+        final PropertyMapper map = PropertyMapper.get();
 
-        PropertyMapper map = PropertyMapper.get();
         map.from(rabbitProperties.getCache().getChannel()::getSize).whenNonNull()
                 .to(channelPoolOptions::maxCacheSize);
 
-        ChannelPool channelPool = ChannelPoolFactory.createChannelPool(
+        final ChannelPool channelPool = ChannelPoolFactory.createChannelPool(
                 senderConnection,
                 channelPoolOptions
         );
@@ -60,8 +66,9 @@ public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, M
 
     @Bean
     public ReactiveMessageListener messageListener(ConnectionFactoryProvider provider) {
-        final Mono<Connection> connection = createSenderConnectionMono(provider.getConnectionFactory(), "listener");
-        Receiver receiver = RabbitFlux.createReceiver(new ReceiverOptions().connectionMono(connection));
+        final Mono<Connection> connection =
+                createConnectionMono(provider.getConnectionFactory(), appName, LISTENER_TYPE);
+        final Receiver receiver = RabbitFlux.createReceiver(new ReceiverOptions().connectionMono(connection));
         final Sender sender = RabbitFlux.createSender(new SenderOptions().connectionMono(connection));
 
         return new ReactiveMessageListener(receiver, new TopologyCreator(sender), maxConcurrency);
@@ -93,9 +100,9 @@ public MessageConverter messageConverter(ObjectMapperSupplier objectMapperSuppli
         return new JacksonMessageConverter(objectMapperSupplier.get());
     }
 
-    Mono<Connection> createSenderConnectionMono(ConnectionFactory factory, String name) {
+    Mono<Connection> createConnectionMono(ConnectionFactory factory, String connectionPrefix, String connectionType) {
         final Scheduler senderScheduler = Schedulers.elastic();
-        return Mono.fromCallable(() -> factory.newConnection(appName + " " + name))
+        return Mono.fromCallable(() -> factory.newConnection(connectionPrefix + " " + connectionType))
                 .doOnError(err ->
                         log.log(Level.SEVERE, "Error creating connection to RabbitMq Broker. Starting retry process...", err)
                 )

async/async-commons-starter/src/test/java/org/reactivecommons/async/impl/config/RabbitMqConfigTest.java

@@ -19,16 +19,20 @@ public class RabbitMqConfigTest {
 
     @Test
     public void retryInitialConnection() throws IOException, TimeoutException {
+        final String connectionType = "sender";
+        final String appName = "appName";
+        final String connectionName = "appName sender";
+
         final AtomicInteger count = new AtomicInteger();
         final Connection connection = mock(Connection.class);
         ConnectionFactory factory = mock(ConnectionFactory.class);
-        when(factory.newConnection("test")).thenAnswer(invocation -> {
+        when(factory.newConnection(connectionName)).thenAnswer(invocation -> {
             if(count.incrementAndGet() == 10){
                 return connection;
             }
             throw new RuntimeException();
         });
-        StepVerifier.withVirtualTime(() -> config.createSenderConnectionMono(factory, "test"))
+        StepVerifier.withVirtualTime(() -> config.createConnectionMono(factory, appName, connectionType))
             .thenAwait(Duration.ofMinutes(2))
             .expectNext(connection).verifyComplete();
     }

async/async-commons/async-commons.gradle

@@ -76,7 +76,7 @@ dependencies {
     compile project(":domain-events-api")
 
     api 'io.projectreactor:reactor-core'
-    api "io.projectreactor.rabbitmq:reactor-rabbitmq:1.4.0.RC1"
+    api "io.projectreactor.rabbitmq:reactor-rabbitmq:1.4.0.RELEASE"
     api 'com.fasterxml.jackson.core:jackson-databind:2.9.7'
     testImplementation 'io.projectreactor:reactor-test'
 }