JBoss Messaging 1.2 User's Guide

JBoss Messaging 1.2.GA will become part of both Application Server Platform (JBoss 4.2 series) and Service Integration Platform (JBoss ESB 4 series) as default JMS provider. Production support will be fully available for these plaforms and it will cover JBoss Messaging.

Red Hat is also working on introducing a development support program, which is different from production support and caters to companies who want to have a safety net even from the development phase. The estimated availability date of this support progam is April 2007.

JBoss Messaging provides:

JBoss Messaging consists of two major parts:

Other JBoss Messaging features include:

JBoss Messaging 1.2.0.GA Clustering provides the following features:

JBossMQ is the JMS implementation currently shipped within JBossAS. Since JBoss Messaging is JMS 1.1 and JMS 1.0.2b compatible, the JMS code written against JBossMQ will run with JBoss Messaging without any changes.

All the final features are available apart from:

Here's a brief overview of how clustering works in JBoss Messaging 1.2.

Clustered destinations (queues and topics) can be deployed at all or none of the nodes of the cluster. A JMS client uses HA JNDI to lookup the connection factory. A client side load balancing policy will automatically chose a node to connect to (This is similar to how EJB clustering chooses a node).

The JMS client has now made a connection to a node where it can create sessions, message producers and message consumers and browsers and send or consume messages, using the standard JMS api. When a distributed queue is deployed across the cluster, individual partial queues are deployed on each node.

When a message is sent from a message producer attached to a particular node to a distributed queue, a routing policy determines which partial queue will receive the message. By default the router will always pass the message to a local queue, if there is one, this is so we avoid unnecessary network traffic. If there is no local queue then a partial queue on a different node will be chosen by the router, by default this will be round robin between remote partial queues.

When a message is sent to a distributed topic while attached to a node, there may be multiple subscriptions on different nodes that need to receive the message. Depending on the number and location of subscriptions, the message may be multicast or unicast across the cluster so the other nodes can pick it up. (All group communication, unicast, multicast and group management is handled by JGroups.)

In the case of shared durable subscriptions, if a durable subscription with the same name exists on more than node, then only one of the instances needs to receive the message. Which one is determined by the same routing policy used to route messages to partial queues. All of this is accomplished without losing the reliability guarantees required by JMS.

Subscriptions (both durable and non durable) can be created on all nodes and will receive messages sent via any node. What happens if the consumers on one queue/subscription are faster/slower than consumers on another? Normally this would result in messages building up on that queue and fast consumers being starved of work on another, thus wasting CPU cycles on the node that could be put to good use. The most degenerate example is of a queue containing many messages then the consumers being closed on that queue. The messages might potentially remain stranded on the queue until another consumer attaches. A routing policy is no use here, since the messages have already been routed to the queuee and the consumers closed / slowed down after they were routed there. JBoss Messaging can deal with this problem by intelligently pulling messages from other less busy nodes, if it detects idle consumers on the fast node and slow consumers on another node.

Normally, persistent messages are persisted in a shared database which is shared by all nodes in the cluster. JBoss Messaging 1.2.1 will contain an option where you can choose to not persist persistent messages in a database, but instead to replicate them between nodes of the cluster. The idea here is the network IO on a fast network should be much faster than persisting to disk. This solution should also be more scalable since different nodes replicate their messages onto different other nodes - there is no "master node". If the messages are replicated onto sufficient nodes and the hardware is set-up with UPS, then we believe a comparable reliability guarantee to persisting messages to disk can be achieved. Of course, this won't be suitable for all situations, but you use the best tool for the job.

One of the fundamental Messaging Core building blocks is the "Post Office". A JBoss Messaging Post Office is message routing component, which accepts messages for delivery and synchronously forwards them to their destination queues or topic subscriptions.

There is a single Post Office instance per JBoss Messaging server (cluster node). Both queues and topics deployed on a JBoss Messaging node are "plugged" into that Post Office instance. Internally JBoss Messaging only deals with the concepts of queues, and considers a topic to just be a set of queues (one for each subscription). Depending on the type of subscription - durable or non-durable - the corresponding queue saves messages to persistent storage or it just holds messages in memory and discards them when the non-durable subscription is closed.

Therefore, for a JMS queue, the Post Office routes messages to one and only one core queue, depending on the queue name, whereas for a JMS topic, the Post Office routes a message to a set of core queues, one for each topic subscription, depending on the topic name.

Clustering across multiple address spaces is achieved by clustering Post Office instances. Each JBoss Messaging cluster node runs a Clustered Post Office instance, which is aware of the presence of all other clustered Post Offices in the cluster. There is an one-to-one relationship between cluster nodes and clustered Post Office instances. So, naturally, the most important piece of clustering configuration is the clustered Post Office service configuration, covered in detail below.

Clustered Post Office instances connect to each other via JGroups and they heavily rely on JGroups group management and notification mechanisms. JGroups stack configuration is an essential part of JBoss Messaging clustering configuration. JGroups configuration is only briefly addressed in this guide. Detailed information on JGroups can be found in JGroups release documentation or on-line at http://www.jgroups.org or http://wiki.jboss.org/wiki/Wiki.jsp?page=JGroups

When routing messages, a clustered Post Office has a choice of forwarding the message to local queues or remote queues, plugged into remote Post Office instances that are part of the same cluster. Local queues are usually preferred, but if a local queue is part of a distributed queue, has no consumers, and other local queues part of the same distributed queue have consumers, messages can be automatically redistributed, subject of the message redistribution policy in effect. This allows us to create distributed queues and distributed topics. Message redistribution configuration is another subject that we will insist on.

The JBoss Messaging release bundle (jboss-messaging-1.2.x.zip) will expand in a jboss-messaging-1.2.x directory that contains:

If you want to experiment with the latest developments you may checkout the latest code from the Messaging SVN trunk. Be aware that the information provided in this manual might then not be accurate. For the latest instructions on how to check out and build source code, please go to Messaging Development wiki page, specifically "Building and Running JBoss Messaging" section.

To run the server, execute the run.bat or run.sh script as appropriate for your operating system, in the $JBOSS_HOME/bin directory.

cd $JBOSS_HOME/bin
./run.sh -c messaging
   

A successful JBoss Messaging deployment generates logging output similar to:

....
01:44:48,317 WARN  [JDBCPersistenceManager]

JBoss Messaging Warning: DataSource connection transaction isolation should be
       READ_COMMITTED, but it is currently NONE.
       Using an isolation level less strict than READ_COMMITTED may lead to data
       consistency problems.
       Using an isolation level more strict than READ_COMMITTED may lead to deadlock.

01:44:50,450 INFO  [ServerPeer] JBoss Messaging 1.2.0.GA server [0] started
01:44:51,311 INFO  [ConnectionFactory] Connector bisocket://192.168.1.104:4457 has
leasing enabled, lease period 10000 milliseconds
01:44:51,311 INFO  [ConnectionFactory] [/ConnectionFactory, /XAConnectionFactory,
java:/ConnectionFactory, java:/XAConnectionFactory] started
01:44:51,441 INFO  [ConnectionFactory] Connector bisocket://192.168.1.104:4457 has
leasing enabled, lease period 10000 milliseconds
01:44:51,441 INFO  [ConnectionFactory] [/NonClusteredConnectionFactory,
/NonClusteredXAConnectionFactory, java:/NonClusteredConnectionFactory,
java:/NonClusteredXAConnectionFactory] started
01:44:51,681 INFO  [QueueService] Queue[/queue/DLQ] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,681 INFO  [QueueService] Queue[/queue/ExpiryQueue] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,732 INFO  [TopicService] Topic[/topic/testTopic] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,732 INFO  [TopicService] Topic[/topic/securedTopic] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,732 INFO  [TopicService] Topic[/topic/testDurableTopic] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,752 INFO  [QueueService] Queue[/queue/testQueue] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,752 INFO  [QueueService] Queue[/queue/A] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,762 INFO  [QueueService] Queue[/queue/B] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,762 INFO  [QueueService] Queue[/queue/C] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,882 INFO  [QueueService] Queue[/queue/D] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,882 INFO  [QueueService] Queue[/queue/ex] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,882 INFO  [QueueService] Queue[/queue/PrivateDLQ] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,892 INFO  [QueueService] Queue[/queue/PrivateExpiryQueue] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,892 INFO  [QueueService] Queue[/queue/QueueWithOwnDLQAndExpiryQueue] started,
fullSize=200000, pageSize=2000, downCacheSize=2000
01:44:51,892 INFO  [TopicService] Topic[/topic/TopicWithOwnDLQAndExpiryQueue] started,
fullSize=200000, pageSize=2000, downCacheSize=2000
01:44:51,942 INFO  [QueueService] Queue[/queue/QueueWithOwnRedeliveryDelay] started,
fullSize=200000, pageSize=2000, downCacheSize=2000
01:44:51,942 INFO  [TopicService] Topic[/topic/TopicWithOwnRedeliveryDelay] started,
fullSize=200000, pageSize=2000, downCacheSize=2000
01:44:51,952 INFO  [QueueService] Queue[/queue/testDistributedQueue] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:51,952 INFO  [TopicService] Topic[/topic/testDistributedTopic] started, fullSize=200000,
pageSize=2000, downCacheSize=2000
01:44:52,372 INFO  [ConnectionFactoryBindingService] Bound ConnectionManager
'jboss.jca:name=JmsXA,service=ConnectionFactoryBinding' to JNDI name 'java:JmsXA'
....
   

