// The startup process is a tedious bootstrap

// process until we get the TopicPublisher object

Properties env = new Properties();

InitialContext jndi = new InitialContext(env);

TopicConnectionFactory factory =

  (TopicConnectionFactory) jndi.lookup("MyBroker");

TopicConnection conn =

  factory.createTopicConnection(username, password);

TopicSession pubSess =

  conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

Topic topub = (Topic)jndi.lookup("My topic");



TopicPublisher publisher = pubSess.createPublisher(topub);

conn.start();



// Now, let's send a message!



StreamMessage mesg = pubSess.createStreamMessage();

mesg.writeString("something important");

mesg.writeFloat( 123.45 );

// complete writing your message content

// mesg.setJMSReplyTo( replyTopic );

publisher.publish(mesg, DeliveryMode.PERSISTENT,

                  Message.DEFAULT_PRIORITY, 18000);

// Provide an implementation of the listener



public class MyListener implements MessageListener {

  public MyListener () { }



  // This is the only method we need to implement

  public void onMessage (Message mesg) {

    TextMessage tmesg = (TextMessage) mesg;

    String text = tmesg.getText();

    // do something ...

  }

}



// The startup process is a tedious bootstrap

// process until we get the TopicSubscriber object.

// Then we set the listener as an instance of the

// MyListener class



Properties env = new Properties();

InitialContext jndi = new InitialContext(env);

TopicConnectionFactory factory =

  (TopicConnectionFactory) jndi.lookup("MyBroker");

TopicConnection conn =

  factory.createTopicConnection(username, password);

TopicSession subSess =



  conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

Topic tosub = (Topic)jndi.lookup("Interesting topic");

TopicSubscriber subscriber = subSess.createSubscriber(tosub);

subscriber.setMessageListener(new MyListener());

conn.start();

 

Code snippets in Listings 10.3 and 10.4 illustrate how to work with point-to point-message queues. Notice how strikingly similar they are compared with listings 10.1 and 10.2.

Listing 10.3. A JMS queue sender in action

// The startup process is a tedious bootstrap

// process until we get the QueueSender object



Properties env = new Properties();

InitialContext jndi = new InitialContext(env);

QueueConnectionFactory factory =

  (QueueConnectionFactory) jndi.lookup("MyBroker");

QueueConnection conn =

  factory.createQueueConnection(username, password);

QueueSession sendSess =

  conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

Queue toque = (Queue)jndi.lookup("My queue");

QueueSender sender = sendSess.createSender(toque);

conn.start();



// Now, let's send a message!



StreamMessage mesg = sendSess.createStreamMessage();

mesg.writeString("something important");

mesg.writeFloat( 123.45 );

// complete writing your message content

// mesg.setJMSReplyTo( replyTopic );

sender.send(mesg, DeliveryMode.PERSISTENT,

                  Message.DEFAULT_PRIORITY, 18000);

Listing 10.4. A JMS queue receiver in action

// Provide an implementation of the listener

public class MyListener implements MessageListener {

  public MyListener () { }



  // This is the only method we need to implement

  public void onMessage (Message mesg) {

    TextMessage tmesg = (TextMessage) mesg;

    String text = tmesg.getText();

    // do something ...

  }

}



// The startup process is a tedious bootstrap

// process until we get the QueueReceiver object.

// Then we set the listener as an instance of the

// MyListener class



Properties env = new Properties();

InitialContext jndi = new InitialContext(env);

QueueConnectionFactory factory =

  (QueueConnectionFactory) jndi.lookup("MyBroker");

QueueConnection conn =

  factory.createQueueConnection(username, password);

QueueSession recSess =

  conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

Queue recque = (Queue)jndi.lookup("Interesting topic");

QueueReceiver receiver = recSess.createReceiver(recque);

receiver.setMessageListener(new MyListener());

conn.start();

10.2.4 Combine the Two Messaging Models

The publish-subscribe and point-to-point messaging models often work together. For example, in a marketplace application, sellers send out multicast messages to buyers who subscribe to certain topics. Interested buyers can get back to the seller via the specified queues.

