Merge pull request #29 from juanpmarin/master

perf: use elastic scheduler in reactive message sender

Author: dbuos

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

@@ -4,6 +4,8 @@
 import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.exceptions.SendFailureNoAckException;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 import reactor.rabbitmq.OutboundMessage;
 import reactor.rabbitmq.Sender;
 
@@ -17,6 +19,8 @@
 
 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;
@@ -30,10 +34,15 @@ public ReactiveMessageSender(Sender sender, String sourceApplication, MessageCon
     }
 
     public <T> Mono<Void> sendWithConfirm(T message, String exchange, String routingKey, Map<String, Object> headers) {
-        return sender.sendWithPublishConfirms(just(toOutboundMessage(message, exchange, routingKey, headers))).flatMap(result -> result.isAck() ?
-            Mono.empty() :
-            Mono.error(new SendFailureNoAckException("Event no ACK in communications"))
-        ).then();
+        return just(toOutboundMessage(message, exchange, routingKey, headers))
+                .publishOn(senderScheduler)
+                .map(Mono::just)
+                .flatMapMany(sender::sendWithPublishConfirms)
+                .flatMap(result -> result.isAck() ?
+                        Mono.empty() :
+                        Mono.error(new SendFailureNoAckException("Event no ACK in communications"))
+                )
+                .then();
     }
 
     private <T> OutboundMessage toOutboundMessage(T object, String exchange, String routingKey, Map<String, Object> headers) {
@@ -48,13 +57,13 @@ private AMQP.BasicProperties buildMessageProperties(Message message, Map<String,
         baseHeaders.putAll(headers);
         baseHeaders.put(SOURCE_APPLICATION, sourceApplication);
         return new AMQP.BasicProperties.Builder()
-            .contentType(properties.getContentType())
-            .appId(sourceApplication)
-            .contentEncoding(properties.getContentEncoding())
-            .deliveryMode(2)
-            .timestamp(new Date())
-            .messageId(UUID.randomUUID().toString())
-            .headers(baseHeaders).build();
+                .contentType(properties.getContentType())
+                .appId(sourceApplication)
+                .contentEncoding(properties.getContentEncoding())
+                .deliveryMode(2)
+                .timestamp(new Date())
+                .messageId(UUID.randomUUID().toString())
+                .headers(baseHeaders).build();
     }
 
     public Sender getSender() {