jinterop/activemq/Receiver.java

   1 //
   2 // Copyright © 2012-2016 Guy M. Allard
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //     http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15 //
  16
  17 import java.util.Random;
  18
  19 import javax.jms.JMSException;
  20 import javax.jms.Message;
  21 import javax.jms.ObjectMessage;
  22 import javax.jms.BytesMessage;
  23 import javax.jms.MapMessage;
  24 import javax.jms.StreamMessage;
  25 import javax.jms.Queue;
  26 import javax.jms.QueueConnection;
  27 import javax.jms.QueueConnectionFactory;
  28 import javax.jms.QueueReceiver;
  29 import javax.jms.QueueSession;
  30 import javax.jms.TextMessage;
  31 import javax.naming.Context;
  32 import javax.naming.InitialContext;
  33 import javax.naming.NamingException;
  34
  35
  36 /**
  37  * Requests messages on the queue.
  38  */
  39 public class Receiver
  40 {
  41
  42   public static void main(String[] args) throws Exception
  43   {
  44                 System.out.println("Requests to receive messages...");
  45     new Receiver().go();
  46   } // end of main
  47
  48
  49   private void go() throws NamingException, JMSException
  50   {
  51         Random gener = new Random(System.currentTimeMillis());
  52     Context ictx = new InitialContext();
  53     Queue queue = (Queue) ictx.lookup("Guys");
  54     QueueConnectionFactory qcf = (QueueConnectionFactory) ictx.lookup("ConnectionFactory");
  55     ictx.close();
  56
  57     QueueConnection qconn = qcf.createQueueConnection();
  58     QueueSession qsess = qconn.createQueueSession(true, 0);
  59     QueueReceiver qrec = qsess.createReceiver(queue);
  60     Message msg;
  61
  62     qconn.start();
  63         System.out.println("Waits are for " + Constants.RECEIVE_WAIT + "(ms)");
  64     int i;
  65     // for (i = 0; i < 10; i++) {
  66     for (i = 0;; i++) {
  67       // msg = qrec.receive();
  68       msg = qrec.receive(Constants.RECEIVE_WAIT);
  69       if (msg == null) break;
  70       if (msg instanceof TextMessage)
  71       {
  72                 System.out.println("Text Msg received: " + ((TextMessage) msg).getText());
  73       }
  74       else if (msg instanceof ObjectMessage)
  75       {
  76                 System.out.println("Object Msg received: "
  77            + ((ObjectMessage) msg).getObject());
  78       }
  79       else if (msg instanceof BytesMessage)
  80       {
  81             BytesMessage bmsg = (BytesMessage)msg;
  82             long len = bmsg.getBodyLength();
  83             byte[] buff = new byte[(int)len];
  84             int rlen = bmsg.readBytes(buff);
  85             String body = new String(buff);
  86                 System.out.println("Bytes Msg received: "
  87                 + "len:" + rlen + "\n" + "body:" + body);
  88       }
  89       else if (msg instanceof StreamMessage)
  90       {
  91                 System.out.println("Stream Msg received: ");
  92             System.out.println(msg);
  93       }
  94       else if (msg instanceof MapMessage)
  95       {
  96                 System.out.print("Map Msg received: ");
  97             System.out.println(msg);
  98       }
  99       else
 100       {
 101                 System.out.println("Some Other Msg received: " + msg);
 102       }
 103       int nextWait = Constants.MIN_WORK + gener.nextInt(Constants.WORK_SPAN);
 104         System.out.println("Next Input Process Wait Length: " + nextWait);
 105       try
 106             {
 107         Thread.sleep(nextWait); // process work
 108             }
 109         catch(InterruptedException ie)
 110             {
 111                 //
 112             }
 113     }
 114     qsess.commit();
 115         System.out.println(i + " messages received.");
 116     qconn.close();
 117
 118   }
 119 } // end of class Receiver