Although JMS is a powerful and standard tool for enterprise messaging, it is designed for the J2EE platform. Most JMS providers are hence too heavy for mobile applications. The next two sections introduce two mobile JMS or JMS-like solution providers: iBus//Mobile from Softwired and WebSphere MQ Everyplace from IBM.

10.3 Mobile JMS from iBus//Mobile

iBus//Mobile (v3.0) is a JMS-based mobile MOM solution. It contains a standalone JMS server that runs on gateway computers (PCs). The real value of iBus//Mobile is its extensive clientside library that runs on mobile devices and supports all popular mobile network protocols.

iBus//Mobile provides lightweight JMS client libraries for all J2ME platforms and a special JMS-like lightweight client for the MIDP. Nonprogrammable devices with native SMS, email, or WAP clients are also supported.

The iBus//Mobile server manages message queues and topics. It can also serve as a gateway to fully JMS-compatible J2EE application servers. Now, let's look at iBus//Mobile's extensive client library.

10.3.1 J2ME JMS Clients

iBus//Mobile provides JMS client libraries for all J2ME platforms, including CDC, PersonalJava, and MIDP. Since the javax.jms package is supported, any existing JMS application can be ported to mobile devices with minimum effort. However, since the iBus//Mobile client does not run in the J2EE environment, it does not have access to JNDI services. Hence, any JNDI-dependent initialization code must be changed to conform to the iBus//Mobile API.

Listing 10.5 demonstrates how to initialize a JMS connection from the client side. The ProtocolStackRegistry class specifies the protocol-dependent message transport, iBus//Mobile gateway URL, and QoS settings. The code snippet is based on the MIDP Queue Producer sample distributed with iBus//Mobile download.

Listing 10.5. iBus//Mobile JMS connection initialization

private QueueConnectionFactory qcf_;

private QueueConnection connection_;

private QueueSession session_;

private Queue queue_;

private QueueSender sender_;

public TextMessage msg_;

public String url_ = "socket://gatewayIP:8738";



public void initJms() {

  try {

    ProtocolStackRegistry.registerQos(

          "qos-tcp",

          url_,

          new Hashtable(),

          "mySessionId"

    );

    Hashtable jndiEnv = new Hashtable();

    jndiEnv.put(Context.INITIAL_CONTEXT_FACTORY,

"ch.softwired.mobileibus.jndi.InitialContextFactory"

    );

    jndiEnv.put(

  ch.softwired.mobileibus.jndi.Context.QOS_NAME,

      "qos-tcp");

    Context ctx = new InitialContext(jndiEnv);



    QueueConnectionFactory qcf_ =

      (QueueConnectionFactory)ctx.lookup(

           "QueueConnectionFactory");

    connection_ = qcf_.createQueueConnection();

    connection_.setClientID("myClientId");

    session_ =

      connection_.createQueueSession(false,

             QueueSession.AUTO_ACKNOWLEDGE);

    queue_ = (Queue)ctx.lookup("queue");

    sender_ = session_.createSender(queue_);

    connection_.start();

  } catch (Exception e) {

    System.out.println("Error: " + e.toString());

  }

}

10.3.2 The Lightweight Client for MIDP

Although iBus//Mobile's J2ME JMS client libraries run on MIDP, it is still too big for some low-end devices. For those devices, iBus//Mobile provides a lightweight JMS-like API called JmsLc. JmsLc also works on the MIDP platform. The use of JmsLc is very simple if you are familiar with basic concepts of JMS. Code Listings 10.6, 10.7, and 10.8 demonstrate how to initialize a new connection and send and receive messages using JmsLc. All code examples are based on the JmsLc tutorial sample application bundled in the iBus//Mobile download.

Listing 10.6. Initialize a JMS connection using JmsLc

public void initJms() {

  try {

    jmsLcConnector_ =

      new JmsLcConnector(url_, debugMode_);



  } catch (JmsLcException ex) {

    // Handle the error

  }

}

Listing 10.7. Send a message to the queue using JmsLc

public void sendToQueue() {

  JmsLcTextMessage aTextMessage = null;



  try {

    aTextMessage = jmsLcConnector_.createTextMessage();

    aTextMessage.setText(MESSAGE_TEXT_ + msgNumber_);

    jmsLcConnector_.sendToQueue(DESTINATION_NAME_,

                         aTextMessage, new Hashtable());

    msgNumber_++;

    displayMessageTextBox("Send message:", aTextMessage.getText());

  } catch (JmsLcException ex) {

    // Handle error

  }

}

