Add prefetch parameter for AMQP

Evolveum · Mar 26, 2019 · d05a28e · d05a28e
1 parent 7ac3895
commit d05a28e
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/infra/schema/src/main/resources/xml/ns/public/common/common-asynchronous-3.xsd b/infra/schema/src/main/resources/xml/ns/public/common/common-asynchronous-3.xsd
@@ -152,6 +152,13 @@
                             </xsd:documentation>
                         </xsd:annotation>
                     </xsd:element>
+                    <xsd:element name="prefetch" type="xsd:int" minOccurs="0">
+                        <xsd:annotation>
+                            <xsd:documentation>
+                                Number of messages to prefetch. The default is 5.
+                            </xsd:documentation>
+                        </xsd:annotation>
+                    </xsd:element>
                 </xsd:sequence>
             </xsd:extension>
         </xsd:complexContent>

diff --git a/...a/com/evolveum/midpoint/provisioning/ucf/impl/builtin/async/Amqp091AsyncUpdateSource.java b/...a/com/evolveum/midpoint/provisioning/ucf/impl/builtin/async/Amqp091AsyncUpdateSource.java
@@ -40,6 +40,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.apache.commons.lang3.ObjectUtils.defaultIfNull;
+
 /**
  *  Async Update source for AMQP 0.9.1 brokers.
  *
@@ -48,6 +50,7 @@
 public class Amqp091AsyncUpdateSource implements AsyncUpdateSource {
 
 	private static final Trace LOGGER = TraceManager.getTrace(Amqp091AsyncUpdateSource.class);
+	private static final int DEFAULT_PREFETCH = 10;
 
 	@NotNull private final Amqp091SourceType sourceConfiguration;
 	@NotNull private final PrismContext prismContext;
@@ -97,6 +100,7 @@ private ListeningActivityImpl(AsyncUpdateMessageListener listener) {
 				state = State.PREPARING;
 				activeConnection = connectionFactory.newConnection();
 				activeChannel = activeConnection.createChannel();
+				activeChannel.basicQos(defaultIfNull(sourceConfiguration.getPrefetch(), DEFAULT_PREFETCH));
 				LOGGER.info("Opened AMQP connection = {}, channel = {}", activeConnection, activeChannel);  // todo debug
 				DeliverCallback deliverCallback = (consumerTag, message) -> {
 					try {