    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.codehaus.activemq.tool;
    19
    20  import javax.jms.Connection;
    21  import javax.jms.JMSException;
    22  import javax.jms.Message;
    23  import javax.jms.MessageConsumer;
    24  import javax.jms.MessageListener;
    25  import javax.jms.Session;
    26  import javax.jms.TextMessage;
    27  import javax.jms.Topic;
    28  import java.io.IOException;
    29
    30  /***
    31   * A simple tool for consuming messages
    32   *
    33   * @version $Revision: 1.11 $
    34   */
    35  public class ConsumerTool extends ToolSupport implements MessageListener {
    36
    37      protected int count = 0;
    38      protected int dumpCount = 10;
    39      protected boolean verbose = true;
    40      protected int maxiumMessages = 0;
    41      private boolean pauseBeforeShutdown;
    42
    43
    44      public static void main(String[] args) {
    45          ConsumerTool tool = new ConsumerTool();
    46          if (args.length > 0) {
    47              tool.url = args[0];
    48          }
    49          if (args.length > 1) {
    50              tool.topic = args[1].equalsIgnoreCase("true");
    51          }
    52          if (args.length > 2) {
    53              tool.subject = args[2];
    54          }
    55          if (args.length > 3) {
    56              tool.durable = args[3].equalsIgnoreCase("true");
    57          }
    58          if (args.length > 4) {
    59              tool.maxiumMessages = Integer.parseInt(args[4]);
    60          }
    61          tool.run();
    62      }
    63
    64      public void run() {
    65          try {
    66              System.out.println("Connecting to URL: " + url);
    67              System.out.println("Consuming " + (topic ? "topic" : "queue") + ": " + subject);
    68              System.out.println("Using " + (durable ? "durable" : "non-durable") + " subscription");
    69
    70              Connection connection = createConnection();
    71              Session session = createSession(connection);
    72              MessageConsumer consumer = null;
    73              if (durable && topic) {
    74                  consumer = session.createDurableSubscriber((Topic) destination, consumerName);
    75              }
    76              else {
    77                  consumer = session.createConsumer(destination);
    78              }
    79              if (maxiumMessages <= 0) {
    80                  consumer.setMessageListener(this);
    81              }
    82              connection.start();
    83
    84              if (maxiumMessages > 0) {
    85                  consumeMessagesAndClose(connection, session, consumer);
    86              }
    87          }
    88          catch (Exception e) {
    89              System.out.println("Caught: " + e);
    90              e.printStackTrace();
    91          }
    92      }
    93
    94      public void onMessage(Message message) {
    95          try {
    96              if (message instanceof TextMessage) {
    97                  TextMessage txtMsg = (TextMessage) message;
    98                  if (verbose) {
    99
    100                 	String msg = txtMsg.getText();
    101                 	if( msg.length() > 50 )
    102                 		msg = msg.substring(0, 50)+"...";
    103
    104                     System.out.println("Received: " + msg);
    105                 }
    106             }
    107             else {
    108                 if (verbose) {
    109                     System.out.println("Received: " + message);
    110                 }
    111             }
    112             /*
    113             if (++count % dumpCount == 0) {
    114                 dumpStats(connection);
    115             }
    116             */
    117         }
    118         catch (JMSException e) {
    119             System.out.println("Caught: " + e);
    120             e.printStackTrace();
    121         }
    122     }
    123
    124
    125     protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer) throws JMSException, IOException {
    126         System.out.println("We are about to wait until we consume: " + maxiumMessages + " message(s) then we will shutdown");
    127
    128         for (int i = 0; i < maxiumMessages; i++) {
    129             Message message = consumer.receive();
    130             onMessage(message);
    131         }
    132         System.out.println("Closing connection");
    133         consumer.close();
    134         session.close();
    135         connection.close();
    136         if (pauseBeforeShutdown) {
    137             System.out.println("Press return to shut down");
    138             System.in.read();
    139         }
    140     }
    141 }