Listing 10.8. Receive a message from the queue using JmsLc

public void receiveFromQueue() {

  try {

    IJmsLcMessage receivedMessage =

      jmsLcConnector_.receiveFromQueue(

        DESTINATION_NAME_, 10000, new Hashtable());



    if (receivedMessage != null) {

      if ( receivedMessage instanceof JmsLcTextMessage) {

        JmsLcTextMessage aTextMessage =

                 (JmsLcTextMessage)receivedMessage;

        displayMessageTextBox("Rec. message:", aTextMessage.getText());

    } else {

      // Handle non-text message

    }

  } catch (JmsLcException ex) {

     // Handle error

  }

}

10.3.3 Non-Programmable Clients

In addition to smart J2ME devices, iBus//Mobile middleware server also supports non-programmable devices. Since there is no way to make API calls, those devices use code words buried inside the message content to trigger appropriate delivery actions on the server side. We can define special mappings between message prefixes and message queues in the iBus//Mobile serverside configure file. For example, we can associate the SX prefix to a queue that is connected to a stock quote producer. In this case, if a user sends "SX IBM" in an SMS message, he will receive a return SMS message of the current IBM stock quote. iBus//Mobile server supports the following native messaging clients.

• SMS: The iBus//Mobile server connects to an SMS service center operated by the wireless carrier or an SMS-enabled smart phone via a cable.

• Email: The iBus//Mobile server connects to SMTP, POP3, and IMAP mail servers.

• WAP: Through a WAP servlet residing on a WML server, iBus//Mobile receives messages from WML form submissions and sends messages as WML responses to WAP clients.

For more information on iBus//Mobile's native device/protocol support, please refer to its documentation.

10.4 The IBM WebSphere MQ Everyplace

The IBM WebSphere MQ Everyplace (or, MQseries Everyplace: WMQe) is another mobile MOM server. The version of WMQe covered in this book is v2.0. It provides a set of JMS-like APIs in the com.ibm.mqe package. The WMQe API supports features beyond the JMS specification, such as message compression and built-in encryption. In addition, it supports a rich set of server administration APIs. For example, it allows the developer to manage queues and server settings programmatically. WMQe supports JMS through an optional adapter JAR file.

10.4.1 A Truly Mobile MOM Solution

WMQe runs on Windows, Linux, and many flavors of UNIX OSs. But like the IBM SMF  and DB2e , its most distinguished feature is that the WMQe server also runs on mobile devices! This allows the ultimate mobility for the middleware server and the mobile network it serves. WMQe runs natively on major mobile platforms, including PocketPC, Symbian OS, Palm OS, Embedded Linux, and QNX. It also runs on any device that supports PersonalJava or IBM's WebSphere Micro Environment. One of those Java environments is required for the Java API to work.

The portable WMQe server allows a TV settop box or a PDA to be the messaging gateway for other pervasive devices in the in-home or in-hand network. WMQe can be integrated with SMF and DB2e to build sophisticated mobile enterprise applications. As we discussed in section 4.5, locally placed mobile middleware can drastically improve overall system performance, availability, and reliability.

Now, let's have a look at the HelloWorld example bundled in the WMQe distribution. It covers some simple administration APIs of WMQe and highlights the differences between WMQe and JMS.

10.4.2 HelloWorld Code Walk Through

The HelloWorld example demonstrates how to use a queue manager to send and receive simple messages. The underlying persistence mechanism for messages is the local file system. The queue manager and its queues (message, system, and administration) are all stored in a directory. First, we have to configure the queue manager. Listing 10.9 illustrates how to set up and initialize the appropriate file system directory and files for the queue manager.

Listing 10.9. Configure the queue manager

String queueManagerName = "HelloWorldQM";

String baseDirectoryName = "./QueueManagers/" + queueManagerName;



// Create all the configuration information needed

// to construct the queue manager in memory.

MQeFields config = new MQeFields();



// Construct the queue manager section parameters.

MQeFields queueManagerSection = new MQeFields();



queueManagerSection.putAscii(MQeQueueManager.Name,

                             queueManagerName);

config.putFields(MQeQueueManager.QueueManager,

                 queueManagerSection);



// Construct the registry section parameters.

// In this example, we use a public registry.

MQeFields registrySection = new MQeFields();



registrySection.putAscii(MQeRegistry.Adapter,

       "com.ibm.mqe.adapters.MQeDiskFieldsAdapter");

registrySection.putAscii(MQeRegistry.DirName,

       baseDirectoryName + "/Registry");



config.putFields("Registry", registrySection);



// Construct a queue manager configuration

// utility object.

MQeQueueManagerConfigure configurator =

  new MQeQueueManagerConfigure(config,

    "com.ibm.mqe.adapters.MQeDiskFieldsAdapter" +

    ":" + baseDirectoryName + "/Queues");



// Define a queue manager.

configurator.defineQueueManager();



// Define some queues on the queue manager.

configurator.defineDefaultAdminQueue();

configurator.defineDefaultAdminReplyQueue();

configurator.defineDefaultDeadLetterQueue();

configurator.defineDefaultSystemQueue();



// Close the queue manager configuration utility.

configurator.close();

After the underlying disk files are properly initiated and configured, we can start the queue manager (Listing 10.10). We have to pass the configured directory to the MQeQueueManager object.

Listing 10.10. Start the queue manager

String queueManagerName = "HelloWorldQM";

String baseDirectoryName = "./QueueManagers/" + queueManagerName;



// Create all the configuration information

// needed to construct the queue manager in memory.

MQeFields config = new MQeFields();



// Construct the queue manager section parameters.

MQeFields queueManagerSection = new MQeFields();



queueManagerSection.putAscii(MQeQueueManager.Name,

                             queueManagerName);

config.putFields(MQeQueueManager.QueueManager,

                 queueManagerSection);



// Construct the registry section parameters.

// In this example, we use a public registry.

MQeFields registrySection = new MQeFields();



registrySection.putAscii(MQeRegistry.Adapter,

    "com.ibm.mqe.adapters.MQeDiskFieldsAdapter");

registrySection.putAscii(MQeRegistry.DirName,

         baseDirectoryName + "/Registry");



config.putFields("Registry", registrySection);



myQueueManager = new MQeQueueManager();

myQueueManager.activate(config);

Now, using the MQeQueueManager instance, the client application can create queues and send out messages. Listing 10.11 demonstrates both sending and receiving messages via the system default queue.

Listing 10.11. Send and receive a message via the default queue

public void put() throws Exception {



  MQeMsgObject msg = new MQeMsgObject();



  // Add my hello world text to the message.

  msg.putUnicode("myFieldName", "Hello World!");



  myQueueManager.putMessage(queueManagerName,

    MQe.System_Default_Queue_Name, msg, null, 0L);

}



public void get() throws Exception {



  MQeMsgObject msg = myQueueManager.getMessage(queueManagerName,

                                  MQe.System_Default_Queue_Name,

                                                null, null, 0L);



  if (msg != null) {

    if (msg.contains("myFieldName")) {

      String textGot = msg.getUnicode("myFieldName");

      System.out.println(

        "Message contained the text '" + textGot + "'");

    }

  }

}

The shutdown process involves closing the queue manager object and removing the underlying files (Listing 10.12).

Listing 10.12. Shutdown the queue manager and delete underlying files

// Close the queue manager

myQueueManager.closeQuiesce(QUIESCE_TIME);

myQueueManager = null;



// Delete the configured directory.

String queueManagerName = "HelloWorldQM";

String baseDirectoryName = "./QueueManagers/" + queueManagerName;



// Create all the configuration information needed

// to construct the queue manager in memory.

MQeFields config = new MQeFields();

// Construct the queue manager section parameters.

MQeFields queueManagerSection = new MQeFields();



queueManagerSection.putAscii(MQeQueueManager.Name,

                             queueManagerName);

config.putFields(MQeQueueManager.QueueManager,

                 queueManagerSection);



// Construct the registry section parameters.

// In this example, we use a public registry.

MQeFields registrySection = new MQeFields();



registrySection.putAscii(MQeRegistry.Adapter,

       "com.ibm.mqe.adapters.MQeDiskFieldsAdapter");

registrySection.putAscii(MQeRegistry.DirName,

       baseDirectoryName + "/Registry");



config.putFields("Registry", registrySection);



System.out.println("Queue manager " +

     queueManagerName + " will now be deleted.");



MQeQueueManagerConfigure configurator =

    new MQeQueueManagerConfigure();



configurator.activate(config,

        "com.ibm.mqe.adapters.MQeDiskFieldsAdapter"

        + ":" + baseDirectoryName + "/Queues");



// Remove the configured entries for the queues

// we defined.

configurator.deleteSystemQueueDefinition();

configurator.deleteDeadLetterQueueDefinition();

configurator.deleteAdminReplyQueueDefinition();

configurator.deleteAdminQueueDefinition();

// Remove the queue manager itself.

configurator.deleteQueueManagerDefinition();

configurator.close();

10.4.3 Storage Adapters

Adapters map WMQe's high-level messaging concepts to underlying, concrete implementations. In the above example, we used the MQeDiskFieldsAdapter class to persist message queues to disk files. A disk file system is not available on all J2ME devices. Even for devices that support disk files, they might not be the most efficient persistence storage. Other persistence mechanisms must be supported if WMQe is to run efficiently on a large variety of devices. All storage adapters inherit from the MQeAdapter class. Table 10.2 lists the methods the storage adapters could override.

WMQe bundles the following standard storage adapters (Table 10.3). Custom storage adapters can be developed by directly subclassing the MQeAdapter class.

10.4.4 The Administration Queue

Any queue manager can be configured at runtime through a special administration queue. Basically, the user or developer can send specially formatted messages to the administration queue and the queue manager will act accordingly.

For example, in the Hello World sample, we only demonstrated how to pass messages through a local queue. Through the administration queue, we can configure the queue manager to manipulate a remote queue or listen for incoming messages at a specified server port (Listing 10.13). We can see that WMQe provides convenient ways to encrypt and compress remote messages for the best security and performance.

Listing 10.13. Administration messages for remote queues and message listeners

// Create a connection from this queue manager

// to a remote queue manager

public void createConnection (String onQMgr,

            String name, String parms,

            String options) throws Exception {



  // TCP/IP adapter

  String adapter = "com.ibm.mqe.adapters.MQeTcpipHistoryAdapter";

  // Or MIDP HTTP adapter for MIDP clients

  // String adapter = "com.ibm.mqe.adapters.MQeMidpHttpAdapter";



  MQeConnectionAdminMsg msg = new MQeConnectionAdminMsg();

  MQeFields msgTest = primeAdminMsg(onQMgr, msg);

  msg.setName(name);



  // Set the admin action to create a new queue

  msg.create(adapter, parms, options,

      "DefaultChannel", "Route to: " + name);

  MQeAdminMsg respMsg = processAdminMsg(onQMgr, msg);

  checkAdminReply(respMsg);

}



// Create a remote listener for the queue manager

public void createListener (String listenerName) throws Exception {

  // Part 1 - Create the listener



  MQeCommunicationsListenerAdminMsg msg =

    new MQeCommunicationsListenerAdminMsg();

  MQeFields msgTest =

    primeAdminMsg(queueManager.getName(), msg);



  String listenAdapter =

    "com.ibm.mqe.adapters.MQeTcpipHistoryAdapter";

  int listenPort = 8092;



  msg.setName(listenerName);

  msg.create(listenAdapter, listenPort);



  // Send the admin message, wait for a reply

  // and check for success.

  MQeAdminMsg respMsg = processAdminMsg(queueManager.getName(), msg);

  checkAdminReply(respMsg);



  // Part 2 - Start the listener



  msg = new MQeCommunicationsListenerAdminMsg();

  MQeFields msgTest2 = primeAdminMsg(queueManager.getName(), msg);



  msg.setName(listenerName);

  msg.start();



  respMsg =

    processAdminMsg(queueManager.getName(), msg);

  checkAdminReply(respMsg);

}



// Create a remote queue

public void createRemoteQueue( String onQMgr,

        String qm, String q, boolean sync,

        String compressor, String cryptor)

                             throws Exception {



  MQeRemoteQueueAdminMsg msg = new MQeRemoteQueueAdminMsg();



  primeAdminMsg(onQMgr, msg);

  msg.setName(qm, q);



  // Define the parameters of the new queue using

  // an MQeFields object

  MQeFields parms = new MQeFields();



  parms.putUnicode(MQeQueueAdminMsg.Queue_Description,

    "Remote Queue " + q + " on qm " + onQMgr);



  // Add cryptor and/or compressor if required

  if (compressor != null) {

    parms.putAscii(MQeQueueAdminMsg.Queue_Compressor, compressor);

  }



  if (cryptor != null) {

    parms.putAscii(MQeQueueAdminMsg.Queue_Cryptor, cryptor);

  }



  // Set queue as being accessed synchronously

  // or asynchronously

  if (sync) {

    parms.putByte(MQeQueueAdminMsg.Queue_Mode,

        MQeQueueAdminMsg.Queue_Synchronous);

  } else {

    parms.putByte(MQeQueueAdminMsg.Queue_Mode,

        MQeQueueAdminMsg.Queue_Asynchronous);

  }



  // Set the admin action to create a new queue

  // with these parameters and send it

  msg.create(parms);

  MQeAdminMsg respMsg = processAdminMsg(onQMgr, msg);

  checkAdminReply(respMsg);

}



// Two utility methods for sending the receiving

// administration messages



// Give the message a unique identifier

public MQeFields primeAdminMsg(String onQMgr,

             MQeAdminMsg msg) throws Exception {

  // Set the target queue manager that will process

  // this message, who to reply to and the 'style'



  // of the message to be a request

  msg.setTargetQMgr(onQMgr);

  msg.putInt(MQe.Msg_Style, MQe.Msg_Style_Request);

  msg.putAscii(MQe.Msg_ReplyToQ, MQe.Admin_Reply_Queue_Name);

  msg.putAscii(MQe.Msg_ReplyToQMgr, onQMgr);



  //  Setup the correl id so we can match the

  // reply to the request

  msg.putArrayOfByte(MQe.Msg_CorrelID,

      Long.toString(MQe.uniqueValue(), 16).getBytes());



  // Ensure matching response message is retrieved

  MQeFields msgTest = new MQeFields();



  msgTest.putArrayOfByte(MQe.Msg_CorrelID,

      msg.getArrayOfByte(MQe.Msg_CorrelID));



  // Return the unique filter for this message

  return msgTest;

}



// Process the message

public MQeAdminMsg processAdminMsg(String onQMgr,

                MQeAdminMsg msg) throws Exception {

  // Setup the match field to check for a matching

  // reply message. The correl id is used as the

  // match parameter.

  MQeFields msgTest = new MQeFields();



  msgTest.putArrayOfByte(MQe.Msg_CorrelID,

      msg.getArrayOfByte(MQe.Msg_CorrelID));



  //  Put the admin message to the admin queue and

  // return the reply

  queueManager.putMessage(onQMgr,

      MQe.Admin_Queue_Name, msg, null, 0);

  MQeAdminMsg respMsg =

      (MQeAdminMsg) waitForReply(onQMgr,

        MQe.Admin_Reply_Queue_Name, msgTest);



  return respMsg;

}

10.4.5 Communications Adapters

In Listing 10.13, we use communication adapters, such as the MQeTcpipHistoryAdapter, to handle underlying message transport over the network. WMQe supports common mobile network communication protocols as message transports. Abstract class MQeCommunicationsAdapter defines a common interface for all communication adapters. Abstract methods in MQeCommunicationsAdapter are listed in Table 10.4. The standard WMQe distribution supplies implementations for several communication adapters support HTTP, TCP/IP, and UDP protocols (Table 10.5).

In theory, WMQe runs on MIDP devices with limited options. But in reality, few MIDP devices have the required memory space and CPU power to support WMQe queue managers. Also, as we have discussed, MIDP queue managers cannot receive messages. A much better approach is to have the pervasive device send and receive messages through its native messaging programs such as the SMS and IM clients. They can then utilize WMQe queue managers running on more capable devices in the local network. WMQe does not provide standard adapters to handle native message transports. If you need to handle SMS/IM messages in your queue managers, you need to write a custom communication adapter, possibly using the J2ME Wireless Messaging API.