Merge branch '1.x.x-stable'

Author: acogoluegnes

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

@@ -21,8 +21,6 @@
 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;
@@ -114,9 +112,14 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
             msgCount++;
 
             if (msgLimit == 0 || msgCount <= msgLimit) {
-                DataInputStream d = new DataInputStream(new ByteArrayInputStream(body));
-                d.readInt();
-                long msgNano = d.readLong();
+                long msgNano;
+                Object timestamp = properties.getHeaders().get("timestamp");
+                if (timestamp == null) {
+                    msgNano = Long.MAX_VALUE;
+                } else {
+                    msgNano = (Long) timestamp;
+                }
+
                 long nano = System.nanoTime();
 
                 if (!autoAck) {

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

@@ -161,6 +161,8 @@ private void publish(MessageBodySource.MessageBodyAndContentType messageBodyAndC
             propertiesBuilder.contentType(messageBodyAndContentType.getContentType());
         }
 
+        propertiesBuilder.headers(Collections.<String, Object>singletonMap("timestamp", System.nanoTime()));
+
         unconfirmedSet.add(channel.getNextPublishSeqNo());
         channel.basicPublish(exchangeName, randomRoutingKey ? UUID.randomUUID().toString() : id,
                              mandatory, false,