fix: bump reactor-rabbitmq version and use sender for topology creator

Author: Juan Marín

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

@@ -55,14 +55,16 @@ public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, M
 
         final Sender sender = RabbitFlux.createSender(new SenderOptions().channelPool(channelPool));
 
-        return new ReactiveMessageSender(sender, brokerConfigProps.getAppName(), converter, new TopologyCreator(senderConnection));
+        return new ReactiveMessageSender(sender, brokerConfigProps.getAppName(), converter, new TopologyCreator(sender));
     }
 
     @Bean
     public ReactiveMessageListener messageListener(ConnectionFactoryProvider provider) {
         final Mono<Connection> connection = createSenderConnectionMono(provider.getConnectionFactory(), "listener");
         Receiver receiver = RabbitFlux.createReceiver(new ReceiverOptions().connectionMono(connection));
-        return new ReactiveMessageListener(receiver, new TopologyCreator(connection), maxConcurrency);
+        final Sender sender = RabbitFlux.createSender(new SenderOptions().connectionMono(connection));
+
+        return new ReactiveMessageListener(receiver, new TopologyCreator(sender), maxConcurrency);
     }
 
     @Bean
@@ -75,9 +77,6 @@ public ConnectionFactoryProvider connectionFactory(RabbitProperties properties)
         map.from(properties::determineUsername).whenNonNull().to(factory::setUsername);
         map.from(properties::determinePassword).whenNonNull().to(factory::setPassword);
         map.from(properties::determineVirtualHost).whenNonNull().to(factory::setVirtualHost);
-        map.from(properties::getRequestedHeartbeat).whenNonNull().asInt(Duration::getSeconds).to(factory::setRequestedHeartbeat);
-        factory.setAutomaticRecoveryEnabled(true);
-        factory.setTopologyRecoveryEnabled(true);
         factory.useNio();
         return () -> factory;
     }

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

async/async-commons/src/main/java/org/reactivecommons/async/impl/communications/ReactiveMessageSender.java

@@ -19,8 +19,6 @@
 
 public class ReactiveMessageSender {
 
-    private static final Scheduler senderScheduler = Schedulers.newElastic("reactive-message-sender");
-
     private final Sender sender;
     private final String sourceApplication;
     private final MessageConverter messageConverter;
@@ -35,7 +33,6 @@ public ReactiveMessageSender(Sender sender, String sourceApplication, MessageCon
 
     public <T> Mono<Void> sendWithConfirm(T message, String exchange, String routingKey, Map<String, Object> headers) {
         return just(toOutboundMessage(message, exchange, routingKey, headers))
-                .publishOn(senderScheduler)
                 .map(Mono::just)
                 .flatMapMany(sender::sendWithPublishConfirms)
                 .flatMap(result -> result.isAck() ?

async/async-commons/src/main/java/org/reactivecommons/async/impl/communications/TopologyCreator.java

@@ -8,6 +8,7 @@
 import reactor.rabbitmq.BindingSpecification;
 import reactor.rabbitmq.ExchangeSpecification;
 import reactor.rabbitmq.QueueSpecification;
+import reactor.rabbitmq.Sender;
 
 import java.io.IOException;
 import java.time.Duration;
@@ -19,63 +20,35 @@
  */
 public class TopologyCreator {
 
-    private final Mono<Channel> channel;
+    private final Sender sender;
 
-    public TopologyCreator(Mono<Connection> connectionMono) {
-        this.channel = connectionMono.map(connection -> {
-            try {
-                return connection.createChannel();
-            } catch (IOException e) {
-                throw new TopologyDefException("Fail to create channel", e);
-            }
-        }).doOnError(e -> log.log(Level.SEVERE, e.getMessage(), e))
-        .retryBackoff(5, Duration.ofMillis(500))
-        .cache();
+    public TopologyCreator(Sender sender) {
+        this.sender = sender;
     }
 
-    public Mono<AMQP.Exchange.DeclareOk> declare(ExchangeSpecification exchange){
-        return channel.map(ch -> {
-            try {
-                return ch.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.isDurable(), exchange.isAutoDelete(), exchange.getArguments());
-            } catch (IOException e) {
-                throw new TopologyDefException("Fail to declare exchange: " + exchange.getName(), e);
-            }
-        });
+    public Mono<AMQP.Exchange.DeclareOk> declare(ExchangeSpecification exchange) {
+        return sender.declare(exchange)
+                .onErrorMap(TopologyDefException::new);
     }
 
-    public Mono<AMQP.Queue.DeclareOk> declare(QueueSpecification queue){
-        return channel.map(ch -> {
-            try {
-                return ch.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(), queue.getArguments());
-            } catch (IOException e) {
-                throw new TopologyDefException("Fail to declare queue: " + queue.getName(), e);
-            }
-        });
+    public Mono<AMQP.Queue.DeclareOk> declare(QueueSpecification queue) {
+        return sender.declare(queue)
+                .onErrorMap(TopologyDefException::new);
     }
 
-    public Mono<AMQP.Queue.BindOk> bind(BindingSpecification binding){
-        return channel.map(ch -> {
-            try {
-                return ch.queueBind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(), binding.getArguments());
-            } catch (IOException e) {
-                throw new TopologyDefException("Fail to bind queue: " + binding.getQueue(), e);
-            }
-        });
+    public Mono<AMQP.Queue.BindOk> bind(BindingSpecification binding) {
+        return sender.bind(binding)
+                .onErrorMap(TopologyDefException::new);
     }
 
     public Mono<AMQP.Queue.UnbindOk> unbind(BindingSpecification binding) {
-       return channel.map(ch -> {
-           try {
-               return ch.queueUnbind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(), binding.getArguments());
-           } catch (IOException e) {
-               throw new TopologyDefException("Fail to unbind queue: " + binding.getQueue(), e);
-           }
-       }) ;
+        return sender.unbind(binding)
+                .onErrorMap(TopologyDefException::new);
     }
 
     public static class TopologyDefException extends RuntimeException {
-        public TopologyDefException(String message, Throwable cause) {
-            super(message, cause);
+        public TopologyDefException(Throwable cause) {
+            super(cause);
         }
     }
 }

build.gradle

@@ -6,6 +6,7 @@ buildscript {
     }
     dependencies {
         classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
+        classpath("com.github.ben-manes:gradle-versions-plugin:$gradleVersionsVersion")
     }
 }
 
@@ -14,5 +15,4 @@ plugins {
 }
 
 apply from: './main.gradle'
-
-
+apply plugin: 'com.github.ben-manes.versions'

gradle.properties

@@ -1,2 +1,4 @@
 version=0.2.1-beta1
-springBootVersion=2.1.1.RELEASE
+springBootVersion=2.2.1.RELEASE
+gradleVersionsVersion=0.27.0
+