The release bundle contains a series of examples that should run "out of the box" and could be used to validate a new installation. Such an example sends a persistent JMS message to a queue called queue/testQueue.

To run the example and validate the installation, open an new command line window and set the JBOSS_HOME environment variable to point to the JBoss AS 4.x installation you've just installed Messaging on. Navigate to the folder where you extracted the release bundle and drill down to /examples/queue. Apache Ant must pe present in your path in order to be able to run the example.

setenv JBOSS_HOME=<your_JBoss_installation>
cd .../examples/queue
$ant

   

A successfull execution log output looks similar to:

$ ant
Buildfile: build.xml

identify:
[echo] ###########################################################################
[echo] #                       Running the QUEUE example                         #
[echo] ###########################################################################
[echo] The queue:      testQueue
[echo] The client jar: ../../../output/lib/jboss-messaging-client.jar

sanity-check:

init:
[mkdir] Created dir: c:\work\src\svn\messaging\docs\examples\queue\output
[mkdir] Created dir: c:\work\src\svn\messaging\docs\examples\common\output

compile:
[javac] Compiling 5 source files to c:\work\src\svn\messaging\docs\examples\common\output
[javac] Compiling 1 source file to c:\work\src\svn\messaging\docs\examples\queue\output

run:
   [java] Queue /queue/testQueue exists
   [java] The message was successfully sent to the testQueue queue
   [java] Received message: Hello!
   [java] The example connected to JBoss Messaging version 1.2.0.GA (1.2)
   [java]
   [java] #####################
   [java] ###    SUCCESS!   ###
   [java] #####################

BUILD SUCCESSFUL
Total time: 9 seconds

It is recommended to run all validation examples available in the example directory (queue, topic, mdb, stateless, etc.). In Chapter 7, Running the Examples, we will have a look at each of those examples.

Open an new command line. Set the JBOSS_HOME environment variable to point at a JBossAS 4.x installation. Navigate to the folder where you exploded the main archive and drill down to /examples/queue. You need to use Apache ant to execute the build.xml file.

Make sure the JBoss server reference by the JBOSS_HOME is started.

public class QueueExample extends ExampleSupport
{

   public void example() throws Exception
   {
      String destinationName = getDestinationJNDIName();

      InitialContext ic = null;
      ConnectionFactory cf = null;
      Connection connection =  null;

      try
      {
         ic = new InitialContext();

         cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");
         Queue queue = (Queue)ic.lookup(destinationName);
         log("Queue " + destinationName + " exists");

         connection = cf.createConnection();
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         MessageProducer sender = session.createProducer(queue);

         TextMessage message = session.createTextMessage("Hello!");
         sender.send(message);
         log("The message was successfully sent to the " + queue.getQueueName() + " queue");

         MessageConsumer consumer =  session.createConsumer(queue);

         connection.start();

         message = (TextMessage)consumer.receive(2000);
         log("Received message: " + message.getText());
         assertEquals("Hello!", message.getText());

         displayProviderInfo(connection.getMetaData());

      }
      finally
      {
         if(ic != null)
         {
            try
            {
               ic.close();
            }
            catch(Exception e)
            {
               throw e;
            }
         }

         // ALWAYS close your connection in a finally block to avoid leaks.
         // Closing connection also takes care of closing its related objects e.g. sessions.
         closeConnection(connection);
      }
   }

   private void closeConnection(Connection con)
   {
      try
      {
         if (con != null)
         {
            con.close();
         }
      }
      catch(JMSException jmse)
      {
         log("Could not close connection " + con +" exception was " + jmse);
      }
   }

   protected boolean isQueueExample()
   {
      return true;
   }

   public static void main(String[] args)
   {
      new QueueExample().run();
   }

}
    

In this example, a standalone Java client publishes a text-based JMS message to a topic and a single subscriber pulls the message off the queue.

Open an new command line. Set the JBOSS_HOME environment variable to point at a JBossAS 4.x installation. Navigate to the folder where you exploded the main archive and drill down to /examples/queue. You need to use Apache Ant to execute the build.xml file Make sure the JBoss server reference by the JBOSS_HOME is started.

public class TopicExample extends ExampleSupport
{
   public void example() throws Exception
   {
      String destinationName = getDestinationJNDIName();

      InitialContext ic = null;
      Connection connection = null;

      try
      {
         ic = new InitialContext();

         ConnectionFactory cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");
         Topic topic = (Topic)ic.lookup(destinationName);
         log("Topic " + destinationName + " exists");

         connection = cf.createConnection();
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         MessageProducer publisher = session.createProducer(topic);
         MessageConsumer subscriber = session.createConsumer(topic);

         ExampleListener messageListener = new ExampleListener();
         subscriber.setMessageListener(messageListener);
         connection.start();

         TextMessage message = session.createTextMessage("Hello!");
         publisher.send(message);
         log("The message was successfully published on the topic");

         messageListener.waitForMessage();

         message = (TextMessage)messageListener.getMessage();
         log("Received message: " + message.getText());
         assertEquals("Hello!", message.getText());

         displayProviderInfo(connection.getMetaData());

      }
      finally
      {
         if (ic != null)
         {
            try
            {
               ic.close();
            }
            catch(Exception e)
            {
               throw e;
            }
         }

         // ALWAYS close your connection in a finally block to avoid leaks.
         // Closing connection also takes care of closing its related objects e.g. sessions.
         closeConnection(connection);
      }
   }

   private void closeConnection(Connection con) throws JMSException
   {

      try
      {
         if (con != null)
         {
            con.close();
         }
      }
      catch(JMSException jmse)
      {
         log("Could not close connection " + con +" exception was " + jmse);
      }
   }

   protected boolean isQueueExample()
   {
      return false;
   }

   public static void main(String[] args)
   {
      new TopicExample().run();
   }

}
   

This example deploys a simple Stateless Session Bean that is used as a proxy to send and receive JMS messages in a managed environment.

public class StatelessSessionExampleBean implements SessionBean
{

   private ConnectionFactory cf = null;

   public void drain(String queueName) throws Exception
   {
      InitialContext ic = new InitialContext();
      Queue queue = (Queue)ic.lookup(queueName);
      ic.close();

      Session session = null;
      Connection conn = null;

      try
      {
         conn = getConnection();
         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
         MessageConsumer consumer = session.createConsumer(queue);
         Message m = null;
         do
         {
            m = consumer.receiveNoWait();
         }
         while(m != null);
      }
      finally
      {
         if (conn != null)
         {
            closeConnection(conn);
         }
      }
   }

   public void send(String txt, String queueName) throws Exception
   {
      InitialContext ic = new InitialContext();

      Queue queue = (Queue)ic.lookup(queueName);

      ic.close();

      Session session = null;
      Connection conn = null;

      try
      {
         conn = getConnection();

         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

         MessageProducer producer = session.createProducer(queue);

         TextMessage tm = session.createTextMessage(txt);

         producer.send(tm);

         System.out.println("message " + txt + " sent to " + queueName);
      }
      finally
      {
         if (conn != null)
         {
            closeConnection(conn);
         }
      }
   }

