EJB Message-Driven Beans

TopicPublisher

The TopicSession is used to create a TopicPublisher, which sends messages from the TravelAgent EJB to the destination specified by the Topic object. Any JMS clients that subscribe to that topic will receive a copy of the message:

TopicPublisher publisher = session.createPublisher(topic);

 

TextMessage textMsg = session.createTextMessage();

textMsg.setText(ticketDescription);

publisher.publish(textMsg);

Message types

In JMS, a message is a Java object with two parts: a header and a message body. The header is composed of delivery information and metadata, while the message body carries the application data, which can take several forms: text, serializable objects, byte streams, etc. The JMS API defines several message types (TextMessage, MapMessage, ObjectMessage, and others) and provides methods for delivering messages to and receiving messages from other applications.

For example, we can change the TravelAgent EJB so that it sends a MapMessage instead of a TextMessage:

TicketDO ticket = new TicketDO(customer,cruise,cabin,price);

...

TopicPublisher publisher = session.createPublisher(topic);

 

MapMessage mapMsg = session.createTextMessage();

mapMsg.setInt("CustomerID", ticket.customerID.intValue());

mapMsg.setInt("CruiseID", ticket.cruiseID.intValue());

mapMsg.setInt("CabinID", ticket.cabinID.intValue());

mapMsg.setDouble("Price", ticket.price);

 

publisher.publish(mapMsg);

The attributes of the MapMessage (CustomerID, CruiseID, CabinID, and Price) can be accessed by name from those JMS clients that receive it.

As an alternative, the TravelAgent EJB could be modified to use the ObjectMessage type, which would allow us to send the entire TicketDO object as the message using Java serialization:

TicketDO ticket = new TicketDO(customer,cruise,cabin,price);

...

TopicPublisher publisher = session.createPublisher(topic);

 

ObjectMessage objectMsg = session.createObjectMessage();

ObjectMsg.setObject(ticket);

 

publisher.publish(objectMsg);

In addition to the TextMessage, MapMessage, and ObjectMessage, JMS provides two other message types: StreamMessage and BytesMessage. StreamMessage can take the contents of an I/O stream as its payload. BytesMessage can take any array of bytes, which it treats as opaque data.

XML deployment descriptor

When a JMS resource is used, it must be declared in the bean's XML deployment descriptor, in a manner similar to the JDBC resource used by the Ship EJB in Chapter 10:

<enterprise-beans>
  <session>
    <ejb-name>TravelAgentBean</ejb-name>
    ...
    <resource-ref>
      <res-ref-name>jms/TopicFactory</res-ref-name>
      <res-type>javax.jms.TopicConnectionFactory</res-type>
      <res-auth>Container</res-auth>
    </resource-ref>
    <resource-ref>
      <res-ref-name>jdbc/titanDB</res-ref-name>
      <res-type>javax.sql.DataSource</res-type>
      <res-auth>Container</res-auth>
    </resource-ref>
    <resource-env-ref>
      <resource-env-ref-name>jms/TicketTopic</resource-env-ref-name>
      <resource-env-ref-type>javax.jms.Topic</resource-env-ref-type>
    </resource-env-ref>
    ...
  </session>

The <resource-ref> for the JMS TopicConnectionFactory is similar to the <resource-ref> declaration for the JDBC DataSource: it declares the JNDI ENC name, interface type, and authorization protocol. In addition to the <resource-ref>, the TravelAgent EJB must also declare the <resource-env-ref>, which lists any "administered objects" associated with a <resource-ref> entry. In this case, we declare the Topic used for sending a ticket message. At deployment time the deployer will map the JMS TopicConnectionFactory and Topic declared by the <resource-ref> and <resource-env-ref> elements to a JMS factory and topic.

JMS application client

To get a better idea of how JMS is used, we can create a Java application whose sole purpose is receiving and processing reservation messages. We will develop a very simple JMS client that simply prints a description of each ticket as it receives the messages. We'll assume that the TravelAgent EJB is using the TextMessage to send a description of the ticket to the JMS clients. The following code shows how the JMS application client might look:

import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.Topic;
import javax.jms.Session;
import javax.jms.TopicSubscriber;
import javax.jms.JMSException;
import javax.naming.InitialContext;
 
 
public class JmsClient_1 implements javax.jms.MessageListener {
 
  public static void main(String [] args) throws Exception {
    
    if(args.length != 2)
      throw new Exception("Wrong number of arguments");
    
    new JmsClient_1(args[0], args[1]);
    
    while(true){Thread.sleep(10000);}
    
  }
    
  public JmsClient_1(String factoryName, String topicName) throws Exception {
      
    InitialContext jndiContext = getInitialContext();
    
    TopicConnectionFactory factory = (TopicConnectionFactory)
      jndiContext.lookup("TopicFactoryNameGoesHere");
    
    Topic topic = (Topic)jndiContext.lookup("TopicNameGoesHere");
 
    TopicConnection connect = factory.createTopicConnection();
 
    TopicSession session =
      connect.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
 
    TopicSubscriber subscriber = session.createSubscriber(topic);
 
    subscriber.setMessageListener(this);
    
    connect.start();
  }
  
  public void onMessage(Message message) {
    try {
    
      TextMessage textMsg = (TextMessage)message;
      String text = textMsg.getText();
      System.out.println("\n RESERVATION RECIEVED:\n"+text);
    
    } catch(JMSException jmsE) {
      jmsE.printStackTrace();
    }
  }
  
  public static InitialContext getInitialContext() {
    // create vendor-specific JNDI context here
  }
}

The constructor of JmsClient_1 obtains the TopicConnectionFactory and Topic from the JNDI InitialContext. This context is created with vendor-specific properties so that the client can connect to the same JMS provider as the one used by the TravelAgent EJB. For example, the getInitialContext() method for the WebLogic application server would be coded as follows:

public static InitialContext getInitialContext() {

    Properties env = new Properties();

    env.put(Context.SECURITY_PRINCIPAL, "guest");

    env.put(Context.SECURITY_CREDENTIALS, "guest");

    env.put(Context.INITIAL_CONTEXT_FACTORY,

       "weblogic.jndi.WLInitialContextFactory");

    env.put(Context.PROVIDER_URL, "t3://localhost:7001");

    return new InitialContext(env);

}

Once the client has the TopicConnectionFactory and Topic, it creates a TopicConnection and a TopicSession in the same way as the TravelAgent EJB. The main difference is that the TopicSession object is used to create a TopicSubscriber instead of a TopicPublisher. The TopicSubscriber is designed specifically to process incoming messages that are published to its specified Topic:

TopicSession session = 

    connect.createTopicSession(false,Session.AUTO_ACKNOWLEDGE); 

 

TopicSubscriber subscriber = session.createSubscriber(topic);

 

subscriber.setMessageListener(this);

        

connect.start();

The TopicSubscriber can receive messages directly, or it can delegate the processing of the messages to a javax.jms.MessageListener. We chose to have JmsClient_1 implement the MessageListener interface so that it can process the messages itself. MessageListener objects implement a single method, onMessage(), which is invoked every time a new message is sent to the subscriber's topic. In this case, every time the TravelAgent EJB sends a reservation message to the topic, the JMS client will have its onMessage() method invoked so that it can receive a copy of the message and process it:

public void onMessage(Message message) {

    try {

        TextMessage textMsg = (TextMessage)message;

        String text = textMsg.getText();

        System.out.println("\n RESERVATION RECIEVED:\n"+text);

        

    } catch(JMSException jmsE) {

        jmsE.printStackTrace();

    }

}