Merge pull request #274 from RafalGoslawski/add-nack-requeue-false

Added --requeue (-re) option to set the requeue flag for nack operation

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/Consumer.java

@@ -36,10 +36,10 @@ public class Consumer extends AgentBase implements Runnable {
     private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
 
     private static final AckNackOperation ACK_OPERATION =
-            (ch, envelope, multiple) -> ch.basicAck(envelope.getDeliveryTag(), multiple);
+            (ch, envelope, multiple, requeue) -> ch.basicAck(envelope.getDeliveryTag(), multiple);
 
     private static final AckNackOperation NACK_OPERATION =
-            (ch, envelope, multiple) -> ch.basicNack(envelope.getDeliveryTag(), multiple, true);
+            (ch, envelope, multiple, requeue) -> ch.basicNack(envelope.getDeliveryTag(), multiple, requeue);
     static final String STOP_REASON_CONSUMER_REACHED_MESSAGE_LIMIT = "Consumer reached message limit";
 
     private volatile ConsumerImpl       q;
@@ -48,6 +48,7 @@ public class Consumer extends AgentBase implements Runnable {
     private final int                   txSize;
     private final boolean               autoAck;
     private final int                   multiAckEvery;
+    private final boolean               requeue;
     private final Stats                 stats;
     private final int                   msgLimit;
     private final Map<String, String>   consumerTagBranchMap = Collections.synchronizedMap(new HashMap<>());
@@ -81,6 +82,7 @@ public Consumer(ConsumerParameters parameters) {
         this.txSize            = parameters.getTxSize();
         this.autoAck           = parameters.isAutoAck();
         this.multiAckEvery     = parameters.getMultiAckEvery();
+        this.requeue           = parameters.isRequeue();
         this.stats             = parameters.getStats();
         this.msgLimit          = parameters.getMsgLimit();
         this.timestampProvider = parameters.getTimestampProvider();
@@ -266,9 +268,9 @@ private void ackIfNecessary(Envelope envelope, int currentMessageCount, final Ch
             if (!autoAck) {
                 dealWithWriteOperation(() -> {
                     if (multiAckEvery == 0) {
-                        ackNackOperation.apply(ch, envelope, false);
+                        ackNackOperation.apply(ch, envelope, false, requeue);
                     } else if (currentMessageCount % multiAckEvery == 0) {
-                        ackNackOperation.apply(ch, envelope, true);
+                        ackNackOperation.apply(ch, envelope, true, requeue);
                     }
                 }, recoveryProcess);
             }
@@ -473,7 +475,7 @@ public boolean simulateLatency() {
     @FunctionalInterface
     private interface AckNackOperation {
 
-        void apply(Channel channel, Envelope envelope, boolean multiple) throws IOException;
+        void apply(Channel channel, Envelope envelope, boolean multiple, boolean requeue) throws IOException;
 
     }
 

src/main/java/com/rabbitmq/perf/ConsumerParameters.java

@@ -43,6 +43,7 @@ public class ConsumerParameters {
     private ExecutorService executorService;
     private boolean polling;
     private boolean nack = false;
+    private boolean requeue = true;
 
     private int pollingInterval;
 
@@ -201,6 +202,15 @@ public ConsumerParameters setNack(boolean nack) {
         return this;
     }
 
+    public boolean isRequeue() {
+        return requeue;
+    }
+
+    public ConsumerParameters setRequeue(boolean requeue) {
+        this.requeue = requeue;
+        return this;
+    }
+ 
     public ConsumerParameters setConsumerArguments(Map<String, Object> consumerArguments) {
         this.consumerArguments = consumerArguments;
         return this;

src/main/java/com/rabbitmq/perf/MulticastParams.java

@@ -112,6 +112,7 @@ public class MulticastParams {
     private int pollingInterval = -1;
 
     private boolean nack = false;
+    private boolean requeue = true;
 
     private boolean jsonBody = false;
     private int bodyFieldCount = 1000;
@@ -437,6 +438,10 @@ public void setNack(boolean nack) {
         this.nack = nack;
     }
 
+    public void setRequeue(boolean requeue) {
+        this.requeue = requeue;
+    }
+
     public void setJsonBody(boolean jsonBody) {
         this.jsonBody = jsonBody;
     }
@@ -539,6 +544,7 @@ public Consumer createConsumer(Connection connection,
                 .setPolling(this.polling)
                 .setPollingInterval(this.pollingInterval)
                 .setNack(this.nack)
+                .setRequeue(this.requeue)
                 .setConsumerArguments(this.consumerArguments)
         );
         this.topologyHandler.next();

src/main/java/com/rabbitmq/perf/PerfTest.java

@@ -181,6 +181,7 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
             int pollingInterval = intArg(cmd, "pi", -1);
 
             boolean nack = hasOption(cmd, "na");
+            boolean requeue = boolArg(cmd, "re", true);
 
             boolean jsonBody = hasOption(cmd, "jb");
             int bodyFieldCount = intArg(cmd, "bfc", 1000);
@@ -372,6 +373,7 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
             p.setPolling(polling);
             p.setPollingInterval(pollingInterval);
             p.setNack(nack);
+            p.setRequeue(requeue);
             p.setJsonBody(jsonBody);
             p.setBodyFieldCount(bodyFieldCount);
             p.setBodyCount(bodyCount);
@@ -680,7 +682,8 @@ public static Options getOptions() {
         options.addOption(new Option("pi", "polling-interval",true, "time to wait before polling with basic.get, " +
                 "in millisecond, default is 0."));
 
-        options.addOption(new Option("na", "nack",false,"nack and requeue messages"));
+        options.addOption(new Option("na", "nack",false,"nack messages"));
+        options.addOption(new Option("re", "requeue",true,"should nacked messages be requeued"));
 
         options.addOption(new Option("jb", "json-body", false, "generate a random JSON document for message body. " +
                 "Use with --size."));