https://issues.apache.org/jira/browse/AMQ-5558 - some more options for producer/consumer tools

activemq-client/src/main/java/org/apache/activemq/util/ConsumerThread.java

@@ -38,6 +38,7 @@ public class ConsumerThread extends Thread {
     int transactions = 0;
     boolean running = false;
     CountDownLatch finished;
+    boolean bytesAsText;
 
     public ConsumerThread(Session session, Destination destination) {
         this.destination = destination;
@@ -56,6 +57,12 @@ public class ConsumerThread extends Thread {
                 Message msg = consumer.receive(receiveTimeOut);
                 if (msg != null) {
                     LOG.info(threadName + " Received " + (msg instanceof TextMessage ? ((TextMessage) msg).getText() : msg.getJMSMessageID()));
+                    if (bytesAsText && (msg instanceof BytesMessage)) {
+                        long length = ((BytesMessage) msg).getBodyLength();
+                        byte[] bytes = new byte[(int) length];
+                        ((BytesMessage) msg).readBytes(bytes);
+                        LOG.info("BytesMessage as text string: " + new String(bytes));
+                    }
                     received++;
                 } else {
                     if (breakOnNull) {
@@ -151,4 +158,12 @@ public class ConsumerThread extends Thread {
     public void setFinished(CountDownLatch finished) {
         this.finished = finished;
     }
+
+    public boolean isBytesAsText() {
+        return bytesAsText;
+    }
+
+    public void setBytesAsText(boolean bytesAsText) {
+        this.bytesAsText = bytesAsText;
+    }
 }

activemq-client/src/main/java/org/apache/activemq/util/ProducerThread.java

@@ -20,10 +20,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.jms.*;
-import java.io.File;
+import java.io.*;
-import java.io.FileReader;
-import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.net.URL;
 import java.util.concurrent.CountDownLatch;
 
@@ -44,6 +41,9 @@ public class ProducerThread extends Thread {
 
     int transactions = 0;
     int sentCount = 0;
+    String message;
+    String messageText = null;
+    String url = null;
     byte[] payload = null;
     boolean running = false;
     CountDownLatch finished;
@@ -114,35 +114,55 @@ public class ProducerThread extends Thread {
     }
 
     protected Message createMessage(int i) throws Exception {
-        Message message = null;
+        Message answer;
         if (payload != null) {
-            message = session.createBytesMessage();
+            answer = session.createBytesMessage();
-            ((BytesMessage)message).writeBytes(payload);
+            ((BytesMessage) answer).writeBytes(payload);
         } else {
             if (textMessageSize > 0) {
-                InputStreamReader reader = null;
+                if (messageText == null) {
+                    messageText = readInputStream(getClass().getResourceAsStream("demo.txt"), textMessageSize, i);
+                }
+            } else if (url != null) {
+                messageText = readInputStream(new URL(url).openStream(), -1, i);
+            } else if (message != null) {
+                messageText = message;
+            } else {
+                messageText = createDefaultMessage(i);
+            }
+            answer = session.createTextMessage(messageText);
+        }
+        if ((msgGroupID != null) && (!msgGroupID.isEmpty())) {
+            answer.setStringProperty("JMSXGroupID", msgGroupID);
+        }
+        return answer;
+    }
+
+    private String readInputStream(InputStream is, int size, int messageNumber) throws IOException {
+        InputStreamReader reader = new InputStreamReader(is);
         try {
-                    InputStream is = getClass().getResourceAsStream("demo.txt");
+            char[] buffer;
-                    reader = new InputStreamReader(is);
+            if (size > 0) {
-                    char[] chars = new char[textMessageSize];
+                buffer = new char[size];
-                    reader.read(chars);
+            } else {
-                    message = session.createTextMessage(String.valueOf(chars));
+                buffer = new char[1024];
-                } catch (Exception e) {
+            }
-                    LOG.warn(Thread.currentThread().getName() + " Failed to load " + textMessageSize + " bytes of demo text. Using default text message instead");
+            int count;
-                    message = session.createTextMessage("test message: " + i);
+            StringBuilder builder = new StringBuilder();
+            while ((count = reader.read(buffer)) != -1) {
+                builder.append(buffer, 0, count);
+                if (size > 0) break;
+            }
+            return builder.toString();
+        } catch (IOException ioe) {
+            return createDefaultMessage(messageNumber);
         } finally {
-                    if (reader != null) {
             reader.close();
         }
     }
-            } else {
+
-                message = session.createTextMessage("test message: " + i);
+    private String createDefaultMessage(int messageNumber) {
-            }
+        return "test message: " + messageNumber;
-        }
-        if ((msgGroupID != null) && (!msgGroupID.isEmpty())) {
-            message.setStringProperty("JMSXGroupID", msgGroupID);
-        }
-        return message;
     }
 
     public void setMessageCount(int messageCount) {
@@ -228,4 +248,20 @@ public class ProducerThread extends Thread {
     public void setFinished(CountDownLatch finished) {
         this.finished = finished;
     }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
 }

activemq-console/src/main/java/org/apache/activemq/console/command/ConsumerCommand.java

@@ -39,6 +39,7 @@ public class ConsumerCommand extends AbstractCommand {
     int sleep;
     int transactionBatchSize;
     int parallelThreads = 1;
+    boolean bytesAsText;
 
     @Override
     protected void runTask(List<String> tokens) throws Exception {
@@ -71,6 +72,7 @@ public class ConsumerCommand extends AbstractCommand {
                 consumer.setSleep(sleep);
                 consumer.setTransactionBatchSize(transactionBatchSize);
                 consumer.setFinished(active);
+                consumer.setBytesAsText(bytesAsText);
                 consumer.start();
             }
 
@@ -146,6 +148,14 @@ public class ConsumerCommand extends AbstractCommand {
         this.parallelThreads = parallelThreads;
     }
 
+    public boolean isBytesAsText() {
+        return bytesAsText;
+    }
+
+    public void setBytesAsText(boolean bytesAsText) {
+        this.bytesAsText = bytesAsText;
+    }
+
     @Override
     protected void printHelp() {
         printHelpFromFile();
@@ -160,5 +170,4 @@ public class ConsumerCommand extends AbstractCommand {
     public String getOneLineDescription() {
         return "Receives messages from the broker";
     }
-
 }

activemq-console/src/main/java/org/apache/activemq/console/command/ProducerCommand.java

@@ -18,14 +18,12 @@ package org.apache.activemq.console.command;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.command.ActiveMQDestination;
-import org.apache.activemq.util.IntrospectionSupport;
 import org.apache.activemq.util.ProducerThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.jms.Connection;
 import javax.jms.Session;
-import java.io.*;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 
@@ -39,6 +37,8 @@ public class ProducerCommand extends AbstractCommand {
     int messageCount = 1000;
     int sleep = 0;
     boolean persistent = true;
+    String message = null;
+    String url = null;
     int messageSize = 0;
     int textMessageSize;
     long msgTTL = 0L;
@@ -77,6 +77,8 @@ public class ProducerCommand extends AbstractCommand {
                 producer.setMsgTTL(msgTTL);
                 producer.setPersistent(persistent);
                 producer.setTransactionBatchSize(transactionBatchSize);
+                producer.setMessage(message);
+                producer.setUrl(url);
                 producer.setMessageSize(messageSize);
                 producer.setMsgGroupID(msgGroupID);
                 producer.setTextMessageSize(textMessageSize);
@@ -196,6 +198,22 @@ public class ProducerCommand extends AbstractCommand {
         this.parallelThreads = parallelThreads;
     }
 
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
     @Override
     protected void printHelp() {
         printHelpFromFile();

activemq-console/src/main/resources/org/apache/activemq/console/command/consumer.txt

@@ -9,3 +9,4 @@ Options :
     [--sleep                                         N] - millisecond sleep period between sends or receives; default 0
     [--transactionBatchSize                          N] - use send transaction batches of size N; default 0, no jms transactions
     [--parallelThreads                               N] - number of threads to run in parallel; default 1
+    [--bytesAsText                          true|false] - try to treat a BytesMessage as a text string

activemq-console/src/main/resources/org/apache/activemq/console/command/producer.txt

@@ -13,4 +13,6 @@ Options :
     [--msgTTL                                        N] - message TTL in milliseconds
     [--messageSize                                   N] - size in bytes of a BytesMessage; default 0, a simple TextMessage is used
     [--textMessageSize                               N] - size in bytes of a TextMessage, a Lorem ipsum demo TextMessage is used
+    [--message                                      ..] - a text string to use as the message body
+    [--url                                         URL] - a url pointing to a document to use as the message body
     [--msgGroupID                                   ..] - JMS message group identifier