   public int browse(String queueName) throws Exception
   {
      InitialContext ic = new InitialContext();
      Queue queue = (Queue)ic.lookup(queueName);
      ic.close();

      Session session = null;
      Connection conn = null;

      try
      {
         conn = getConnection();
         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
         QueueBrowser browser = session.createBrowser(queue);

         ArrayList list = new ArrayList();
         for(Enumeration e = browser.getEnumeration(); e.hasMoreElements(); )
         {
            list.add(e.nextElement());
         }

         return list.size();
      }
      finally
      {
         if (conn != null)
         {
            closeConnection(conn);
         }
      }
   }

   public String receive(String queueName) throws Exception
   {
      InitialContext ic = new InitialContext();
      Queue queue = (Queue)ic.lookup(queueName);
      ic.close();

      Session session = null;
      Connection conn = null;

      // WARN! this is an anti-pattern - please see above warning

      try
      {
         conn = getConnection();
         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

         MessageConsumer consumer = session.createConsumer(queue);

         System.out.println("blocking to receive message from queue " + queueName + " ...");
         TextMessage tm = (TextMessage)consumer.receive(5000);

         if (tm == null)
         {
            throw new Exception("No message!");
         }

         System.out.println("Message " + tm.getText() + " received");

         return tm.getText();
      }
      finally
      {
         if (conn != null)
         {
            closeConnection(conn);
         }
      }
   }

   public Connection getConnection() throws Exception
   {

      Connection connection = null;

      try
      {
         connection = cf.createConnection();

         connection.start();
      }
      catch(Exception e )
      {
         if(connection != null)
         {
            closeConnection(connection);
         }
         System.out.println("Failed to get connection...exception is " + e);
         throw e;
      }

      return connection;
   }

   public void closeConnection(Connection con) throws Exception
   {
      try
      {
         con.close();
      }
      catch(JMSException jmse)
      {
         System.out.println("Could not close connection " + con +" exception was " + jmse);
         throw jmse;
      }
   }

   public void setSessionContext(SessionContext ctx) throws EJBException, RemoteException
   {
   }

   public void ejbCreate()
   {
      try
      {
         InitialContext ic = new InitialContext();

         cf = (ConnectionFactory)ic.lookup("java:/JmsXA");

         ic.close();
      }
      catch(Exception e)
      {
         e.printStackTrace();
         throw new EJBException("Initalization failure: " + e.getMessage());
      }
   }

   public void ejbRemove() throws EJBException
   {
      try
      {
         if(cf != null)
         {
            cf = null;
         }
      }
      catch(Exception e)
      {
         throw new EJBException("ejbRemove ", e);
      }
   }

   public void ejbActivate() throws EJBException, RemoteException
   {
   }

   public void ejbPassivate() throws EJBException, RemoteException
   {
   }

}
    

This example deploys a simple Message Driven Bean that processes messages sent to a test queue. Once it receives a message and "processes" it, the MDB sends an acknowledgment message to a temporary destination created by the sender for this purpose. The example is considered successful if the sender receives the acknowledgment message.

The MDB ejb-jar.xml descriptor

<ejb-jar>
  <enterprise-beans>
    <message-driven>
      <ejb-name>MDBExample</ejb-name>
      <ejb-class>org.jboss.example.jms.mdb.MDBExample</ejb-class>
      <transaction-type>Container</transaction-type>
    </message-driven>
  </enterprise-beans>
</ejb-jar>
    

The MDB jboss.xml descriptor

<enterprise-beans>
  <message-driven>
    <ejb-name>MDBExample</ejb-name>
    <destination-jndi-name>queue/@QUEUE_NAME@</destination-jndi-name>
  </message-driven>
</enterprise-beans>
    

public class MDBExample implements MessageDrivenBean, MessageListener
{
   private MessageDrivenContext ctx;

   private ConnectionFactory cf = null;

   public void onMessage(Message m)
   {
      Session session = null;
      Connection conn = null;

      try
      {
         TextMessage tm = (TextMessage)m;

         String text = tm.getText();
         System.out.println("message " + text + " received");
         String result = process(text);
         System.out.println("message processed, result: " + result);

         conn = getConnection();
         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

         Destination replyTo = m.getJMSReplyTo();
         MessageProducer producer = session.createProducer(replyTo);
         TextMessage reply = session.createTextMessage(result);

         producer.send(reply);
         producer.close();

      }
      catch(Exception e)
      {
         ctx.setRollbackOnly();
         e.printStackTrace();
         System.out.println("The Message Driven Bean failed!");
      }
      finally
      {
         if (conn != null)
         {
            try
            {
               closeConnection(conn);
            }
            catch(Exception e)
            {
               System.out.println("Could not close the connection!" +e);
            }
         }
      }
   }

   private String process(String s)
   {
      // flip the string

      String result = "";

      for (int i = 0; i < text.length(); i++)
      {
         result = text.charAt(i) + result;
      }
      return result;
   }

   public Connection getConnection() throws Exception
   {
      Connection connection = null;

      try
      {
         connection = cf.createConnection();
         connection.start();
      }
      catch(Exception e )
      {
         if(connection != null)
         {
            closeConnection(connection);
         }
         System.out.println("Failed to get connection...exception is " + e);
         throw e;
      }

      return connection;
   }

   public void closeConnection(Connection con) throws Exception
   {
      try
      {
         con.close();
      }
      catch(JMSException e)
      {
         System.out.println("Could not close connection " + con + " exception was " + e);
      }
   }

   public void ejbCreate()
   {
      try
      {
         InitialContext ic = new InitialContext();

         cf = (ConnectionFactory)ic.lookup("java:/JmsXA");

         ic.close();
      }
      catch(Exception e)
      {
         e.printStackTrace();
         throw new EJBException("Failure to get connection factory: " + e.getMessage());
      }
   }

   public void ejbRemove() throws EJBException
   {
      try
      {
         if(cf != null)
         {
            cf = null;
         }
      }
      catch(Exception e)
      {
         throw new EJBException("ejbRemove", e);
      }
   }

   public void setMessageDrivenContext(MessageDrivenContext ctx)
   {
      this.ctx = ctx;
   }


}

   

This example deploys a simple EJB3 Message Driven Bean that processes messages sent to a test queue. Once it receives a message and "processes" it, the MDB sends an acknowledgment message to a temporary destination created by the sender for this purpose. The example is considered successful if the sender receives the acknowledgment message.

This example relies on having access to a running JBoss Messaging instance. The JBoss Messaging instance must be installed and started according to the "Installation" chapter of this document. The example will automatically deploy its own queue, unless a queue with the same name is already deployed.

This example also relies on having access to the jboss-messaging-client.jar archive that comes with the release bundle. If you run this example from an unzipped installation bundle, the example run script is correctly configured to find the client jar. Otherwise, you must modify example's build.xml accordingly.

The example was designed to deploy its server-side artifacts under a JBoss' messaging configuration. If you intend to use the script with a JBoss configuration that is named differently, please modify the example's build.xml accordingly.

C:\work\src\cvs\jboss-head\jms\docs\examples\ejb3mdb\build.xml:60: EJB3 does not seem
to be installed in C:\work\src\jboss-4.0.3-src\build\output\jboss-4.0.3/server/messaging!
Install it and try again.
         

The EJB3 Message Driven Bean source code:

@MessageDriven(activationConfig =
{
      @ActivationConfigProperty(propertyName="destinationType", propertyValue="javax.jms.Queue"),
      @ActivationConfigProperty(propertyName="destination", propertyValue="queue/testQueue"),
      @ActivationConfigProperty(propertyName="DLQMaxResent", propertyValue="10")
})
public class EJB3MDBExample implements MessageListener
{
   public void onMessage(Message m)
   {
      businessLogic(m);
   }

