Merge pull request #27 from juanpmarin/master

fix: avoid publish blocking with channels pooling

Author: dbuos

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

@@ -26,9 +26,17 @@ public RabbitMqConfig(String appName) {
         this.appName = appName;
     }
 
-    public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter) {
+    public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter,
+                                               RabbitProperties rabbitProperties) {
         final Mono<Connection> senderConnection = createSenderConnectionMono(provider.getConnectionFactory(), "sender");
-        final Sender sender = RabbitFlux.createSender(new SenderOptions().connectionMono(senderConnection));
+
+        ChannelPool channelPool = ChannelPoolFactory.createChannelPool(
+                senderConnection,
+                new ChannelPoolOptions().maxCacheSize(rabbitProperties.getChannelPoolMaxCacheSize())
+        );
+
+        final Sender sender = RabbitFlux.createSender(new SenderOptions().channelPool(channelPool));
+
         return new ReactiveMessageSender(sender, appName, converter, new TopologyCreator(senderConnection));
     }
 

async/async-commons-standalone/src/main/java/org/reactivecommons/async/impl/config/RabbitProperties.java

@@ -9,4 +9,5 @@ public class RabbitProperties {
     private String username = "guest";
     private String password = "guest";
     private String virtualHost;
+    private Integer channelPoolMaxCacheSize;
 }

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

@@ -36,10 +36,23 @@ public class RabbitMqConfig {
     private Integer maxConcurrency;
 
     @Bean
-    public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter, BrokerConfigProps props) {
-        final Mono<Connection> senderConnection = createSenderConnectionMono(provider.getConnectionFactory(), "sender");
-        final Sender sender = RabbitFlux.createSender(new SenderOptions().connectionMono(senderConnection));
-        return new ReactiveMessageSender(sender, props.getAppName(), converter, new TopologyCreator(senderConnection));
+    public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, MessageConverter converter,
+                                               BrokerConfigProps brokerConfigProps, RabbitProperties rabbitProperties) {
+        Mono<Connection> senderConnection = createSenderConnectionMono(provider.getConnectionFactory(), "sender");
+        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions();
+
+        PropertyMapper map = PropertyMapper.get();
+        map.from(rabbitProperties.getCache().getChannel()::getSize).whenNonNull()
+                .to(channelPoolOptions::maxCacheSize);
+
+        ChannelPool channelPool = ChannelPoolFactory.createChannelPool(
+                senderConnection,
+                channelPoolOptions
+        );
+
+        final Sender sender = RabbitFlux.createSender(new SenderOptions().channelPool(channelPool));
+
+        return new ReactiveMessageSender(sender, brokerConfigProps.getAppName(), converter, new TopologyCreator(senderConnection));
     }
 
     @Bean

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.0.0.RELEASE"
+    api "io.projectreactor.rabbitmq:reactor-rabbitmq:1.2.0.RELEASE"
     api 'com.fasterxml.jackson.core:jackson-databind:2.9.7'
     testImplementation 'io.projectreactor:reactor-test'
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,6 @@
+#Wed Jul 24 11:42:46 COT 2019
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-5.0-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-5.0-all.zip

samples/async/simpleConsumer/simple-consumer.gradle

@@ -2,5 +2,5 @@ dependencies {
 //    compile 'org.springframework.boot:spring-boot-starter-amqp'
     compile('org.springframework.boot:spring-boot-starter')
 //    compile project(":async:libs:reactor-rabbitmq")
-    compile "io.projectreactor.rabbitmq:reactor-rabbitmq:1.0.0.RC1"
+    compile "io.projectreactor.rabbitmq:reactor-rabbitmq:1.2.0.RELEASE"
 }