Calculate latency from body timestamp, if possible

Calculating the latency using an header showed a 10% throughput
drop, so PerfTest now uses a timestamp from the message body
(as before), and falls back to an header when it has no control
over the content of the body (i.e. when using the --body option).

Fixes #58

Author: acogoluegnes

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

@@ -15,12 +15,15 @@
 
 package com.rabbitmq.perf;
 
+import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.ShutdownSignalException;
 
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
@@ -44,11 +47,12 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
     private final CountDownLatch        latch = new CountDownLatch(1);
     private final Map<String, String>   consumerTagBranchMap = Collections.synchronizedMap(new HashMap<String, String>());
     private final ConsumerLatency       consumerLatency;
+    private final TimestampExtractor    timestampExtractor;
 
     public Consumer(Channel channel, String id,
                     List<String> queueNames, int txSize, boolean autoAck,
-                    int multiAckEvery, Stats stats, float rateLimit, int msgLimit, int timeLimit,
-                    int consumerLatencyInMicroSeconds) {
+                    int multiAckEvery, Stats stats, float rateLimit, int msgLimit, final int timeLimit,
+                    int consumerLatencyInMicroSeconds, boolean extractTimestampFromHeader) {
 
         this.channel       = channel;
         this.id            = id;
@@ -67,6 +71,24 @@ public Consumer(Channel channel, String id,
         } else {
             this.consumerLatency = new BusyWaitConsumerLatency(consumerLatencyInMicroSeconds * 1000);
         }
+        if (extractTimestampFromHeader) {
+            this.timestampExtractor = new TimestampExtractor() {
+                @Override
+                public long extract(BasicProperties properties, byte[] body) throws IOException {
+                    Object timestamp = properties.getHeaders().get(Producer.TIMESTAMP_HEADER);
+                    return timestamp == null ? Long.MAX_VALUE : (Long) timestamp;
+                }
+            };
+        } else {
+            this.timestampExtractor = new TimestampExtractor() {
+                @Override
+                public long extract(BasicProperties properties, byte[] body) throws IOException {
+                    DataInputStream d = new DataInputStream(new ByteArrayInputStream(body));
+                    d.readInt();
+                    return d.readLong();
+                }
+            };
+        }
     }
 
     public void run() {
@@ -108,13 +130,8 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
             msgCount++;
 
             if (msgLimit == 0 || msgCount <= msgLimit) {
-                long msgNano;
-                Object timestamp = properties.getHeaders().get("timestamp");
-                if (timestamp == null) {
-                    msgNano = Long.MAX_VALUE;
-                } else {
-                    msgNano = (Long) timestamp;
-                }
+                long msgNano = timestampExtractor.extract(properties, body);
+
 
                 long nano = System.nanoTime();
 
@@ -209,4 +226,10 @@ public void simulateLatency() {
             while(System.nanoTime() - start < delay);
         }
     }
+
+    private interface TimestampExtractor {
+
+        long extract(BasicProperties properties, byte[] body) throws IOException;
+
+    }
 }

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

@@ -242,16 +242,20 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
         MessageBodySource messageBodySource = null;
+        boolean timestampInHeader;
         if (bodyFiles.size() > 0) {
             messageBodySource = new LocalFilesMessageBodySource(bodyFiles, bodyContentType);
+            timestampInHeader = true;
         } else {
             messageBodySource = new TimeSequenceMessageBodySource(minMsgSize);
+            timestampInHeader = false;
         }
         final Producer producer = new Producer(channel, exchangeName, id,
                                                randomRoutingKey, flags, producerTxSize,
                                                producerRateLimit, producerMsgCount,
                                                timeLimit,
-                                               confirm, confirmTimeout, messageBodySource, stats);
+                                               confirm, confirmTimeout, messageBodySource,
+                                               timestampInHeader, stats);
         channel.addReturnListener(producer);
         channel.addConfirmListener(producer);
         return producer;
@@ -263,9 +267,18 @@ public Consumer createConsumer(Connection connection, Stats stats, String id) th
         List<String> generatedQueueNames = configureQueues(connection, id);
         if (consumerPrefetch > 0) channel.basicQos(consumerPrefetch);
         if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
