MessageBodyCreator => MessageBodySource

Author: michaelklishin

src/main/java/com/rabbitmq/perf/FromFilesMessageBodyCreator.java renamed to src/main/java/com/rabbitmq/perf/FromFilesMessageBodySource.java

@@ -22,13 +22,13 @@
 /**
  *
  */
-public class FromFilesMessageBodyCreator implements MessageBodyCreator {
+public class FromFilesMessageBodySource implements MessageBodySource {
 
     private final List<byte[]> bodies;
 
     private final String contentType;
 
-    public FromFilesMessageBodyCreator(List<String> filesNames, String contentType) throws IOException {
+    public FromFilesMessageBodySource(List<String> filesNames, String contentType) throws IOException {
         bodies = new ArrayList<byte[]>(filesNames.size());
         for (String fileName : filesNames) {
             File file = new File(fileName.trim());
@@ -51,7 +51,7 @@ public FromFilesMessageBodyCreator(List<String> filesNames, String contentType)
         this.contentType = contentType;
     }
 
-    public FromFilesMessageBodyCreator(List<String> filesNames) throws IOException {
+    public FromFilesMessageBodySource(List<String> filesNames) throws IOException {
         this(filesNames, null);
     }
 

src/main/java/com/rabbitmq/perf/MessageBodyCreator.java renamed to src/main/java/com/rabbitmq/perf/MessageBodySource.java

@@ -18,9 +18,10 @@
 import java.io.IOException;
 
 /**
- *
+ * Sources produce message bodies and content type
+ * used by publishers.
  */
-public interface MessageBodyCreator {
+public interface MessageBodySource {
 
     MessageBodyAndContentType create(int sequenceNumber) throws IOException;
 

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

@@ -223,17 +223,17 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
         if (!predeclared || !exchangeExists(connection, exchangeName)) {
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
-        MessageBodyCreator messageBodyCreator = null;
+        MessageBodySource messageBodySource = null;
         if (bodyFiles.size() > 0) {
-            messageBodyCreator = new FromFilesMessageBodyCreator(bodyFiles, bodyContentType);
+            messageBodySource = new FromFilesMessageBodySource(bodyFiles, bodyContentType);
         } else {
-            messageBodyCreator = new SequenceTimeMessageBodyCreator(minMsgSize);
+            messageBodySource = new SequenceTimeMessageBodySource(minMsgSize);
         }
         final Producer producer = new Producer(channel, exchangeName, id,
                                                randomRoutingKey, flags, producerTxSize,
                                                producerRateLimit, producerMsgCount,
                                                timeLimit,
-                                               confirm, messageBodyCreator, stats);
+                                               confirm, messageBodySource, stats);
         channel.addReturnListener(producer);
         channel.addConfirmListener(producer);
         return producer;

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

@@ -18,11 +18,8 @@
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConfirmListener;
-import com.rabbitmq.client.MessageProperties;
 import com.rabbitmq.client.ReturnListener;
 
-import java.io.ByteArrayOutputStream;
-import java.io.DataOutputStream;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
@@ -47,7 +44,7 @@ public class Producer extends ProducerConsumerBase implements Runnable, ReturnLi
 
     private final Stats   stats;
 
-    private final MessageBodyCreator messageBodyCreator;
+    private final MessageBodySource messageBodySource;
 
     private Semaphore confirmPool;
     private final SortedSet<Long> unconfirmedSet =
@@ -56,7 +53,7 @@ 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, MessageBodyCreator messageBodyCreator, Stats stats)
+                    long confirm, MessageBodySource messageBodySource, Stats stats)
         throws IOException {
 
         this.channel            = channel;
@@ -70,7 +67,7 @@ public Producer(Channel channel, String exchangeName, String id, boolean randomR
         this.rateLimit          = rateLimit;
         this.msgLimit           = msgLimit;
         this.timeLimit          = 1000L * timeLimit;
-        this.messageBodyCreator = messageBodyCreator;
+        this.messageBodySource = messageBodySource;
         if (confirm > 0) {
             this.confirmPool  = new Semaphore((int)confirm);
         }
@@ -136,7 +133,7 @@ public void run() {
                 if (confirmPool != null) {
                     confirmPool.acquire();
                 }
-                publish(messageBodyCreator.create(totalMsgCount));
+                publish(messageBodySource.create(totalMsgCount));
                 totalMsgCount++;
                 msgCount++;
 
@@ -154,7 +151,7 @@ public void run() {
         }
     }
 
-    private void publish(MessageBodyCreator.MessageBodyAndContentType messageBodyAndContentType)
+    private void publish(MessageBodySource.MessageBodyAndContentType messageBodyAndContentType)
         throws IOException {
 
         AMQP.BasicProperties.Builder propertiesBuilder = new AMQP.BasicProperties.Builder();

src/main/java/com/rabbitmq/perf/SequenceTimeMessageBodyCreator.java renamed to src/main/java/com/rabbitmq/perf/SequenceTimeMessageBodySource.java

@@ -22,11 +22,11 @@
 /**
  *
  */
-public class SequenceTimeMessageBodyCreator implements MessageBodyCreator {
+public class SequenceTimeMessageBodySource implements MessageBodySource {
 
     private final byte[] message;
 
-    public SequenceTimeMessageBodyCreator(int minMsgSize) {
+    public SequenceTimeMessageBodySource(int minMsgSize) {
         this.message = new byte[minMsgSize];
     }
 

src/test/java/com/rabbitmq/perf/FromFilesMessageBodyCreatorTest.java renamed to src/test/java/com/rabbitmq/perf/FromFilesMessageBodySourceTest.java

@@ -15,15 +15,13 @@
 
 package com.rabbitmq.perf;
 
-import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
 import java.io.File;
 import java.io.FileWriter;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 
 import static java.util.Arrays.asList;
@@ -33,7 +31,7 @@
 /**
  *
  */
-public class FromFilesMessageBodyCreatorTest {
+public class FromFilesMessageBodySourceTest {
 
     @Rule
     public TemporaryFolder folder = new TemporaryFolder();
@@ -42,7 +40,7 @@ public class FromFilesMessageBodyCreatorTest {
         File file = folder.newFile("content.txt");
         String content = "dummy content";
         write(file, content);
-        MessageBodyCreator creator = new FromFilesMessageBodyCreator(asList(file.getAbsolutePath()));
+        MessageBodySource creator = new FromFilesMessageBodySource(asList(file.getAbsolutePath()));
         byte[] body1 = creator.create(1).getBody();
         byte[] body2 = creator.create(1).getBody();
         assertEquals(content, new String(body1, "UTF-8"));
@@ -58,7 +56,7 @@ public class FromFilesMessageBodyCreatorTest {
             files.add(file.getAbsolutePath());
         }
 
-        MessageBodyCreator creator = new FromFilesMessageBodyCreator(files);
+        MessageBodySource creator = new FromFilesMessageBodySource(files);
         byte[] body0 = creator.create(0).getBody();
         assertEquals("content0", new String(body0, "UTF-8"));
         byte[] body1 = creator.create(1).getBody();
@@ -72,7 +70,7 @@ public class FromFilesMessageBodyCreatorTest {
     @Test public void createFileDoesNotExist() throws Exception {
         File file = new File(folder.getRoot(), "dummy.txt");
         try {
-            new FromFilesMessageBodyCreator(asList(file.getAbsolutePath()));
+            new FromFilesMessageBodySource(asList(file.getAbsolutePath()));
             fail("File does not exist, exception should have thrown");
         } catch (IllegalArgumentException e) {
             // ok