jinterop/artemis/Receiver.java

   1 //
   2 // Copyright © 2012-2018 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
  53     Context ictx = new InitialContext();
  54     QueueConnectionFactory qcf = (QueueConnectionFactory) ictx.lookup("ConnectionFactory");
  55     Queue queue = (Queue) ictx.lookup("queue/exampleQueue");
  56     ictx.close();
  57
  58     QueueConnection qconn = qcf.createQueueConnection();
  59     QueueSession qsess = qconn.createQueueSession(true, 0);
  60
  61     QueueReceiver qrec = qsess.createReceiver(queue);
  62     Message msg;
  63
  64     qconn.start();
  65         System.out.println("Waits are for " + Constants.RECEIVE_WAIT + "(ms)");
  66     int i;
  67     // for (i = 0; i < 10; i++) {
  68     for (i = 0;; i++) {
  69       // msg = qrec.receive();
  70       msg = qrec.receive(Constants.RECEIVE_WAIT);
  71       if (msg == null) break;
  72       if (msg instanceof TextMessage)
  73       {
  74                 System.out.println("Text Msg received: " + ((TextMessage) msg).getText());
  75       }
  76       else if (msg instanceof ObjectMessage)
  77       {
  78                 System.out.println("Object Msg received: "
  79            + ((ObjectMessage) msg).getObject());
  80       }
  81       else if (msg instanceof BytesMessage)
  82       {
  83             BytesMessage bmsg = (BytesMessage)msg;
  84             long len = bmsg.getBodyLength();
  85             byte[] buff = new byte[(int)len];
  86             int rlen = bmsg.readBytes(buff);
  87             String body = new String(buff);
  88                 System.out.println("Bytes Msg received: "
  89                 + "len:" + rlen + "\n" + "body:" + body);
  90       }
  91       else if (msg instanceof StreamMessage)
  92       {
  93                 System.out.println("Stream Msg received: ");
  94             System.out.println(msg);
  95       }
  96       else if (msg instanceof MapMessage)
  97       {
  98                 System.out.print("Map Msg received: ");
  99             System.out.println(msg);
 100       }
 101
 102       else
 103       {
 104                 System.out.println("Some Other Msg received: " + msg);
 105       }
 106       int nextWait = Constants.MIN_WORK + gener.nextInt(Constants.WORK_SPAN);
 107         System.out.println("Next Input Process Wait Length: " + nextWait);
 108       try
 109             {
 110         Thread.sleep(nextWait); // process work
 111             }
 112         catch(InterruptedException ie)
 113             {
 114                 //
 115             }
 116     }
 117     qsess.commit();
 118         System.out.println(i + " messages received.");
 119     qconn.close();
 120
 121   }
 122 } // end of class Receiver