feat: add prefetch count property

Author: juanmarin-co

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

@@ -13,7 +13,6 @@
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.config.props.AsyncProps;
 import org.reactivecommons.async.impl.config.props.BrokerConfigProps;
-import org.reactivecommons.async.impl.config.props.FluxProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.converters.json.DefaultObjectMapperSupplier;
 import org.reactivecommons.async.impl.converters.json.JacksonMessageConverter;
@@ -64,8 +63,6 @@ public ReactiveMessageSender messageSender(ConnectionFactoryProvider provider, M
         map.from(rabbitProperties.getCache().getChannel()::getSize).whenNonNull()
                 .to(channelPoolOptions::maxCacheSize);
 
-
-
         final ChannelPool channelPool = ChannelPoolFactory.createChannelPool(
                 senderConnection,
                 channelPoolOptions
@@ -87,7 +84,8 @@ public ReactiveMessageListener messageListener(ConnectionFactoryProvider provide
 
         return new ReactiveMessageListener(receiver,
                 new TopologyCreator(sender),
-                asyncProps.getFlux().getMaxConcurrency());
+                asyncProps.getFlux().getMaxConcurrency(),
+                asyncProps.getPrefetchCount());
     }
 
     @Bean
@@ -99,6 +97,7 @@ public ConnectionFactoryProvider connectionFactory(RabbitProperties properties)
         map.from(properties::determinePort).to(factory::setPort);
         map.from(properties::determineUsername).whenNonNull().to(factory::setUsername);
         map.from(properties::determinePassword).whenNonNull().to(factory::setPassword);
+        map.from(properties::determineVirtualHost).whenNonNull().to(factory::setVirtualHost);
         factory.useNio();
         return () -> factory;
     }

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

@@ -22,6 +22,8 @@ public class AsyncProps {
 
     private Integer maxRetries = 10;
 
+    private Integer prefetchCount = 250;
+
     private Integer retryDelay = 1000;
 
     private Boolean withDLQRetry = false;

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

@@ -1,33 +1,22 @@
 package org.reactivecommons.async.impl.communications;
 
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
 import reactor.rabbitmq.Receiver;
 
+
+@Getter
+@RequiredArgsConstructor
 public class ReactiveMessageListener {
 
     private final Receiver receiver;
     private final TopologyCreator topologyCreator;
     private final Integer maxConcurrency;
+    private final Integer prefetchCount;
 
     public ReactiveMessageListener(Receiver receiver, TopologyCreator topologyCreator) {
-        this(receiver, topologyCreator, 250);
-    }
-
-    public ReactiveMessageListener(Receiver receiver, TopologyCreator topologyCreator, Integer maxConcurrency) {
-        this.receiver = receiver;
-        this.topologyCreator = topologyCreator;
-        this.maxConcurrency = maxConcurrency;
-    }
-
-    public TopologyCreator getTopologyCreator() {
-        return topologyCreator;
-    }
-
-    public Receiver getReceiver() {
-        return receiver;
+        this(receiver, topologyCreator, 250, 250);
     }
 
-    public Integer getMaxConcurrency() {
-        return maxConcurrency;
-    }
 }
 

async/async-commons/src/main/java/org/reactivecommons/async/impl/listeners/GenericMessageListener.java

@@ -13,6 +13,7 @@
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.rabbitmq.AcknowledgableDelivery;
+import reactor.rabbitmq.ConsumeOptions;
 import reactor.rabbitmq.Receiver;
 
 import java.time.Duration;
@@ -64,8 +65,12 @@ public void startListener() {
         } else {
             log.log(Level.INFO, "ATTENTION! Using infinite fast retries as Retry Strategy");
         }
+
+        ConsumeOptions consumeOptions = new ConsumeOptions();
+        consumeOptions.qos(messageListener.getPrefetchCount());
+
         setUpBindings(messageListener.getTopologyCreator()).thenMany(
-                receiver.consumeManualAck(queueName)
+                receiver.consumeManualAck(queueName, consumeOptions)
                         .transform(this::consumeFaultTolerant)
                         .transform(this::outerFailureProtection))
                 .subscribe();

domain/domain-events/domain-events-api.gradle

@@ -72,5 +72,5 @@ bintray {
 }
 
 dependencies {
-    api group: 'org.reactivestreams', name: 'reactive-streams'
-}
+    api group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.3'
+}

gradle.properties

@@ -1,3 +1,3 @@
-version=0.4.3
+version=0.4.5
 springBootVersion=2.2.6.RELEASE
 gradleVersionsVersion=0.28.0

samples/async/receiver-responder/src/main/resources/application.properties

@@ -1 +1 @@
-spring.application.name=Receiver2
+spring.application.name=Receiver2