   private void businessLogic(Message m)
   {
      Connection conn = null;
      Session session = null;

      try
      {
         TextMessage tm = (TextMessage)m;

         String text = tm.getText();
         System.out.println("message " + text + " received");

         // flip the string
         String result = "";
         for(int i = 0; i < text.length(); i++)
         {
            result = text.charAt(i) + result;
         }

         System.out.println("message processed, result: " + result);


         InitialContext ic = new InitialContext();
         ConnectionFactory cf = (ConnectionFactory)ic.lookup("java:/JmsXA");
         ic.close();

         conn = cf.createConnection();
         conn.start();
         session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

         Destination replyTo = m.getJMSReplyTo();
         MessageProducer producer = session.createProducer(replyTo);
         TextMessage reply = session.createTextMessage(result);

         producer.send(reply);
         producer.close();

      }
      catch(Exception e)
      {
         e.printStackTrace();
         System.out.println("The Message Driven Bean failed!");
      }
      finally
      {
         if (conn != null)
         {
            try
            {
               conn.close();
            }
            catch(Exception e)
            {
               System.out.println("Could not close the connection!" +e);
            }
         }
      }
   }
}
     

The basic test examples in this chapter serve as the sanity check for your JBoss Messaging installation. They also provide basic programming examples. To develop your own JMS services, you probably need to configure JBoss Messaging to setup your own destinations and connection factories etc. In Chapter 8, Configuration, we will discuss JBoss Messaging configuration files and options.

The Server Peer is the "heart" of the JBoss Messaging JMS facade. The server's configuration, resides in messaging-service.xml configuration file.

All JBoss Messaging services are rooted at the server peer

An example of a Server Peer configuration is presented below. Note that not all values for the server peer's attributes are specified in the example

 <mbean code="org.jboss.jms.server.ServerPeer"
        name="jboss.messaging:service=ServerPeer"
        xmbean-dd="xmdesc/ServerPeer-xmbean.xml">

   <constructor>
     <!-- ServerPeerID -->
     <arg type="int" value="0"/>
     <!-- DefaultQueueJNDIContext -->
     <arg type="java.lang.String" value="/queue"/>
     <!-- DefaultTopicJNDIContext -->
     <arg type="java.lang.String" value="/topic"/>
   </constructor>

   <attribute name="PostOffice">jboss.messaging:service=PostOffice</attribute>
   <attribute name="SecurityDomain">java:/jaas/messaging</attribute>
   <attribute name="DefaultSecurityConfig">
      <security>
        <role name="guest" read="true" write="true" create="true"/>
      </security>
   </attribute>
   <attribute name="DefaultDLQ">
       jboss.messaging.destination:service=Queue,name=DLQ</attribute>
   <attribute name="DefaultMaxDeliveryAttempts">10</attribute>
   <attribute name="DefaultExpiryQueue">
       jboss.messaging.destination:service=Queue,name=ExpiryQueue</attribute>
   <attribute name="DefaultRedeliveryDelay">0</attribute>
   <attribute name="QueueStatsSamplePeriod">5000</attribute>
   <attribute name="FailoverStartTimeout">60000</attribute>
   <attribute name="FailoverCompleteTimeout">300000</attribute>
   <attribute name="DefaultMessageCounterHistoryDayLimit">-1</attribute>

   <depends optional-attribute-name="PersistenceManager">
       jboss.messaging:service=PersistenceManager</depends>
   <depends optional-attribute-name="JMSUserManager">
       jboss.messaging:service=JMSUserManager</depends>
   <depends>jboss.messaging:service=Connector,transport=bisocket</depends>

 </mbean>
     

Several JBoss Messaging services interact with persistent storage. They include: The Persistence Manager, The PostOffice and the JMS User Manager. The Persistence Manager is used to handle the message-related persistence. The Post Office handles binding related persistence. The JMS User manager handles user related persistence The configuration for all these MBeans is handled in the xxx-persistence-service.xml file

If the database you want to switch to is one of MySQL, Oracle, PostgreSQL, MS SQL Sever or Sybase, persistence configuration files are already available in the examples/config directory of the release bundle.

In order to enable support for one of these databases, just replace the default hsqldb-persistence-service.xml configuration file with the database-specific configuration file and restart the server.

Also, be aware that by default, the Messaging services relying on a datastore are referencing "java:/DefaultDS" for the datasource. If you are deploying a datasource with a different JNDI name, you need to update all the DataSource attribute in the persistence configuration file. Example data source configurations for each of the popular databases are available in the distribution.

It is the job of the post office to route messages to their destination(s).

The post office maintains the mappings between addresses to which messages can be sent and their final queues.

For example when sending a message with an address that represents a JMS queue name, the post office will route this to a single queue - the JMS queue. When sending a message with an address that repesents a JMS topic name, the post office will route this to a set of queues - one for each JMS subscription.

The post office also handles the persistence for the mapping of addresses

JBoss Messaging comes with two different post office implementations - depending on whether you want clustering or not. The clustered post office has the ability to route messages to other nodes in the cluster

Whether you use the clustered post office or not depends on whether you deploy the clustered-xxx-persistence-service.xml MBean config or just the non clustered xxx-persistence-service.xml file.

It is likely that in future releases of JBoss Messaging the clustered and non clustered post offices will be combined into a single post office for ease of configuration

<mbean code="org.jboss.messaging.core.plugin.DefaultPostOfficeService"
      name="jboss.messaging:service=PostOffice"
      xmbean-dd="xmdesc/DefaultPostOffice-xmbean.xml">
      <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends>
      <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
      <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
      <attribute name="PostOfficeName">JMS</attribute>
      <attribute name="DataSource">java:/DefaultDS</attribute>
      <attribute name="CreateTablesOnStartup">true</attribute>
      <attribute name="SqlProperties"><![CDATA[
CREATE_POSTOFFICE_TABLE=CREATE TABLE JBM_POSTOFFICE (POSTOFFICE_NAME VARCHAR(255), NODE_ID INTEGER, QUEUE_NAME VARCHAR(1023), COND VARCHAR(1023), SELECTOR VARCHAR(1023), CHANNEL_ID BIGINT, CLUSTERED CHAR(1))
INSERT_BINDING=INSERT INTO JBM_POSTOFFICE (POSTOFFICE_NAME, NODE_ID, QUEUE_NAME, COND, SELECTOR, CHANNEL_ID, CLUSTERED) VALUES (?, ?, ?, ?, ?, ?, ?)
DELETE_BINDING=DELETE FROM JBM_POSTOFFICE WHERE POSTOFFICE_NAME=? AND NODE_ID=? AND QUEUE_NAME=?
LOAD_BINDINGS=SELECT NODE_ID, QUEUE_NAME, COND, SELECTOR, CHANNEL_ID, CLUSTERED FROM JBM_POSTOFFICE WHERE POSTOFFICE_NAME  = ?
      ]]></attribute>
   </mbean>

       

