JMS tutorial

1. ให้ JMS ใช้คนละ thread กับ glassfish
Configuration -> Java Message Service, set JMS service to "LOCAL" rather than "EMBEDDED."

restart glassfish server.

ลอง ping เด๊ะ

dsin@dsin-laptop:~/glassfish$ ~/glassfish/bin/asadmin jms-ping
JMS Ping Status = RUNNING
Command jms-ping executed successfully.

2. create new CF
- JNDI name
- type javax.jms.ConnectionFactory คือ อะไรก็ได้
create new DS
- JNDI name
- Physical destination name
- type javax.jms.Queue : point-to-point



Test first

create new java project
- project name
Add external JARs

• glassfish/lib/javaee.jar
• glassfish/lib/appserv-rt.jar
• glassfish/lib/appserv-admin.jar
• glassfish/lib/install/applications/jmsra/imqjmsra.jar
  

สร้าง jUnit test

- JUnit 4 test
- package
- name

public class ReqReplyTest extends TestCase {

private static final String CONNECTION_FACTORY_NAME = "jms/ConnectionFactory";

private static final String REQUEST_QUEUE_NAME = "jms/RequestQueue"; // ที่ config ใน jndi
private static final String REPLY_QUEUE_NAME = "jms/ReplyQueue";

private Connection connection;
private Session session;
private Destination replyQueue;
private MessageProducer requestProducer;
private MessageConsumer replyConsumer;

public void setUp() throws NamingException, JMSException { // setUp
 // Get connection factory for the local JMS server
 Context jndiContext = new InitialContext();
 ConnectionFactory connectionFactory =
        (ConnectionFactory)jndiContext.lookup( CONNECTION_FACTORY_NAME ); //

 // Get connection to the JMS server
 System.out.println( "Creating connection" );
 connection = connectionFactory.createConnection();

 // Create a session on the connection that is not transacted
 System.out.println( "Creating session" );
 session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE ); // Session.AUTO_ACKNOWLEDGE, false = non-transaction

 // Set up a message producer or consumer for each queue
 Destination requestQueue = (Destination)jndiContext.lookup( REQUEST_QUEUE_NAME ); // jndiContext.lookup
 requestProducer = session.createProducer(requestQueue); // createProducer(requestQueue), new producer
 replyQueue = (Destination)jndiContext.lookup( REPLY_QUEUE_NAME );
 replyConsumer = session.createConsumer(replyQueue); // .createConsumer(replyQueue), new consumer

 // Enable receipt of messages on this connection
 connection.start(); //
}

// Test request-reply pattern
public void testReqReply() throws JMSException {

 // Send a message using the request queue
 TextMessage requestMessage = session.createTextMessage();
 requestMessage.setText("Hello world."); // setText
 requestMessage.setJMSReplyTo(replyQueue); // setJMSReplyTo, set return address
 requestProducer.send(requestMessage); // send

 // Output sent message
 System.out.println("Sent request");
 System.out.println("\tTime:       " + System.currentTimeMillis() + " ms");
 System.out.println("\tMessage ID: " + requestMessage.getJMSMessageID()); // JMSMessageID
 System.out.println("\tCorrel. ID: " + requestMessage.getJMSCorrelationID()); // JMSCorrelationID
 System.out.println("\tReply to:   " + requestMessage.getJMSReplyTo()); // JMSReplyTo
 System.out.println("\tContents:   " + requestMessage.getText()); // Text

 // "Block" until a reply is received on the reply queue
 System.out.println("Waiting for response...");
 Message msg = replyConsumer.receive();
 assertTrue( msg instanceof TextMessage ); // instanceof TextMessage

 // Output the reply message
 TextMessage replyMessage = (TextMessage)msg; // (TextMessage)
 System.out.println("Received reply ");
 System.out.println("\tTime:       " + System.currentTimeMillis() + " ms");
 System.out.println("\tMessage ID: " + replyMessage.getJMSMessageID());
 System.out.println("\tCorrel. ID: " + replyMessage.getJMSCorrelationID());
 System.out.println("\tReply to:   " + replyMessage.getJMSReplyTo());
 System.out.println("\tContents:   " + replyMessage.getText());
}

// Clean up code run after every test
public void tearDown() throws JMSException { // tearDown()
 connection.close(); //
}
}

สร้าง EJB project ชื่อ request-reply-ejb
สร้าง class

package ait.jmstutorial.messagebeans;

// MessageDriven bean
@MessageDriven( mappedName = "jms/RequestQueue", activationConfig = { // jms/RequestQueue, listen to Request
@ActivationConfigProperty( propertyName="destinationType", propertyValue="javax.jms.Queue" ), Queue, point-to-point
@ActivationConfigProperty( propertyName="acknowledgeMode", propertyValue="Auto-acknowledge")
})


