Implementing Messaging in Java and .NET

The discussion regarding messaging continues with defining messaging semantics under Java and .NET.

Messaging with Java

Java offers a mature standards-based approach to messaging via the Java Message Service (JMS) interface, [JMS]. Multiple industry vendors implement their messaging middleware in compliance with this standard to allow Java applications to asynchronously interact in a coherent manner. Any Java application, standalone or enterprise, can leverage a JMS-based infrastructure. For enterprise applications, Java EE developers may leverage the message-driven bean, EJB, introduced in the EJB 2.0 specification, [EJBSpec], which simplifies implementation of the messaging interaction by delegating some of the work to the underlying application server. Depending on the application design it might be best to select point-to-point application integration with javax.jms.Queue or the publish-subscribe integration with javax.jms.Topic libraries. Examples that follow highlight implementation details of building messaging with message-driven beans and a standalone Java application. In both cases implementation is based on JMS standards.

Using Java Message Service(JMS)

The series of three main steps that follow are outlined to show how to create a JMS-based application. This chapter demonstrates both publish-subscribe and point-to-point integration that share the same way of creating a connection to the JMS destination and a session used to send and receive messages.

ConnectionFactory and Destination Look Up

The first step involves a Java Naming and Directory Interface (JNDI) API, [JNDI], call to look up the ConnectionFactory and Destination. A javax.jms.TopicConnectionFactory or javax.jms.QueueConnectionFactory is then used to create a connection to a message queue.

The first step is to create an initial context, javax.naming.InitialContext, object:

InitialContext context = new InitialContext(); 

A NamingException should be caught when creating the InitialContext. Next is to look up the queue connection factory and a JMS destination:

QueueConnectionFactory connectionFactory    =(QueueConnectionFactory)context.lookup("QueueCF"); Queue destination =(Queue) context.lookup(destName); 

For publish-subscribe, the topic connection lookup calls are similar:

TopicConnectionFactory connectionFactory    =(TopicConnectionFactory)context.lookup("TopicCF"); Topic destination =(Topic) jndiContext.lookup(destName); 

Typically, a ServiceLocator component is used to look up ConnectionFactory and get Topic or Queue.

Create a Connection and a Session

Now the connection factory is used to create a JMS connection to send and receive messages from a JMS provider:

 QueueConnection connection =      connectionFactory.createQueueConnection(); 

When creating a session, false should be passed as a parameter, indicating that session is not transacted. The session automatically acknowledges received messages by passing the Session.AUTO_ACKNOWLEDGE parameter:

QueueSession session =   connection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE); 

Creating a connection and the JNDI context operations can throw exceptions, thus they have to be performed in a try catch block.

Asynchronously Sending and Receiving Messages

A MessageProducer is used to send multiple messages to a destination:

MessageProducer producer = session.createProducer(destination); 

Now it's time to create a message, javax.jmx.TextMessage, and populate it with XML-based PurchaseOrder data to transfer an XML message to the destination:

message = session.createTextMessage(); message.setText(purchaseOrder)); producer.send(message); 

For receiving a message, a javax.jms.MessageConsumer needs to be created:

MessageConsumer consumer = session.createConsumer(destination); 

Also a message listener, TextListener, should be created for the MessageConsumer:

TextListener listener = new TextListener(); consumer.setMessageListener(listener); 

Next is to start the connection to allow message delivery and to receive an answer message whenever asynchronous processing of the original request is over, as shown in Listing 9-1:

Listing 9-1. Answer to the Original Request

connection.start(); System.out.println("To end program, enter Q or q, then <return>"); inputStreamReader = new InputStreamReader(System.in); while(!((answer == 'q') ||(answer == 'Q'))) {     try {         answer =(char) inputStreamReader.read();     } catch(IOException e) {         System.out.println("I/O exception: " + e.toString());     } } 

The TextListener implements the onMessage method that transforms the incoming message into a TextMessage, as seen in Listing 9-2:

Listing 9-2. TextListener onMessage Method