<mbean code="org.jboss.messaging.core.plugin.ClusteredPostOfficeService"
      name="jboss.messaging:service=PostOffice"
      xmbean-dd="xmdesc/ClusteredPostOffice-xmbean.xml">
      <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends>
      <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
      <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
      <attribute name="PostOfficeName">Clustered JMS</attribute>
      <attribute name="DataSource">java:/DefaultDS</attribute>
      <attribute name="CreateTablesOnStartup">true</attribute>
      <attribute name="SqlProperties"><![CDATA[
CREATE_POSTOFFICE_TABLE=CREATE TABLE JBM_POSTOFFICE (POSTOFFICE_NAME VARCHAR(255), NODE_ID INTEGER, QUEUE_NAME VARCHAR(1023), COND VARCHAR(1023), SELECTOR VARCHAR(1023), CHANNEL_ID BIGINT, CLUSTERED CHAR(1))
INSERT_BINDING=INSERT INTO JBM_POSTOFFICE (POSTOFFICE_NAME, NODE_ID, QUEUE_NAME, COND, SELECTOR, CHANNEL_ID, CLUSTERED) VALUES (?, ?, ?, ?, ?, ?, ?)
DELETE_BINDING=DELETE FROM JBM_POSTOFFICE WHERE POSTOFFICE_NAME=? AND NODE_ID=? AND QUEUE_NAME=?
LOAD_BINDINGS=SELECT NODE_ID, QUEUE_NAME, COND, SELECTOR, CHANNEL_ID, CLUSTERED FROM JBM_POSTOFFICE WHERE POSTOFFICE_NAME  = ?
      ]]></attribute>
      <attribute name="GroupName">DefaultPostOffice</attribute>
      <attribute name="StateTimeout">5000</attribute>
      <attribute name="CastTimeout">5000</attribute>
      <attribute name="StatsSendPeriod">10000</attribute>
      <attribute name="MessagePullPolicy">org.jboss.messaging.core.plugin.postoffice.cluster.NullMessagePullPolicy</attribute>
      <attribute name="ClusterRouterFactory">org.jboss.messaging.core.plugin.postoffice.cluster.DefaultRouterFactory</attribute>


      <attribute name="ChannelFactoryName">jgroups.mux:name=Multiplexer</attribute>
      <attribute name="SyncChannelName">udp-sync</attribute>
      <attribute name="AsyncChannelName">udp</attribute>
      <attribute name="ChannelPartitionName">${jboss.partition.name:DefaultPartition}-JMS</attribute>

      <attribute name="AsyncChannelConfig">
         <config>
            <UDP mcast_recv_buf_size="500000" down_thread="false" ip_mcast="true" mcast_send_buf_size="32000"
           mcast_port="45567" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
           mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="true" ucast_send_buf_size="32000" ip_ttl="32" bind_addr="127.0.0.1"/>
            <AUTOCONF down_thread="false" up_thread="false"/>
            <PING timeout="2000" down_thread="false" num_initial_members="3" up_thread="false"/>
            <MERGE2 max_interval="10000" down_thread="false" min_interval="5000" up_thread="false"/>
            <FD_SOCK down_thread="false" up_thread="false"/>
            <FD timeout="20000" max_tries="3" down_thread="false" up_thread="false" shun="true"/>
            <VERIFY_SUSPECT timeout="1500" down_thread="false" up_thread="false"/>
            <pbcast.NAKACK max_xmit_size="8192" down_thread="false" use_mcast_xmit="true" gc_lag="50" up_thread="false"
                         retransmit_timeout="100,200,600,1200,2400,4800"/>
            <UNICAST timeout="1200,2400,3600" down_thread="false" up_thread="false"/>
            <pbcast.STABLE stability_delay="1000" desired_avg_gossip="20000" down_thread="false" max_bytes="0" up_thread="false"/>
            <FRAG frag_size="8192" down_thread="false" up_thread="false"/>
            <VIEW_SYNC avg_send_interval="60000" down_thread="false" up_thread="false" />
            <pbcast.GMS print_local_addr="true" join_timeout="3000" down_thread="false" join_retry_timeout="2000" up_thread="false" shun="true"/>
         </config>
      </attribute>

      <attribute name="SyncChannelConfig">
         <config>
            <UDP mcast_recv_buf_size="500000" down_thread="false" ip_mcast="true" mcast_send_buf_size="32000"
           mcast_port="45568" ucast_recv_buf_size="500000" use_incoming_packet_handler="false"
           mcast_addr="228.8.8.8" use_outgoing_packet_handler="true" loopback="true" ucast_send_buf_size="32000" ip_ttl="32" bind_addr="127.0.0.1"/>
            <AUTOCONF down_thread="false" up_thread="false"/>
            <PING timeout="2000" down_thread="false" num_initial_members="3" up_thread="false"/>
            <MERGE2 max_interval="10000" down_thread="false" min_interval="5000" up_thread="false"/>
            <FD_SOCK down_thread="false" up_thread="false"/>
            <FD timeout="20000" max_tries="3" down_thread="false" up_thread="false" shun="true"/>
            <VERIFY_SUSPECT timeout="1500" down_thread="false" up_thread="false"/>
            <pbcast.NAKACK max_xmit_size="8192" down_thread="false" use_mcast_xmit="true" gc_lag="50" up_thread="false"
                         retransmit_timeout="100,200,600,1200,2400,4800"/>
            <UNICAST timeout="1200,2400,3600" down_thread="false" up_thread="false"/>
            <pbcast.STABLE stability_delay="1000" desired_avg_gossip="20000" down_thread="false" max_bytes="0" up_thread="false"/>
            <FRAG frag_size="8192" down_thread="false" up_thread="false"/>
            <VIEW_SYNC avg_send_interval="60000" down_thread="false" up_thread="false" />
            <pbcast.GMS print_local_addr="true" join_timeout="3000" down_thread="false" join_retry_timeout="2000" up_thread="false" shun="true"/>
            <pbcast.STATE_TRANSFER down_thread="false" up_thread="false"/>
         </config>
      </attribute>
   </mbean>

       

It is the job of the persistence manager to manage all message related persistence.

JBoss Messaging ships with a JDBC Persistence Manager used for handling persistence of message data in a relational database accessed via JDBC. The Persistence Manager implementation is pluggable (the Persistence Manager is a Messaging server plug-in), this making possible to provide other implementations for persisting message data in non relational stores, file stores etc.

The configuration of "persistent" services is grouped in a xxx-persistence-service.xml file, where the actual file prefix is usually inferred from its corresponding database JDBC connection string. By default, Messaging ships with a hsqldb-persistence-service.xml, which configures the Messaging server to use the in-VM Hypersonic database instance that comes by default with any JBossAS instance.

JBoss Messaging also ships with pre-made Persistence Manager configurations for MySQL, Oracle, PostgreSQL, Sybase and MS SQL Server. The example mysql-persistence-service.xml, oracle-persistence-service.xml, postgres-persistence-service.xml and sybase-persistence-service.xml and mssql-persistence-service.xml configuration files are available in the examples/config directory of the release bundle.

Users are encouraged to contribute their own configuration files where we will thoroughly test them before certifying them for suppported use with JBoss Messaging. The JDBC Persistence Manager has been designed to use standard SQL for the DML so writing a JDBC Persistence Manager configuration for another database is usually only a fairly simple matter of changing DDL in the configuration which is likely to be different for different databases.

The default Hypersonic persistence configuration file is listed below:

      <server>

         <mbean code="org.jboss.messaging.core.plugin.JDBCPersistenceManagerService"
            name="jboss.messaging:service=PersistenceManager"
            xmbean-dd="xmdesc/JDBCPersistenceManager-xmbean.xml">
            <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
            <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
            <attribute name="DataSource">java:/DefaultDS</attribute>
            <attribute name="CreateTablesOnStartup">true</attribute>
            <attribute name="UsingBatchUpdates">false</attribute>
            <attribute name="MaxParams">500</attribute>
         </mbean>

         <mbean code="org.jboss.messaging.core.plugin.DefaultPostOfficeService"
            name="jboss.messaging:service=PostOffice"
            xmbean-dd="xmdesc/DefaultPostOffice-xmbean.xml">
            <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends>
            <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
            <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
            <attribute name="PostOfficeName">JMS</attribute>
            <attribute name="DataSource">java:/DefaultDS</attribute>
            <attribute name="CreateTablesOnStartup">true</attribute>
         </mbean>

         <mbean code="org.jboss.jms.server.plugin.JDBCJMSUserManagerService"
            name="jboss.messaging:service=JMSUserManager"
            xmbean-dd="xmdesc/JMSUserManager-xmbean.xml">
            <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
            <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
            <attribute name="DataSource">java:/DefaultDS</attribute>
            <attribute name="CreateTablesOnStartup">true</attribute>
            <attribute name="SqlProperties"><![CDATA[
      POPULATE.TABLES.1=INSERT INTO JBM_USER (USER_ID,PASSWD,CLIENTID) VALUES ('dilbert','dogbert','dilbert-id')
            ]]></attribute>
         </mbean>

      </server>

   