public class ReplierBean implements javax.jms.MessageListener {
@Resource(mappedName="jms/ConnectionFactory")
private ConnectionFactory connectionFactory;

@Resource( mappedName = "jms/InvalidQueue" )
private Destination invalidDestination;

public void onMessage( Message message ) { // onMessage
 try {
   // Get JMS connection and session
   // send reply message     
   Connection connection = connectionFactory.createConnection();
   Session session = connection.createSession( true, 0 );

   // Check validity of message
   if ((message instanceof TextMessage) && (message.getJMSReplyTo() != null)) { // instanceof TextMessage, getJMSReplyTo() != null
     TextMessage requestMessage = (TextMessage) message;

     System.out.println("Received request");
     System.out.println("\tTime:       " + System.currentTimeMillis() + " ms");
     System.out.println("\tMessage ID: " + requestMessage.getJMSMessageID());
     System.out.println("\tCorrel. ID: " + requestMessage.getJMSCorrelationID());
     System.out.println("\tReply to:   " + requestMessage.getJMSReplyTo());
     System.out.println("\tContents:   " + requestMessage.getText());

     String contents = requestMessage.getText(); // getText()
     Destination replyDestination = message.getJMSReplyTo(); // getJMSReplyTo()

     MessageProducer replyProducer = session.createProducer( replyDestination );
     TextMessage replyMessage = session.createTextMessage();
     replyMessage.setText(contents); // send hello world back
     replyMessage.setJMSCorrelationID(requestMessage.getJMSMessageID()); // setJMSCorrelationID
     replyProducer.send(replyMessage);

     System.out.println("Sent reply");
     System.out.println("\tTime:       " + System.currentTimeMillis() + " ms");
     System.out.println("\tMessage ID: " + replyMessage.getJMSMessageID());
     System.out.println("\tCorrel. ID: " + replyMessage.getJMSCorrelationID());
     System.out.println("\tReply to:   " + replyMessage.getJMSReplyTo());
     System.out.println("\tContents:   " + replyMessage.getText());
   } else {  // if sth wrong
     System.out.println("Invalid message detected");
     System.out.println("\tType:       " + message.getClass().getName());
     System.out.println("\tTime:       " + System.currentTimeMillis() + " ms");
     System.out.println("\tMessage ID: " + message.getJMSMessageID());
     System.out.println("\tCorrel. ID: " + message.getJMSCorrelationID());
     System.out.println("\tReply to:   " + message.getJMSReplyTo());
     // send msg to invalidDestinationQueue
     message.setJMSCorrelationID(message.getJMSMessageID());
     MessageProducer invalidProducer = session.createProducer( invalidDestination );
     invalidProducer.send(message);

     System.out.println("Sent to invalid message queue");
     System.out.println("\tType:       " + message.getClass().getName());
     System.out.println("\tTime:       " + System.currentTimeMillis() + " ms");
     System.out.println("\tMessage ID: " + message.getJMSMessageID());
     System.out.println("\tCorrel. ID: " + message.getJMSCorrelationID());
     System.out.println("\tReply to:   " + message.getJMSReplyTo());
  }
  session.close();
  connection.close( ); // don't forget to close
 } catch(Exception e) {
     throw new EJBException(e);
 }
}
}

Export to glassfish
/domains/domain1/autodeploy

run as > junit test

>> console

Creating session
Sent request
Time:       1236931646622 ms
Message ID: ID:43-127.0.1.1(c8:26:59:71:1:62)-59395-1236931646585
Correl. ID: null
Reply to:   Sun Java System MQ Destination
getName():        ReplyQueue
Class:            com.sun.messaging.Queue
getVERSION():        3.0
isReadonly():        false
getProperties():    {imqDestinationName=ReplyQueue, imqDestinationDescription=A Description for the Destination Object}
Contents:   Hello world.
Waiting for response...
Received reply
Time:       1236931646935 ms
Message ID: ID:223-127.0.1.1(dc:8d:ed:1a:1:49)-35614-1236931646903
Correl. ID: ID:43-127.0.1.1(c8:26:59:71:1:62)-59395-1236931646585
Reply to:   null
Contents:   Hello world.

notice that Msg id อันที่สอง ตรงกับ corelation id in request

Message ID: ID:43-127.0.1.1(c8:26:59:71:1:62)-59395-1236931646585
Correl. ID: null

Correl. ID: ID:43-127.0.1.1(c8:26:59:71:1:62)-59395-1236931646585

>> glassfish server log

 INFO: Received request
INFO:     Time:       1236931646900 ms
INFO:     Message ID: ID:43-127.0.1.1(c8:26:59:71:1:62)-59395-1236931646585
INFO:     Correl. ID: null
INFO:     Reply to:   Sun Java System MQ Destination
getName():        ReplyQueue
Class:            com.sun.messaging.Queue
getVERSION():        3.0
isReadonly():        false
getProperties():    {imqDestinationName=ReplyQueue, imqDestinationDescription=A Description for the Destination Object}
INFO:     Contents:   Hello world.

INFO: Sent reply
INFO:     Time:       1236931646909 ms
INFO:     Message ID: ID:223-127.0.1.1(dc:8d:ed:1a:1:49)-35614-1236931646903
INFO:     Correl. ID: ID:43-127.0.1.1(c8:26:59:71:1:62)-59395-1236931646585
INFO:     Reply to:   null
INFO:     Contents:   Hello world.

ref : mdailey