+
+        boolean timestampInHeader;
+        if (bodyFiles.size() > 0) {
+            timestampInHeader = true;
+        } else {
+            timestampInHeader = false;
+        }
+
         return new Consumer(channel, id, generatedQueueNames,
                                          consumerTxSize, autoAck, multiAckEvery,
-                                         stats, consumerRateLimit, consumerMsgCount, timeLimit, consumerLatencyInMicroseconds);
+                                         stats, consumerRateLimit, consumerMsgCount, timeLimit,
+                                         consumerLatencyInMicroseconds, timestampInHeader);
     }
 
     public boolean shouldConfigureQueues() {

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

@@ -32,6 +32,8 @@
 public class Producer extends ProducerConsumerBase implements Runnable, ReturnListener,
         ConfirmListener
 {
+
+    public static final String TIMESTAMP_HEADER = "timestamp";
     private final Channel channel;
     private final String  exchangeName;
     private final String  id;
@@ -46,6 +48,7 @@ public class Producer extends ProducerConsumerBase implements Runnable, ReturnLi
 
     private final MessageBodySource messageBodySource;
 
+    private final PropertiesBuilderProcessor propertiesBuilderProcessor;
     private Semaphore confirmPool;
     private int confirmTimeout;
     private final SortedSet<Long> unconfirmedSet =
@@ -54,7 +57,8 @@ public class Producer extends ProducerConsumerBase implements Runnable, ReturnLi
     public Producer(Channel channel, String exchangeName, String id, boolean randomRoutingKey,
                     List<?> flags, int txSize,
                     float rateLimit, int msgLimit, int timeLimit,
-                    long confirm, int confirmTimeout, MessageBodySource messageBodySource, Stats stats)
+                    long confirm, int confirmTimeout, MessageBodySource messageBodySource,
+                    boolean timestampInHeader, Stats stats)
         throws IOException {
 
         this.channel            = channel;
@@ -68,6 +72,22 @@ public Producer(Channel channel, String exchangeName, String id, boolean randomR
         this.msgLimit           = msgLimit;
         this.timeLimit          = 1000L * timeLimit;
         this.messageBodySource = messageBodySource;
+        if (timestampInHeader) {
+            this.propertiesBuilderProcessor = new PropertiesBuilderProcessor() {
+                @Override
+                public AMQP.BasicProperties.Builder process(AMQP.BasicProperties.Builder builder) {
+                    builder.headers(Collections.<String, Object>singletonMap(TIMESTAMP_HEADER, System.nanoTime()));
+                    return builder;
+                }
+            };
+        } else {
+            this.propertiesBuilderProcessor = new PropertiesBuilderProcessor() {
+                @Override
+                public AMQP.BasicProperties.Builder process(AMQP.BasicProperties.Builder builder) {
+                    return builder;
+                }
+            };
+        }
         if (confirm > 0) {
             this.confirmPool  = new Semaphore((int)confirm);
             this.confirmTimeout = confirmTimeout;
@@ -173,7 +193,7 @@ private void publish(MessageBodySource.MessageBodyAndContentType messageBodyAndC
             propertiesBuilder.contentType(messageBodyAndContentType.getContentType());
         }
 
-        propertiesBuilder.headers(Collections.<String, Object>singletonMap("timestamp", System.nanoTime()));
+        propertiesBuilder = this.propertiesBuilderProcessor.process(propertiesBuilder);
 
         unconfirmedSet.add(channel.getNextPublishSeqNo());
         channel.basicPublish(exchangeName, randomRoutingKey ? UUID.randomUUID().toString() : id,
@@ -182,4 +202,10 @@ private void publish(MessageBodySource.MessageBodyAndContentType messageBodyAndC
                              messageBodyAndContentType.getBody());
     }
 
+    private interface PropertiesBuilderProcessor {
+
+        AMQP.BasicProperties.Builder process(AMQP.BasicProperties.Builder builder);
+
+    }
+
 }