Calculate latency based on header

The latency to receive a message was based on a value in the message payload.
This latency is then used to conditionally display the metrics on received messages.

This worked well as long as the payload contained the time when the message
was sent (the default) but it doesn't work anymore when specifying a given payload.

Feature #23 introduced content based on a file, so the latency wasn't correctly
calculated.

Latency is now calculated based on an header value, which is always sent,
whatever the MessageBodySource is used.

Fixes #50

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;
@@ -110,9 +108,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,