Polish "--use-millis" option

Some renaming, use lambdas where possible, use JDK-provided
functional interfaces where possible.

Author: acogoluegnes

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

@@ -15,7 +15,6 @@
 
 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;
@@ -31,6 +30,7 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
 
 public class Consumer extends ProducerConsumerBase implements Runnable {
 
@@ -47,7 +47,7 @@ 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;
+    private final BiFunction<BasicProperties, byte[], Long> timestampExtractor;
     private final TimestampProvider     timestampProvider;
 
     public Consumer(Channel channel, String id,
@@ -77,21 +77,20 @@ public Consumer(Channel channel, String id,
         }
 
         if (timestampProvider.isTimestampInHeader()) {
-            this.timestampExtractor = new TimestampExtractor() {
-                @Override
-                public long extract(BasicProperties properties, byte[] body) throws IOException {
+            this.timestampExtractor = (properties, body) -> {
                     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));
+            this.timestampExtractor = (properties, body) -> {
+                DataInputStream d = new DataInputStream(new ByteArrayInputStream(body));
+                try {
                     d.readInt();
                     return d.readLong();
+                } catch (IOException e) {
+                    throw new RuntimeException("Error while extracting timestamp from body");
                 }
+
             };
         }
     }
@@ -135,8 +134,8 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
             msgCount++;
 
             if (msgLimit == 0 || msgCount <= msgLimit) {
-                long msg_ts = timestampExtractor.extract(properties, body);
-                long now_ts = timestampProvider.getCurrentTime();
+                long messageTimestamp = timestampExtractor.apply(properties, body);
+                long nowTimestamp = timestampProvider.getCurrentTime();
 
                 if (!autoAck) {
                     if (multiAckEvery == 0) {
@@ -150,10 +149,10 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
                     channel.txCommit();
                 }
 
-                now = System.currentTimeMillis();
-
-                long diff_time = timestampProvider.getDifference(now_ts, msg_ts);
+                long diff_time = timestampProvider.getDifference(nowTimestamp, messageTimestamp);
                 stats.handleRecv(id.equals(envelope.getRoutingKey()) ? diff_time : 0L);
+
+                now = System.currentTimeMillis();
                 if (rateLimit > 0.0f) {
                     delay(now);
                 }
@@ -231,9 +230,4 @@ public void simulateLatency() {
         }
     }
 
-    private interface TimestampExtractor {
-
-        long extract(BasicProperties properties, byte[] body) throws IOException;
-
-    }
 }

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

@@ -264,7 +264,7 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
         if (!predeclared || !exchangeExists(connection, exchangeName)) {
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
-        MessageBodySource messageBodySource = null;
+        MessageBodySource messageBodySource;
         TimestampProvider tsp;
         if (bodyFiles.size() > 0) {
             tsp = new TimestampProvider(useMillis, true);

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

@@ -107,13 +107,7 @@ public static void main(String[] args) {
                     file.delete();
                 }
                 output = new PrintWriter(new BufferedWriter(new FileWriter(file)), true);
-                Runtime.getRuntime().addShutdownHook(new Thread() {
-
-                    @Override
-                    public void run() {
-                        output.close();
-                    }
-                });
+                Runtime.getRuntime().addShutdownHook(new Thread(() -> output.close()));
             } else {
                 output = null;
             }

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

@@ -28,6 +28,7 @@
 import java.util.UUID;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
+import java.util.function.UnaryOperator;
 
 public class Producer extends ProducerConsumerBase implements Runnable, ReturnListener,
         ConfirmListener
@@ -48,7 +49,7 @@ public class Producer extends ProducerConsumerBase implements Runnable, ReturnLi
 
     private final MessageBodySource messageBodySource;
 
-    private final PropertiesBuilderProcessor propertiesBuilderProcessor;
+    private final UnaryOperator<AMQP.BasicProperties.Builder> propertiesBuilderProcessor;
     private Semaphore confirmPool;
     private int confirmTimeout;
     private final SortedSet<Long> unconfirmedSet =
@@ -74,20 +75,12 @@ public Producer(Channel channel, String exchangeName, String id, boolean randomR
         this.timeLimitMillis   = 1000L * timeLimitSecs;
         this.messageBodySource = messageBodySource;
         if (tsp.isTimestampInHeader()) {
-            this.propertiesBuilderProcessor = new PropertiesBuilderProcessor() {
-                @Override
-                public AMQP.BasicProperties.Builder process(AMQP.BasicProperties.Builder builder) {
-                    builder.headers(Collections.<String, Object>singletonMap(TIMESTAMP_HEADER, tsp.getCurrentTime()));
-                    return builder;
-                }
+            this.propertiesBuilderProcessor = builder -> {
+                builder.headers(Collections.<String, Object>singletonMap(TIMESTAMP_HEADER, tsp.getCurrentTime()));
+                return builder;
             };
         } else {
-            this.propertiesBuilderProcessor = new PropertiesBuilderProcessor() {
-                @Override
-                public AMQP.BasicProperties.Builder process(AMQP.BasicProperties.Builder builder) {
-                    return builder;
-                }
-            };
+            this.propertiesBuilderProcessor = UnaryOperator.identity();
         }
         if (confirm > 0) {
             this.confirmPool  = new Semaphore((int)confirm);
@@ -141,7 +134,7 @@ private void handleAckNack(long seqNo, boolean multiple,
 
     public void run() {
         long now;
-        long startTime;
+        final long startTime;
         startTime = now = System.currentTimeMillis();
         lastStatsTime = startTime;
         msgCount = 0;
@@ -194,7 +187,7 @@ private void publish(MessageBodySource.MessageBodyAndContentType messageBodyAndC
             propertiesBuilder.contentType(messageBodyAndContentType.getContentType());
         }
 
-        propertiesBuilder = this.propertiesBuilderProcessor.process(propertiesBuilder);
+        propertiesBuilder = this.propertiesBuilderProcessor.apply(propertiesBuilder);
 
         unconfirmedSet.add(channel.getNextPublishSeqNo());
         channel.basicPublish(exchangeName, randomRoutingKey ? UUID.randomUUID().toString() : id,
@@ -203,10 +196,4 @@ private void publish(MessageBodySource.MessageBodyAndContentType messageBodyAndC
                              messageBodyAndContentType.getBody());
     }
 
-    private interface PropertiesBuilderProcessor {
-
-        AMQP.BasicProperties.Builder process(AMQP.BasicProperties.Builder builder);
-
-    }
-
 }