JBoss.org Community Documentation
Example 7.6, “A TextMessage processing MDB” shows an message driven bean (MDB) that transforms the TextMessages
it receives and sends the transformed messages to the queue found in the incoming message JMSReplyTo
header.
package org.jboss.book.jms.ex2; import javax.ejb.MessageDrivenBean; import javax.ejb.MessageDrivenContext; import javax.ejb.EJBException; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueSender; import javax.jms.QueueSession; import javax.jms.TextMessage; import javax.naming.InitialContext; import javax.naming.NamingException; /** * An MDB that transforms the TextMessages it receives and send the * transformed messages to the Queue found in the incoming message * JMSReplyTo header. * * @author Scott.Stark@jboss.org * @version $Revision: 1.9 $ */ public class TextMDB implements MessageDrivenBean, MessageListener { private MessageDrivenContext ctx = null; private QueueConnection conn; private QueueSession session; public TextMDB() { System.out.println("TextMDB.ctor, this="+hashCode()); } public void setMessageDrivenContext(MessageDrivenContext ctx) { this.ctx = ctx; System.out.println("TextMDB.setMessageDrivenContext, this=" + hashCode()); } public void ejbCreate() { System.out.println("TextMDB.ejbCreate, this="+hashCode()); try { setupPTP(); } catch (Exception e) { throw new EJBException("Failed to init TextMDB", e); } } public void ejbRemove() { System.out.println("TextMDB.ejbRemove, this="+hashCode()); ctx = null; try { if (session != null) { session.close(); } if (conn != null) { conn.close(); } } catch(JMSException e) { e.printStackTrace(); } } public void onMessage(Message msg) { System.out.println("TextMDB.onMessage, this="+hashCode()); try { TextMessage tm = (TextMessage) msg; String text = tm.getText() + "processed by: "+hashCode(); Queue dest = (Queue) msg.getJMSReplyTo(); sendReply(text, dest); } catch(Throwable t) { t.printStackTrace(); } } private void setupPTP() throws JMSException, NamingException { InitialContext iniCtx = new InitialContext(); Object tmp = iniCtx.lookup("java:comp/env/jms/QCF"); QueueConnectionFactory qcf = (QueueConnectionFactory) tmp; conn = qcf.createQueueConnection(); session = conn.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE); conn.start(); } private void sendReply(String text, Queue dest) throws JMSException { System.out.println("TextMDB.sendReply, this=" + hashCode() + ", dest="+dest); QueueSender sender = session.createSender(dest); TextMessage tm = session.createTextMessage(text); sender.send(tm); sender.close(); } }
Example 7.6. A TextMessage processing MDB
The MDB ejb-jar.xml
and jboss.xml
deployment descriptors are shown in Example 7.7, “The MDB ejb-jar.xml descriptor” and Example 7.8, “The MDB jboss.xml descriptor”.
<?xml version="1.0"?> <!DOCTYPE ejb-jar PUBLIC "-//Sun Microsystems, Inc.//DTD Enterprise JavaBeans 2.0//EN" "http://java.sun.com/dtd/ejb-jar_2_0.dtd"> <ejb-jar> <enterprise-beans> <message-driven> <ejb-name>TextMDB</ejb-name> <ejb-class>org.jboss.book.jms.ex2.TextMDB</ejb-class> <transaction-type>Container</transaction-type> <acknowledge-mode>AUTO_ACKNOWLEDGE</acknowledge-mode> <message-driven-destination> <destination-type>javax.jms.Queue</destination-type> </message-driven-destination> <res-ref-name>jms/QCF</res-ref-name> <resource-ref> <res-type>javax.jms.QueueConnectionFactory</res-type> <res-auth>Container</res-auth> </resource-ref> </message-driven> </enterprise-beans> </ejb-jar>
Example 7.7. The MDB ejb-jar.xml descriptor
<?xml version="1.0"?> <jboss> <enterprise-beans> <message-driven> <ejb-name>TextMDB</ejb-name> <destination-jndi-name>queue/B</destination-jndi-name> <resource-ref> <res-ref-name>jms/QCF</res-ref-name> <jndi-name>ConnectionFactory</jndi-name> </resource-ref> </message-driven> </enterprise-beans> </jboss>
Example 7.8. The MDB jboss.xml descriptor
Example 7.9, “A JMS client that interacts with the TextMDB” shows a variation of the P2P client that sends several messages to the queue/B
destination and asynchronously receives the messages as modified by TextMDB
from queue A
.
package org.jboss.book.jms.ex2; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueReceiver; import javax.jms.QueueSender; import javax.jms.QueueSession; import javax.jms.TextMessage; import javax.naming.InitialContext; import javax.naming.NamingException; import EDU.oswego.cs.dl.util.concurrent.CountDown; /** * A complete JMS client example program that sends N TextMessages to * a Queue B and asynchronously receives the messages as modified by * TextMDB from Queue A. * * @author Scott.Stark@jboss.org * @version $Revision: 1.9 $ */ public class SendRecvClient { static final int N = 10; static CountDown done = new CountDown(N); QueueConnection conn; QueueSession session; Queue queA; Queue queB; public static class ExListener implements MessageListener { public void onMessage(Message msg) { done.release(); TextMessage tm = (TextMessage) msg; try { System.out.println("onMessage, recv text="+tm.getText()); } catch(Throwable t) { t.printStackTrace(); } } } public void setupPTP() throws JMSException, NamingException { InitialContext iniCtx = new InitialContext(); Object tmp = iniCtx.lookup("ConnectionFactory"); QueueConnectionFactory qcf = (QueueConnectionFactory) tmp; conn = qcf.createQueueConnection(); queA = (Queue) iniCtx.lookup("queue/A"); queB = (Queue) iniCtx.lookup("queue/B"); session = conn.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE); conn.start(); } public void sendRecvAsync(String textBase) throws JMSException, NamingException, InterruptedException { System.out.println("Begin sendRecvAsync"); // Setup the PTP connection, session setupPTP(); // Set the async listener for queA QueueReceiver recv = session.createReceiver(queA); recv.setMessageListener(new ExListener()); // Send a few text msgs to queB QueueSender send = session.createSender(queB); for(int m = 0; m < 10; m ++) { TextMessage tm = session.createTextMessage(textBase+"#"+m); tm.setJMSReplyTo(queA); send.send(tm); System.out.println("sendRecvAsync, sent text=" + tm.getText()); } System.out.println("End sendRecvAsync"); } public void stop() throws JMSException { conn.stop(); session.close(); conn.close(); } public static void main(String args[]) throws Exception { System.out.println("Begin SendRecvClient,now=" + System.currentTimeMillis()); SendRecvClient client = new SendRecvClient(); client.sendRecvAsync("A text msg"); client.done.acquire(); client.stop(); System.exit(0); System.out.println("End SendRecvClient"); } }
Example 7.9. A JMS client that interacts with the TextMDB
Run the client as follows:
[examples]$ ant -Dchap=jms -Dex=2 run-example ... run-example2: ... [java] Begin SendRecvClient, now=1102900541558 [java] Begin sendRecvAsync [java] sendRecvAsync, sent text=A text msg#0 [java] sendRecvAsync, sent text=A text msg#1 [java] sendRecvAsync, sent text=A text msg#2 [java] sendRecvAsync, sent text=A text msg#3 [java] sendRecvAsync, sent text=A text msg#4 [java] sendRecvAsync, sent text=A text msg#5 [java] sendRecvAsync, sent text=A text msg#6 [java] sendRecvAsync, sent text=A text msg#7 [java] sendRecvAsync, sent text=A text msg#8 [java] sendRecvAsync, sent text=A text msg#9 [java] End sendRecvAsync [java] onMessage, recv text=A text msg#0processed by: 12855623 [java] onMessage, recv text=A text msg#5processed by: 9399816 [java] onMessage, recv text=A text msg#9processed by: 6598158 [java] onMessage, recv text=A text msg#3processed by: 8153998 [java] onMessage, recv text=A text msg#4processed by: 10118602 [java] onMessage, recv text=A text msg#2processed by: 1792333 [java] onMessage, recv text=A text msg#7processed by: 14251014 [java] onMessage, recv text=A text msg#1processed by: 10775981 [java] onMessage, recv text=A text msg#8processed by: 6056676 [java] onMessage, recv text=A text msg#6processed by: 15679078
The corresponding JBoss server console output is:
19:15:40,232 INFO [EjbModule] Deploying TextMDB 19:15:41,498 INFO [EJBDeployer] Deployed: file:/jboss-4.2.0.GA/server/production/deploy/ jms-ex2.jar 19:15:45,606 INFO [TextMDB] TextMDB.ctor, this=10775981 19:15:45,620 INFO [TextMDB] TextMDB.ctor, this=1792333 19:15:45,627 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=10775981 19:15:45,638 INFO [TextMDB] TextMDB.ejbCreate, this=10775981 19:15:45,640 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=1792333 19:15:45,640 INFO [TextMDB] TextMDB.ejbCreate, this=1792333 19:15:45,649 INFO [TextMDB] TextMDB.ctor, this=12855623 19:15:45,658 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=12855623 19:15:45,661 INFO [TextMDB] TextMDB.ejbCreate, this=12855623 19:15:45,742 INFO [TextMDB] TextMDB.ctor, this=8153998 19:15:45,744 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=8153998 19:15:45,744 INFO [TextMDB] TextMDB.ejbCreate, this=8153998 19:15:45,763 INFO [TextMDB] TextMDB.ctor, this=10118602 19:15:45,764 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=10118602 19:15:45,764 INFO [TextMDB] TextMDB.ejbCreate, this=10118602 19:15:45,777 INFO [TextMDB] TextMDB.ctor, this=9399816 19:15:45,779 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=9399816 19:15:45,779 INFO [TextMDB] TextMDB.ejbCreate, this=9399816 19:15:45,792 INFO [TextMDB] TextMDB.ctor, this=15679078 19:15:45,798 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=15679078 19:15:45,799 INFO [TextMDB] TextMDB.ejbCreate, this=15679078 19:15:45,815 INFO [TextMDB] TextMDB.ctor, this=14251014 19:15:45,816 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=14251014 19:15:45,817 INFO [TextMDB] TextMDB.ejbCreate, this=14251014 19:15:45,829 INFO [TextMDB] TextMDB.ctor, this=6056676 19:15:45,831 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=6056676 19:15:45,864 INFO [TextMDB] TextMDB.ctor, this=6598158 19:15:45,903 INFO [TextMDB] TextMDB.ejbCreate, this=6056676 19:15:45,906 INFO [TextMDB] TextMDB.setMessageDrivenContext, this=6598158 19:15:45,906 INFO [TextMDB] TextMDB.ejbCreate, this=6598158 19:15:46,236 INFO [TextMDB] TextMDB.onMessage, this=12855623 19:15:46,238 INFO [TextMDB] TextMDB.sendReply, this=12855623, dest=QUEUE.A 19:15:46,734 INFO [TextMDB] TextMDB.onMessage, this=9399816 19:15:46,736 INFO [TextMDB] TextMDB.onMessage, this=8153998 19:15:46,737 INFO [TextMDB] TextMDB.onMessage, this=6598158 19:15:46,768 INFO [TextMDB] TextMDB.sendReply, this=9399816, dest=QUEUE.A 19:15:46,768 INFO [TextMDB] TextMDB.sendReply, this=6598158, dest=QUEUE.A 19:15:46,774 INFO [TextMDB] TextMDB.sendReply, this=8153998, dest=QUEUE.A 19:15:46,903 INFO [TextMDB] TextMDB.onMessage, this=10118602 19:15:46,904 INFO [TextMDB] TextMDB.sendReply, this=10118602, dest=QUEUE.A 19:15:46,927 INFO [TextMDB] TextMDB.onMessage, this=1792333 19:15:46,928 INFO [TextMDB] TextMDB.sendReply, this=1792333, dest=QUEUE.A 19:15:47,002 INFO [TextMDB] TextMDB.onMessage, this=14251014 19:15:47,007 INFO [TextMDB] TextMDB.sendReply, this=14251014, dest=QUEUE.A 19:15:47,051 INFO [TextMDB] TextMDB.onMessage, this=10775981 19:15:47,051 INFO [TextMDB] TextMDB.sendReply, this=10775981, dest=QUEUE.A 19:15:47,060 INFO [TextMDB] TextMDB.onMessage, this=6056676 19:15:47,061 INFO [TextMDB] TextMDB.sendReply, this=6056676, dest=QUEUE.A 19:15:47,064 INFO [TextMDB] TextMDB.onMessage, this=15679078 19:15:47,065 INFO [TextMDB] TextMDB.sendReply, this=15679078, dest=QUEUE.A
Items of note in this example include:
The JMS client has no explicit knowledge that it is dealing with an MDB. The client simply uses the standard JMS APIs to send messages to a queue and receive messages from another queue.
The MDB declares whether it will listen to a queue or topic in the ejb-jar.xml
descriptor. The name of the queue or topic must be specified using a jboss.xml
descriptor. In this example the MDB also sends messages to a JMS queue. MDBs may act as queue senders or topic publishers within their onMessage
callback.
The messages received by the client include a "processed by: NNN" suffix, where NNN is the hashCode
value of the MDB instance that processed the message. This shows that many MDBs may actively process messages posted to a destination. Concurrent processing is one of the benefits of MDBs.