Patterns. Broker Interactions for Intra- and Inter-Enterprise

9.7 Developing real-time message flows

Real-time flows allow JMS clients to publish and retrieve topic information using TCP/IP. The following steps outline the development process and guidelines for developing a simple real-time message flow.

9.7.1 Prepare the subscriber applications

In order to retrieve topic information, the application implements a message-driven bean.

Create the message-driven bean

Example 9-2 shows a simple message-driven bean that takes topic information and prints it to the console.

Example 9-2: Message-driven bean

package eb; import javax.jms.JMSException; import javax.jms.TextMessage; /** * Bean implementation class for Enterprise Bean: Wholesale */ public class WholesaleBean implements javax.ejb.MessageDrivenBean, javax.jms.MessageListener { private javax.ejb.MessageDrivenContext fMessageDrivenCtx; /** * getMessageDrivenContext */ public javax.ejb.MessageDrivenContext getMessageDrivenContext() { return fMessageDrivenCtx; } /** * setMessageDrivenContext */ public void setMessageDrivenContext(javax.ejb.MessageDrivenContext ctx) { fMessageDrivenCtx = ctx; } /** * ejbCreate */ public void ejbCreate() { } /** * onMessage */ public void onMessage(javax.jms.Message msg) { try { TextMessage textMessage = (TextMessage)msg; System.out.println("Message: "+textMessage.getText()); } catch (Exception e) { System.out.println(e.getMessage()); e.printStackTrace(); } } /** * ejbRemove */ public void ejbRemove() { } }

Figure 9-32 shows the EJB deployment descriptor characteristics for the message-driven bean.

Figure 9-32: WebSphere bindings for the message-driven bean

When you deploy the application, you will need to bind the bean to the message listener port. The application developer can set the default binding in WebSphere Studio in the EJB deployment descriptor using the Listener Port Name field in the Beans tab. The setting is stored in the META-INF/ibm-ejb-jar-bnd.xmi file. The administrator can override this at install time.

Deployment considerations

The use of a message-driven bean necessitates the definition of JMS resources to the server. You will need the following components:

• A topic connection factory

• A topic destination

• A listener port

For our example, these are defined using the WebSphere administrative console in the following way:

1. Select Resources -> WebSphere MQ JMS Provider.

2. The first thing you need is a topic connection factory. Select WebSphere MQ Topic Connection Factories in the Additional Properties table.

3. Click New and enter the following attributes and click OK:

   • Name: WholesaleTCF_EB_IP

   • JNDI Name: jms/eb/ip/WholesaleTCF

   • Host: ip_address_of_EventBroker (kaa5070)

   • Port: ip_port_of_EventBroker (7081)

   • Transport Type: DIRECT

   • Broker Version: Advanced

   • XA Enabled: false

4. Select WebSphere MQ Topic Destinations and click New. Enter the following attributes and click OK.

   • Name: SalesForecast_EB_IP

   • JNDI Name: jms/eb/ip/SalesForecast

   • Base Topic Name: SalesForecast_EB_IP

5. Select Servers in the navigation bar and select your server.

6. Select Message Listener Service in the Additional Properties table.

7. Click Listener Ports.

8. Click New, enter the following attributes and click OK.

   • Name: SalesForecastLP_EB_IP

   • Connection Factory JNDI name: jms/eb/ip/WholesaleTCF

   • Destination JNDI name: jms/eb/ip/SalesForecast

9. Select Application Servers > whostx > Message Listener Service > Custom Properties and set the following parameters:

   • Name: non.asf.receive.timeout

   • Value: >0 (for example: 5000)

10. Save the configuration and restart the server.

When you install the application, make sure that you bind the listener port to the message-driven bean.

9.7.2 Prepare the publisher application

Example 9-3 shows a simple servlet that can be used to publish information to a topic. The input to the servlet includes the following parameters:

• host: The host IP address of the broker

• port: The port number for the Real-timeOptimizedFlow node

• topic: The pub/sub topic. This must match the topic specified as the base topic name in the target application's topic destination definition.

• message: The information to publish.

Example 9-3: Publishing to a topic

import java.io.IOException; import javax.jms.JMSException; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicPublisher; import javax.jms.TopicSession; import javax.servlet.Servlet; import javax.servlet.ServletException; import javax.servlet.http.HttpServlet; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import com.ibm.mq.jms.MQTopicConnectionFactory; /** * @version 1.0 * @author */ public class TestEB extends HttpServlet implements Servlet { /** * @see javax.servlet.http.HttpServlet#void (javax.servlet.http.HttpServletRequest, javax.servlet.http.HttpServletResponse) */ public void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { try { String host = req.getParameter("host"); int port = Integer.parseInt(req.getParameter("port")); String topicName = req.getParameter("topic"); String message = req.getParameter("message"); MQTopicConnectionFactory mqtcf = new MQTopicConnectionFactory(); mqtcf.setHostName(host); mqtcf.setPort(port); mqtcf.setTransportType(com.ibm.mq.jms.JMSC.MQJMS_TP_DIRECT_TCPIP); // Create a topic connection TopicConnection conn = mqtcf.createTopicConnection(); // Don't forget to start the connection conn.start(); // Create a topic session from the connection TopicSession sess = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); // Create the Topic Topic t = sess.createTopic(topicName); // Now that we have the Topic, we can start the main loop of the // program, remembering to create either the TopicPublisher or the // TopicSubscriber on the first run round the loop TopicPublisher pub = sess.createPublisher(t); TextMessage msg = sess.createTextMessage(); // ...set the message content... msg.setText(message); // ...and publish it pub.publish(msg); System.out.println("Message published"); pub.close(); sess.close(); conn.close(); } catch (NumberFormatException e) { System.out.println(e.getMessage()); e.printStackTrace(); } catch (JMSException e) { e.printStackTrace(); e.printStackTrace(); } } /** * @see javax.servlet.http.HttpServlet#void (javax.servlet.http.HttpServletRequest, javax.servlet.http.HttpServletResponse) */ public void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { } }

There are no deployment considerations for this application. It is a simple Web application and doesn't require any JMS resource definitions.

9.7.3 Build and deploy the message flow

The message flow for this scenario only requires one node (Figure 9-33). The Real-timeOptimizedFlow node simply facilitates the communication between the clients (both those that publish and those that retrieve information).

Figure 9-33: Real-time message flow

The node defines the port it will operate on. Clients who are publishing information use this port to send information to. Message-driven bean clients define this port in the topic connection factory (Figure 9-34).

Figure 9-34: Real-timeOptimizedFlow node settings

Note that a topic is not defined. The topic is specified by the clients so there must be some agreement between them to define a common name.

Deployment of the message flow is handled in the same manner as described in "Deploy the message flow" on page 248.