An example of a Persistence Manager configuration for a MySQL database follows:

   <server>

      <mbean code="org.jboss.messaging.core.plugin.JDBCPersistenceManagerService"
            name="jboss.messaging:service=PersistenceManager"
            xmbean-dd="xmdesc/JDBCPersistenceManager-xmbean.xml">
            <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
            <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
            <attribute name="DataSource">java:/DefaultDS</attribute>
            <attribute name="CreateTablesOnStartup">true</attribute>
            <attribute name="UsingBatchUpdates">true</attribute>
            <attribute name="SqlProperties"><![CDATA[
      CREATE_MESSAGE_REFERENCE=CREATE TABLE JBM_MSG_REF (CHANNEL_ID BIGINT, MESSAGE_ID BIGINT, TRANSACTION_ID BIGINT, STATE CHAR(1), ORD BIGINT, PAGE_ORD BIGINT, DELIVERY_COUNT INTEGER, SCHED_DELIVERY BIGINT, PRIMARY KEY(CHANNEL_ID, MESSAGE_ID))
      CREATE_IDX_MESSAGE_REF_TX=CREATE INDEX JBM_MSG_REF_TX ON JBM_MSG_REF (TRANSACTION_ID)
      CREATE_IDX_MESSAGE_REF_ORD=CREATE INDEX JBM_MSG_REF_ORD ON JBM_MSG_REF (ORD)
      CREATE_IDX_MESSAGE_REF_PAGE_ORD=CREATE INDEX JBM_MSG_REF_PAGE_ORD ON JBM_MSG_REF (PAGE_ORD)
      CREATE_IDX_MESSAGE_REF_MESSAGE_ID=CREATE INDEX JBM_MSG_REF_MESSAGE_ID ON JBM_MSG_REF (MESSAGE_ID)
      CREATE_IDX_MESSAGE_REF_SCHED_DELIVERY=CREATE INDEX JBM_MSG_REF_SCHED_DELIVERY ON JBM_MSG_REF (SCHED_DELIVERY)
      CREATE_MESSAGE=CREATE TABLE JBM_MSG (MESSAGE_ID BIGINT, RELIABLE CHAR(1), EXPIRATION BIGINT, TIMESTAMP BIGINT, PRIORITY TINYINT, HEADERS MEDIUMBLOB, PAYLOAD LONGBLOB, CHANNEL_COUNT INTEGER, TYPE TINYINT, PRIMARY KEY (MESSAGE_ID))
      CREATE_TRANSACTION=CREATE TABLE JBM_TX (TRANSACTION_ID BIGINT, BRANCH_QUAL VARBINARY(254), FORMAT_ID INTEGER, GLOBAL_TXID VARBINARY(254), PRIMARY KEY (TRANSACTION_ID))
      CREATE_COUNTER=CREATE TABLE JBM_COUNTER (NAME VARCHAR(255), NEXT_ID BIGINT, PRIMARY KEY(NAME))
      INSERT_MESSAGE_REF=INSERT INTO JBM_MSG_REF (CHANNEL_ID, MESSAGE_ID, TRANSACTION_ID, STATE, ORD, PAGE_ORD, DELIVERY_COUNT, SCHED_DELIVERY) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
      DELETE_MESSAGE_REF=DELETE FROM JBM_MSG_REF WHERE MESSAGE_ID=? AND CHANNEL_ID=? AND STATE='C'
      UPDATE_MESSAGE_REF=UPDATE JBM_MSG_REF SET TRANSACTION_ID=?, STATE='-' WHERE MESSAGE_ID=? AND CHANNEL_ID=? AND STATE='C'
      UPDATE_PAGE_ORDER=UPDATE JBM_MSG_REF SET PAGE_ORD = ? WHERE MESSAGE_ID=? AND CHANNEL_ID=?
      COMMIT_MESSAGE_REF1=UPDATE JBM_MSG_REF SET STATE='C', TRANSACTION_ID = NULL WHERE TRANSACTION_ID=? AND STATE='+'
      COMMIT_MESSAGE_REF2=DELETE FROM JBM_MSG_REF WHERE TRANSACTION_ID=? AND STATE='-'
      ROLLBACK_MESSAGE_REF1=DELETE FROM JBM_MSG_REF WHERE TRANSACTION_ID=? AND STATE='+'
      ROLLBACK_MESSAGE_REF2=UPDATE JBM_MSG_REF SET STATE='C', TRANSACTION_ID = NULL WHERE TRANSACTION_ID=? AND STATE='-'
      LOAD_PAGED_REFS=SELECT MESSAGE_ID, DELIVERY_COUNT, PAGE_ORD, SCHED_DELIVERY FROM JBM_MSG_REF WHERE CHANNEL_ID = ? AND PAGE_ORD BETWEEN ? AND ? ORDER BY PAGE_ORD
      LOAD_UNPAGED_REFS=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'C' AND CHANNEL_ID = ? AND PAGE_ORD IS NULL ORDER BY ORD
      LOAD_REFS=SELECT MESSAGE_ID, DELIVERY_COUNT, SCHED_DELIVERY FROM JBM_MSG_REF WHERE STATE = 'C' AND CHANNEL_ID = ? ORDER BY ORD
      UPDATE_REFS_NOT_PAGED=UPDATE JBM_MSG_REF SET PAGE_ORD = NULL WHERE PAGE_ORD BETWEEN ? AND ? AND CHANNEL_ID=?
      SELECT_MIN_MAX_PAGE_ORD=SELECT MIN(PAGE_ORD), MAX(PAGE_ORD) FROM JBM_MSG_REF WHERE CHANNEL_ID = ?
      SELECT_EXISTS_REF=SELECT MESSAGE_ID FROM JBM_MSG_REF WHERE CHANNEL_ID = ? AND MESSAGE_ID = ?
      SELECT_EXISTS_REF_MESSAGE_ID=SELECT MESSAGE_ID FROM JBM_MSG_REF WHERE MESSAGE_ID = ?
      UPDATE_DELIVERY_COUNT=UPDATE JBM_MSG_REF SET DELIVERY_COUNT = ? WHERE CHANNEL_ID = ? AND MESSAGE_ID = ?
      UPDATE_CHANNEL_ID=UPDATE JBM_MSG_REF SET CHANNEL_ID = ? WHERE CHANNEL_ID = ?
      LOAD_MESSAGES=SELECT MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, TYPE FROM JBM_MSG
      INSERT_MESSAGE=INSERT INTO JBM_MSG (MESSAGE_ID, RELIABLE, EXPIRATION, TIMESTAMP, PRIORITY, HEADERS, PAYLOAD, CHANNEL_COUNT, TYPE) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
      INC_CHANNEL_COUNT=UPDATE JBM_MSG SET CHANNEL_COUNT = CHANNEL_COUNT + 1 WHERE MESSAGE_ID=?
      DEC_CHANNEL_COUNT=UPDATE JBM_MSG SET CHANNEL_COUNT = CHANNEL_COUNT - 1 WHERE MESSAGE_ID=?
      DELETE_MESSAGE=DELETE FROM JBM_MSG WHERE MESSAGE_ID=? AND CHANNEL_COUNT=0
      MESSAGE_ID_COLUMN=MESSAGE_ID
      MESSAGE_EXISTS=SELECT MESSAGE_ID FROM JBM_MSG WHERE MESSAGE_ID = ? FOR UPDATE
      INSERT_TRANSACTION=INSERT INTO JBM_TX (TRANSACTION_ID, BRANCH_QUAL, FORMAT_ID, GLOBAL_TXID) VALUES(?, ?, ?, ?)
      DELETE_TRANSACTION=DELETE FROM JBM_TX WHERE TRANSACTION_ID = ?
      SELECT_PREPARED_TRANSACTIONS=SELECT TRANSACTION_ID, BRANCH_QUAL, FORMAT_ID, GLOBAL_TXID FROM JBM_TX
      SELECT_MESSAGE_ID_FOR_REF=SELECT MESSAGE_ID, CHANNEL_ID FROM JBM_MSG_REF WHERE TRANSACTION_ID = ? AND STATE = '+' ORDER BY ORD
      SELECT_MESSAGE_ID_FOR_ACK=SELECT MESSAGE_ID, CHANNEL_ID FROM JBM_MSG_REF WHERE TRANSACTION_ID = ? AND STATE = '-' ORDER BY ORD
      UPDATE_COUNTER=UPDATE JBM_COUNTER SET NEXT_ID = ? WHERE NAME=?
      SELECT_COUNTER=SELECT NEXT_ID FROM JBM_COUNTER WHERE NAME=? FOR UPDATE
      INSERT_COUNTER=INSERT INTO JBM_COUNTER (NAME, NEXT_ID) VALUES (?, ?)
      SELECT_ALL_CHANNELS=SELECT DISTINCT(CHANNEL_ID) FROM JBM_MSG_REF
            ]]></attribute>
            <attribute name="MaxParams">500</attribute>
      </mbean>

      <mbean code="org.jboss.messaging.core.plugin.DefaultPostOfficeService"
         name="jboss.messaging:service=PostOffice"
         xmbean-dd="xmdesc/DefaultPostOffice-xmbean.xml">
         <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends>
         <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
         <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
         <attribute name="PostOfficeName">JMS</attribute>
         <attribute name="DataSource">java:/DefaultDS</attribute>
         <attribute name="CreateTablesOnStartup">true</attribute>
         <attribute name="SqlProperties"><![CDATA[
   CREATE_POSTOFFICE_TABLE=CREATE TABLE JBM_POSTOFFICE (POSTOFFICE_NAME VARCHAR(255), NODE_ID INTEGER, QUEUE_NAME VARCHAR(1023), COND VARCHAR(1023), SELECTOR VARCHAR(1023), CHANNEL_ID BIGINT, CLUSTERED CHAR(1))
   INSERT_BINDING=INSERT INTO JBM_POSTOFFICE (POSTOFFICE_NAME, NODE_ID, QUEUE_NAME, COND, SELECTOR, CHANNEL_ID, CLUSTERED) VALUES (?, ?, ?, ?, ?, ?, ?)
   DELETE_BINDING=DELETE FROM JBM_POSTOFFICE WHERE POSTOFFICE_NAME=? AND NODE_ID=? AND QUEUE_NAME=?
   LOAD_BINDINGS=SELECT NODE_ID, QUEUE_NAME, COND, SELECTOR, CHANNEL_ID, CLUSTERED FROM JBM_POSTOFFICE WHERE POSTOFFICE_NAME  = ?
         ]]></attribute>
      </mbean>

      <mbean code="org.jboss.jms.server.plugin.JDBCJMSUserManagerService"
         name="jboss.messaging:service=JMSUserManager"
         xmbean-dd="xmdesc/JMSUserManager-xmbean.xml">
         <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
         <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
         <attribute name="DataSource">java:/DefaultDS</attribute>
         <attribute name="CreateTablesOnStartup">true</attribute>
         <attribute name="SqlProperties"><![CDATA[
   CREATE_USER_TABLE=CREATE TABLE JBM_USER (USER_ID VARCHAR(32) NOT NULL, PASSWD VARCHAR(32) NOT NULL, CLIENTID VARCHAR(128), PRIMARY KEY(USER_ID))
   CREATE_ROLE_TABLE=CREATE TABLE JBM_ROLE (ROLE_ID VARCHAR(32) NOT NULL, USER_ID VARCHAR(32) NOT NULL, PRIMARY KEY(USER_ID, ROLE_ID))
   SELECT_PRECONF_CLIENTID=SELECT CLIENTID FROM JBM_USER WHERE USER_ID=?
   POPULATE.TABLES.1=INSERT INTO JBM_USER (USER_ID,PASSWD,CLIENTID) VALUES ('dilbert','dogbert','dilbert-id')
         ]]></attribute>
      </mbean>

   </server>

  