public void onMessage(Message message) {    TextMessage msg = null;    try {      if(message instanceof TextMessage) {          msg =(TextMessage) message;          System.out.println("Reading message: " +              msg.getText());      } else {      ... } 

Clean Up

After the message has been sent/received, the JMS connection must be closed:

connection.close(); 

Closing a connection should be enclosed in a finally block to ensure that any runtime exceptions do not prevent the program from releasing resources. The session and MessageConsumer are closed automatically.

Deployment Descriptors

When deploying Java components, a deployment descriptor needs to be created, which is an XML file that specifies a connection factory and destination.

This chapter demonstrates via samples how to exchange SOAP messages over JMS.

Using Java EE Message-Driven Beans

A message-driven bean is a listener that allows reliable message consumption. With message-driven beans the underlying application server provides a listener service that monitors JMS destinations and passes incoming messages to message-driven beans. Each listener monitors a JMS queue or a JMS topic destination and when a message arrives it automatically fires an event to message listeners. A message-driven bean implements the javax.jms.MessageListener's onMessage() method, which is invoked when the message arrives.

A Java EE application server transparently manages JMS listeners. This model contributes toward better application scalability. The underlying application server recycles these short-lived EJBs, thus reducing the processing cycle and pools these beans to allow concurrent message processing.

As an EJB, a message-driven bean needs to implement the ejbCreate and ejbRemove methods. In addition setMessageDrivenContext needs to be implemented to manage a transactional context. Either container-managed transactions or bean-managed transactions can be specified with message-driven beans. As already mentioned, the bean also has to implement the onMessage methods. The onMessage method is very similar to the one seen in the TextListener class, shown earlier. The only difference is that when an exception occurs, a message-driven bean calls the MessageDrivenContext.setRollbackOnly method to roll back the transaction. Listing 9-3 displays the onMessage method of the message-driven EJB:

Listing 9-3. Message-Driven Bean's onMessage Method

public void onMessage(Message msg) {        TextMessage txtMsg = null;        try {            if(msg instanceof TextMessage) {                txtMsg =(TextMessage) inMessage;            } else {                // continue with other type checking            }        } catch(JMSException exe1) {          // Rollback transaction        } catch(Throwable exe2) {          // Manage runtime errors        }     } 

Messaging Under .NET

Building asynchronous workflow under .NET in most cases relies on integration with MSMQ [MSMQReference]. The .NET Framework provides System.Messaging classes used to post messages to a queue for asynchronous processing. The following is a set of steps necessary to send or receive a message:

Specifying a Queue

There are several options to use under .NET to specify a queue. They include explicitly specifying a queue location path, specifying a Format Name that refers to the actual message queue, or using a label while the actual path is configured via the MSMQ admin interface. Here is an example of specifying a Format Name:

String queuePath = @".\Private$\ReplenishStockOrders"; 

Creating a Message Queue

In the second step an instance of the System.Messaging.MessageQueue is created:

MessageQueue msgQueue = new MessageQueue(queuePath); 

All messages can also be marked as Recoverable to guarantee reliable messaging in case of a rollback:

msgQueue.DefaultPropertiesToSend.Recoverable = true; 

Sending and Receiving a Message

To send a message, it is first converted into an XML format and then sent to the queue, as shown in Listing 9-4:

Listing 9-4. Sending a Message to MSMQ

// Convert PurchaseOrder to the XML format String message = order.toXML(); // Send order to the message queue msgQueue.Send(message); 

On the receiving end the following operation can be performed to get the incoming message, as in Listing 9-5:

Listing 9-5. Receiving a Message from MSMQ

// Check if the queue exists if( MessageQueue.Exists(msgQueue) { try {   // Receive the inventory check reques   Message msg = msgQueue.Receive();   String request =(String)msg.Body; } 

A status notification can then be sent back to the queue:

msgQueue.Send("Purchase Order received"); 

There are three message formatters available in .NET Framework, XMLMessageFormatter, BinaryMessageFormatter, and ActiveXMessageFormatter. The binary formatter is common to interchange images and large binary data, given that this formatter generates a serialization stream with a small footprint. The XML formatter is useful to serialize and deserialize objects using human readable XML format. The ActiveXMessageFormatter is used to interchange messages compatible with the MSMQ ActiveX Component.

To asynchronously receive messages, the BeginReceive() and EndReceive() methods of the System.Messaging.MessageQueue can be used. One needs to implement an event handler, System.Messaging.receiveCompleteEventHandler, to notify the program when the message retrieval is complete. This logic of asynchronous message retrieval is shown in Listing 9-6.

Listing 9-6. Asynchronous Message Retrieval with MSMQ

using System; using System.Messaging; public class AsyncMsgQueue{ MessageQueue msgQueu; public void sendMsg(){     string msgQueuePath = ".\\purchaseOrders";     // Make sure to validate that the message queue exists     // Connect to the queue     msgQueue = MessageQueue(msgQueuePath);     // Pass PurchaseOrder to the XML Formatter     ((XmlMessageFormatter) msgQueue.Formatter).TargetTypeNames =          new string []{"PurchaseOrder"};     // Asynchronously receive messages     msgQueue.ReceiveCompleted +=          new ReceiveCompletedEventHandler( AsyncReceiveCompleted ); // Beginning of the asynchronous read     msgQueue.BeginReceive(); } // Once the read is complete the AsyncReceiveCompleted is invoked void AsyncReceiveCompleted(Object src, ReceiveCompletedEventArgs args) {     try {         PurchaseOrder po =           (PurchaseOrder)msgQueue.EndReceive(args.AsyncResult).Body;         // For a generic message:         // Message msg = msgQueue.EndReceive( args.AsyncResult );         // To restart the asynchronous read         msgQueue.BeginReceive();     } catch( MessageQueueException exe ) {         // Handle exception     } } 

Cleaning Up

Closing the message queue ensures that all system resources are released accordingly:

msgQueue.Close(); 

Architect's Note

This section outlined details of sending and receiving asynchronous messages in both Java and .NET. It is important to note that for most enterprise applications, transactions are a big part of the inter-application workflow. Transactions ensure consistency of the data sent across applications. Both the Java and .NET platforms are quite flexible in their transactional support. Transactional semantics of Java EE and .NET and transactional interoperability are discussed in a separate chapter.

Addressing alternative integration flows with accurate error handling is another aspect of asynchronous integration with messaging. It is often the case that a message acknowledgement has to be timed out after a certain period of time or a submitted message has to be followed up with retry logic. These design aspects help to adequately address applications' non-functional requirements.