The JMS user manager handles the mapping of pre-configured client IDs to users and also managers the user and role tables which may or may not be used depending on which login module you have configured

Here is an example JMSUserManager configuration

   <mbean code="org.jboss.jms.server.plugin.JDBCJMSUserManagerService"
      name="jboss.messaging:service=JMSUserManager"
      xmbean-dd="xmdesc/JMSUserManager-xmbean.xml">
      <depends>jboss.jca:service=DataSourceBinding,name=DefaultDS</depends>
      <depends optional-attribute-name="TransactionManager">jboss:service=TransactionManager</depends>
      <attribute name="DataSource">java:/DefaultDS</attribute>
      <attribute name="CreateTablesOnStartup">true</attribute>
      <attribute name="SqlProperties"><![CDATA[
CREATE_USER_TABLE=CREATE TABLE JBM_USER (USER_ID VARCHAR(32) NOT NULL, PASSWD VARCHAR(32) NOT NULL, CLIENTID VARCHAR(128), PRIMARY KEY(USER_ID))
CREATE_ROLE_TABLE=CREATE TABLE JBM_ROLE (ROLE_ID VARCHAR(32) NOT NULL, USER_ID VARCHAR(32) NOT NULL, PRIMARY KEY(USER_ID, ROLE_ID))
SELECT_PRECONF_CLIENTID=SELECT CLIENTID FROM JBM_USER WHERE USER_ID=?
POPULATE.TABLES.1=INSERT INTO JBM_USER (USER_ID,PASSWD,CLIENTID) VALUES ('dilbert','dogbert','dilbert-id')
      ]]></attribute>
   </mbean>

      

With the default configuration JBoss Messaging binds just one connection factory in JNDI at start-up. This connection factory has no client ID and is bound into the following JNDI contexts: /ConnectionFactory, /XAConnectionFactory, java:/ConnectionFactory, java:/XAConnectionFactory

You may want to configure additional connection factories, for instance if you want to provide a default client id for a connection factory, or if you want to bind it in different places in JNDI, or if you want different connection factories to use different transports. Deploying a new connection factory is equivalent with adding a new ConnectionFactory MBean configuration to connection-factories-service.xml.

It is also possible to create an entirely new service deployment descriptor xxx-service.xml altogether and deploy it in $JBOSS_HOME/server/messaging/deploy.

Connection factories can either be clustered or non clustered. Clustered connection factories create subsequent connections on different nodes of the cluster according to the load balancing policy. The default load balancing policy is round robin.

An example connection factory configuration is presented below:

<server>
   <loader-repository>
      jboss.messaging:loader=ScopedLoaderRepository
         <loader-repository-config>
            java2ParentDelegation=false
         </loader-repository-config>
   </loader-repository>
   <mbean
      code="org.jboss.jms.server.connectionfactory.ConnectionFactory"
      name="jboss.messaging.destination:service=ConnectionFactory"
      xmbean-dd="xmdesc/ConnectionFactory-xmbean.xml">
      <constructor>
         <arg type="java.lang.String" value="myClientID"/>
      </constructor>
      <depends optional-attribute-name="ServerPeer">
         jboss.messaging:service=ServerPeer
      </depends>
      <depends optional-attribute-name="Connector">
         jboss.messaging:service=Connector,transport=socket
      </depends>
      <attribute name="PrefetchSize">10</attribute>
      <attribute name="DefaultTempQueueFullSize">1000</attribute>
      <attribute name="DefaultTempQueuePageSize">50</attribute>
      <attribute name="DefaultTempQueueDownCacheSize">50</attribute>
      <attribute name="JNDIBindings">
         <bindings>
            <binding>/MyConnectionFactory1</binding>
            <binding>/factories/cf1</binding>>
         </bindings>
      </attribute>
   </mbean>
</server>

    

The above example would create a connection factory with pre-configured client ID myClientID and bind the connection factory in two places in the JNDI tree: /MyConnectionFactory and /factories/cf. The connection factory will use the default remoting connector. To use a different remoting connector with the connection factory change the Connector attribute to specify the service name of the connector you wish to use.

JBoss Messaging uses JBoss Remoting for all client to server communication. For full details of what JBoss Remoting is capable of and how it is configured please consult the JBoss Remoting documentation.

The default configuration includes a single remoting connector which is used by the single default connection factory. Each connection factory can be configured to use its own connector.

The default connector is configured to use the remoting bisocket transport. The bisocket transport is a TCP socket based transport which only listens and accepts connections on the server side. I.e. connections are always initiated from the client side. This means it works well in typical firewall scenarios where only inbound connections are allowed on the server. Or where onlu outbound connections are allowed from the client.

The bisocket transport can be configured to use SSL where a higher level of security is required.

The other supported transport is the HTTP transport. This uses the HTTP protocol to communicate between client and server. Data is received on the client by the client periodically polling the server for messages. This transport is well suited to situations where there is a firewall between client and server which only allows incoming HTTP traffic on the server. Please note this transport will not be as performant as the bisocket transport due to the nature of polling and the HTTP protocl. Also please note it is not designed for high load situations.

No other remoting transports are currently supported by JBoss Messaging

You can look at remoting configuration under:

<JBoss>/server/<YourMessagingServer>/deploy/jboss-messaging.sar/remoting-service.xml

By default JBoss Messaging binds to ${jboss.bind.address} which can be defined by: ./run.sh -c <yourconfig> -b yourIP.

You can change remoting-service.xml if you want for example use a different communication port.

If you are using the JBoss AS ServiceBindingManager to provide different servers with different port ranges, then you must make sure that the JBoss Messaging remoting configuration specified in the JBoss Messaging section of the ServiceBindingManager xml file exactly matches that in remoting-service.xml

JBoss Messaging uses a callback mechanism from Remoting that needs a Socket for callback operations. These socket properties are passed to the server by a remote call when the connection is being estabilished. As we said before we will support bidirectional protocols in future releases.

By default JBoss Messaging will execute InetAddress.getLocalHost().getHostAddress() to access your local host IP, but in case you need to setup a different IP, you can define a system property in your java arguments:

Use java -Djboss.messaging.callback.bind.address=YourHost - That will determine the callBack host in your client.

The client port will be selected randomly for any non used port. But if you defined -Djboss.messaging.callback.bind.port=NumericPort in your System Properties that number is going to be used for the call back client port.

When a message is sent to a queue on particular node of the cluster, the system must decide whether the current node will handle it or whether it should send it to another node to handle.

The same applies if there are shared durable subscriptions on a topic and the message is being sent to a topic

The correct decision to make depends on your application topology.

If your application consists of a set of servers with the same MDBs deployed on each server, and you have many well distributed producers sending messages on all the nodes of the cluster, then the best policy is to always favour the local queue, since extra network trips are more expensive and the other nodes are also havng local messages sent to them

However if your application consists of a set of homogenous MDBs but you have few or badly distributed producers, then always favouring the local producer will mean the other nodes are being starved of messages and not using their CPUs cycles efficiently for messaging processing.

In this case, a good policy is to use a round robin routing policy where messages are round robin distributed to different nodes as they arrive.

In general, use the DefaultRoutingPolicy (this always favours the local queue) if you have many well distributed producers, and use the RoundRobinRoutingPolicy if you have few or badly distributed producers.

This are specified in the clustered post office config, by specifying the following attribute

To favour the local queue:

       <attribute name="ClusterRouterFactory">org.jboss.messaging.core.plugin.postoffice.cluster.DefaultRouterFactory</attribute>
      

To round robin:

         <attribute name="ClusterRouterFactory">org.jboss.messaging.core.plugin.postoffice.cluster.RoundRobinRouterFactory</attribute>
      

Once messages have arrived on queues in a cluster, in an ideal world they will all be consumed at the same rate, and there will be consumers on each node.

However, in some application topologies, consumes may close on queues on a node, leaving messages otherwise stranded, or perhaps consumers on some nodes are fast than consumers on other nodes causing messages to build up on one node or another and adversely effecting latency.

JBoss Messaging allows messages to pulled from one node to another as load dictates in order to cope with such situations

Whether or not you should activate message pulling (message redistribution) depends on your application topology

For an application that consists of a set of servers with a heterogenous bank of MDBs (or other consumers) deployed on each node, consuming at approximately the same rate, then message redistribution is not necessary.

However, if your application consists of different numbers or rates of consumers on different nodes then message redistribution may help

In general, use message redistribution when your consumers are not well distributed across the cluster or if they have greatly varying rates.

Message redistribution is set by setting the following attribute in the clustered post office configuration:

For no message redistribution:

           <attribute name="MessagePullPolicy">org.jboss.messaging.core.plugin.postoffice.cluster.NullMessagePullPolicy</attribute>
      

For message redistribution:

           <attribute name="MessagePullPolicy">org.jboss.messaging.core.plugin.postoffice.cluster.DefaultMessagePullPolicy</attribute>
      

When selecting message redistribution you should also choose a value of StatsSendPeriod appropriately.

To run performance tests side-by-side on the same machine, we assume that you create two JBoss AS configurations with the JBoss Messaging and JBossMQ modules respectively. We assume that the JBoss Messaging module is installed in the server/messaging directory (see Chapter 5, JBoss Messaging Non-Clustered Installation), and the default JBossMQ module is installed in server/jbossmq directory (just copy the original default directory that comes with the server).

Now, if you run the two configurations on the same server, there will be port conflicts. To avoid that, we use the JBoss ServiceBindingManager to increase the port numbers in the jbossmq configuration by 100 (i.e., the JNDI service will be available at port 1199 instead of 1099). To do that, un-comment the following line in server/jbossmq/conf/jboss-service.xml

<mbean code="org.jboss.services.binding.ServiceBindingManager"
       name="jboss.system:service=ServiceBindingManager">

  <attribute name="ServerName">ports-01</attribute>WWWr
  <attribute name="StoreURL">
    ../docs/examples/binding-manager/sample-bindings.xml
  </attribute>
  <attribute name="StoreFactoryClassName">
    org.jboss.services.binding.XMLServicesStoreFactory
  </attribute>
</mbean>
    

Now, you can start the messaging and jbossmq configurations side-by-side for testing.

run -c messaging
run -c jbossmq
    

The performance framework relies on distributed executors to send messages into the providers being tested. The executors can run standalone in their own VM and act as "remote" JMS connections, or colocated, in which case they are deployed as JBoss services and simulate "colocated" JMS connections.

In order to correctly deploy the colocated executors, the framework relies on the JBOSS_HOME environment variable. It assumes directories $JBOSS_HOME/server/messaging and $JBOSS_HOME/server/jbossmq exist.

ant sar
ant start-executors
    

Next, we need to deploy test message destinations. They are in the messaging-destinations-service.xml, jbossmq-destinations-service.xml files. Feel free to add your own destinations (must add equivalent ones in both files) and then deploy them via the following command.

ant deploy-destinations
    

The etc/perf.xml file is used to configure tests. In our setting (i.e., jbossmq runs in +100 port range from default), the <providers> section should look like the following. We can easily run the two JMS server configurations on different machines or in other port ranges. You just need to change the host and port numbers here for tests.

<provider name="JBossMessaging">
  <factory>org.jnp.interfaces.NamingContextFactory</factory>
  <url>jnp://localhost:1099</url>
  <pkg>org.jboss.naming:org.jnp.interfaces</pkg>
  <executor name="REMOTE" url="rmi://localhost:7777/standalone"/>
  <executor name="REMOTE2" url="rmi://localhost:7777/standalone2"/>
  <executor name="COLOCATED" url="rmi://localhost:7777/local-messaging"/>
  <executor name="COLOCATED2" url="rmi://localhost:7777/local-messaging2"/>
</provider>

<provider name="JBossMQ">
  <factory>org.jnp.interfaces.NamingContextFactory</factory>
  <url>jnp://localhost:1199</url>
  <pkg>org.jboss.naming:org.jnp.interfaces</pkg>
  <executor name="REMOTE" url="rmi://localhost:7777/standalone"/>
  <executor name="REMOTE2" url="rmi://localhost:7777/standalone2"/>
  <executor name="COLOCATED" url="rmi://localhost:7777/local-jbossmq"/>
  <executor name="COLOCATED2" url="rmi://localhost:7777/local-jbossmq2"/>
</provider>
    

The performance configuration section configures how to ramp up the load from 200 messages / sec to 3000 message / sec. We will gather statistics on the number of processed messages versus the number of sent messages.

<performance-test name="Throughput O KB Message
   Non-Persistent Non-Transactional, 1 sender, 1 receiver">

  <message-size>0</message-size>
  <messages>10000</messages>

  <drain/>

  <parallel>
    <send rate="200" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="400" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="800" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="1000" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="1500" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="2000" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="2500" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <parallel>
    <send rate="3000" executor="COLOCATED"/>
    <receive executor="COLOCATED2"/>
  </parallel>

  <execution provider="JBossMessaging"/>
  <execution provider="JBossMQ"/>

</performance-test>
    

To run the tests, simply execute ant run from the command line. You can access the benchmark result graphs from output/results/benchmark-results.html.

After running the test, you can clean up the executors and test destinations using the following commands.

ant stop-executors
ant undeploy-destinations