Divert Example

ActiveMQ diverts allow messages to be transparently "diverted" from one address to another with just some simple configuration defined on the server side.

Diverts can be defined to be exclusive or non-exclusive.

With an exclusive divert the message is intercepted and does not get sent to the queues originally bound to that address - it only gets diverted.

With a non-exclusive divert the message continues to go to the queues bound to the address, but also a copy of the message gets sent to the address specified in the divert. Consequently non-exclusive diverts can be used to "snoop" on another address

Diverts can also be configured to have an optional filter. If specified then only matching messages will be diverted.

Diverts can be configured to apply a Transformer. If specified, all diverted messages will have the opportunity of being transformed by the Transformer.

Diverts are a very sophisticated concept, which when combined with bridges can be used to create interesting and complex routings. The set of diverts can be thought of as a type of routing table for messages.

Example step-by-step

In this example we will imagine a fictitious company which has two offices; one in London and another in New York.

The company accepts orders for it's products only at it's London office, and also generates price-updates for it's products, also only from it's London office. However only the New York office is interested in receiving price updates for New York products. Any prices for New York products need to be forwarded to the New York office.

There is an unreliable WAN linking the London and New York offices.

The company also requires a copy of any order received to be available to be inspected by management.

In order to achieve this, we will create a queue orderQueue on the London server in to which orders arrive.

We will create a topic, spyTopic on the London server, and there will be two subscribers both in London.

We will create a non-exclusive divert on the London server which will siphon off a copy of each order received to the topic spyTopic.

Here's the xml config for that divert, from activemq-configuration.xml

        
     <divert name="order-divert">                 
         <address>jms.queue.orders</address>
         <forwarding-address>jms.topic.spyTopic</forwarding-address>         
         <exclusive>false</exclusive>
      </divert>

For the prices we will create a topic on the London server, priceUpdates to which all price updates are sent. We will create another topic on the New York server newYorkPriceUpdates to which all New York price updates need to be forwarded.

Diverts can only be used to divert messages from one local address to another local address so we cannot divert directly to an address on another server.

Instead we divert to a local store and forward queue they we define in the configuration. This is just a normal queue that we use for storing messages before forwarding to another node.

Here's the configuration for it:

        
     <queues>     
        <queue name="jms.queue.priceForwarding">
           <address>jms.queue.priceForwarding</address>
        </queue>
     </queues>

Here's the configuration for the divert:

        
     <divert name="prices-divert">                  
	     <address>jms.topic.priceUpdates</address>
	     <forwarding-address>jms.queue.priceForwarding</forwarding-address>    
	     <filter string="office='New York'"/>
	     <transformer-class-name>org.apache.activemq.jms.example.AddForwardingTimeTransformer</transformer-class-name>
	     <exclusive>true</exclusive>
	  </divert>

Note we specify a filter in the divert, so only New York prices get diverted. We also specify a Transformer class since we are going to insert a header in the message at divert time, recording the time the diversion happened.

And finally we define a bridge that moves messages from the local queue to the address on the New York server. Bridges move messages from queues to remote addresses and are ideal to use when the target server may be stopped and started independently, and/or the network might be unreliable. Bridges guarantee once and only once delivery of messages from their source queues to their target addresses.

Here is the bridge configuration: 

	     
	  <bridges>
	     <bridge name="price-forward-bridge">
	        <queue-name>jms.queue.priceForwarding</queue-name>
	        <forwarding-address>jms.topic.newYorkPriceUpdates</forwarding-address>         
	        <reconnect-attempts>-1</reconnect-attempts>
          <static-connectors>
             <connector-ref>newyork-connector</connector-ref>
          </static-connectors>
	     </bridge>
      </bridges>

To run the example, simply type mvn verify from this directory

  1. Create an initial context to perform the JNDI lookup on the London server
    2.            
     initialContext0 = getContext(0);
  2. Look-up the queue orderQueue on the London server - this is the queue any orders are sent to
    3.            
     Queue queue = (Queue) initialContext.lookup("/queue/exampleQueue");
  3. Look-up the topic priceUpdates on the London server- this is the topic that any price updates are sent to
    4.            
     Topic priceUpdates = (Topic)initialContextLondon.lookup("/topic/priceUpdates");
  4. Look-up the spy topic on the London server- this is what we will use to snoop on any orders
    5.            
     Topic spyTopic = (Topic)initialContextLondon.lookup("/topic/spyTopic");
  5. Create an initial context to perform the JNDI lookup on the New York server.
    6.            
     initialContextNewYork = getContext(1);
  6. Look-up the topic newYorkPriceUpdates on the New York server - any price updates sent to priceUpdates on the London server will be diverted to the queue priceForward on the London server, and a bridge will consume from that queue and forward them to the address newYorkPriceUpdates on the New York server where they will be distributed to the topic subscribers on the New York server. 
    7.           
     Topic newYorkPriceUpdates = (Topic)initialContextNewYork.lookup("/topic/newYorkPriceUpdates");
  7. Perform a lookup on the Connection Factory on the London server
    8.            
    ConnectionFactory cfLondon = (ConnectionFactory)initialContextLondon.lookup("/ConnectionFactory");
  8. Perform a lookup on the Connection Factory on the New York server.
    9.            
    ConnectionFactory cfNewYork = (ConnectionFactory)initialContextNewYork.lookup("/ConnectionFactory");
  9. Create a JMS Connection on the London server
    10.            
    connectionLondon = cfLondon.createConnection();
  10. Create a JMS Connection on the New York server
    11.            
    connectionNewYork = cfNewYork.createConnection();
  11. Create a JMS Session on the London server.
    12.            
    Session sessionLondon = connectionLondon.createSession(false, Session.AUTO_ACKNOWLEDGE);
  12. Create a JMS Session on the New York server.
    13.            
    Session sessionNewYork = connectionNewYork.createSession(false, Session.AUTO_ACKNOWLEDGE);
  13. Create a JMS MessageProducer orderProducer that sends to the queue orderQueue on the London server.
    14.            
    MessageProducer orderProducer = sessionLondon.createProducer(orderQueue);
           /code>
  14. Create a JMS MessageProducer priceProducer that sends to the topic priceUpdates on the London server.
    15.            
    MessageProducer priceProducer = sessionLondon.createProducer(priceUpdates);
           /code>
  15. Create a JMS subscriber which subscribes to the spyTopic on the London server
    16.            
    MessageConsumer spySubscriberA = sessionLondon.createConsumer(spyTopic);
  16. Create another JMS subscriber which also subscribes to the spyTopic on the London server
    17.            
    MessageConsumer spySubscriberB = sessionLondon.createConsumer(spyTopic);
  17. Create a JMS MessageConsumer which consumes orders from the order queue on the London server
    18.            
    MessageConsumer orderConsumer = sessionLondon.createConsumer(orderQueue);
  18. Create a JMS subscriber which subscribes to the priceUpdates topic on the London server
    19.            
    MessageConsumer priceUpdatesSubscriberLondon = sessionLondon.createConsumer(priceUpdates);
  19. Create a JMS subscriber which subscribes to the newYorkPriceUpdates topic on the New York server
    20.            
    MessageConsumer newYorkPriceUpdatesSubscriberA = sessionNewYork.createConsumer(newYorkPriceUpdates);
  20. Create another JMS subscriber which also subscribes to the newYorkPriceUpdates topic on the New York server
    21.            
    MessageConsumer newYorkPriceUpdatesSubscriberB = sessionNewYork.createConsumer(newYorkPriceUpdates);
  21. Start the connections
    22.            
    connectionLondon.start();

    connectionNewYork.start();
  22. Create an order message
    23.            
    TextMessage orderMessage = sessionLondon.createTextMessage("This is an order");
  23. Send the order message to the order queue on the London server
    24.            
    orderProducer.send(orderMessage);

    System.out.println("Sent message: " + orderMessage.getText());
  24. The order message is consumed by the orderConsumer on the London server
    25.            
    TextMessage receivedOrder = (TextMessage)orderConsumer.receive(5000);

    System.out.println("Received order: " + receivedOrder.getText());
  25. A copy of the order is also received by the spyTopic subscribers on the London server
    26.            
    TextMessage spiedOrder1 = (TextMessage)spySubscriberA.receive(5000);

    System.out.println("Snooped on order: " + spiedOrder1.getText());

    TextMessage spiedOrder2 = (TextMessage)spySubscriberB.receive(5000);

    System.out.println("Snooped on order: " + spiedOrder2.getText());
  26. Create and send a price update message, destined for London
    27.            
    TextMessage priceUpdateMessageLondon = sessionLondon.createTextMessage("This is a price update for London");
                 
    priceUpdateMessageLondon.setStringProperty("office", "London");
         
    priceProducer.send(priceUpdateMessageLondon);
  27. The price update *should* be received by the local subscriber since we only divert messages where office = New York
    28.            
    TextMessage receivedUpdate = (TextMessage)priceUpdatesSubscriberLondon.receive(2000);

    System.out.println("Received price update locally: " + receivedUpdate.getText());
  28. The price update *should not* be received in New York
    29.            
    TextMessage priceUpdate1 = (TextMessage)newYorkPriceUpdatesSubscriberA.receive(1000);

    if (priceUpdate1 != null)
    {
       return false;
    }
   
    System.out.println("Did not received price update in New York, look it's: " + priceUpdate1);
   
    TextMessage priceUpdate2 = (TextMessage)newYorkPriceUpdatesSubscriberB.receive(1000);

    if (priceUpdate2 != null)
    {
       return false;
    }
   
    System.out.println("Did not received price update in New York, look it's: " + priceUpdate2);
  29. Create a price update message, destined for New York
    30.            
    TextMessage priceUpdateMessageNewYork = sessionLondon.createTextMessage("This is a price update for New York");
         
    priceUpdateMessageNewYork.setStringProperty("office", "New York");
  30. Send the price update message to the priceUpdates topic on the London server
    31.            
   priceProducer.send(priceUpdateMessageNewYork);
  31. The price update *should not* be received by the local subscriber to the priceUpdates topic since it has been *exclusively* diverted to the priceForward queue, because it has a header saying it is destined for the New York office
    32.            
   Message message = priceUpdatesSubscriberLondon.receive(1000);

   if (message != null)
   {
      return false;
   }

   System.out.println("Didn't receive local price update, look, it's: " + message);
  32. The remote subscribers on server 1 *should* receive a copy of the price update since it has been diverted to a local priceForward queue which has a bridge consuming from it and which forwards it to the same address on server 1. We notice how the forwarded messages have had a special header added by our custom transformer that we told the divert to use
    33.            
   priceUpdate1 = (TextMessage)newYorkPriceUpdatesSubscriberA.receive(5000);

   System.out.println("Received forwarded price update on server 1: " + priceUpdate1.getText());
   System.out.println("Time of forward: " + priceUpdate1.getLongProperty("time_of_forward"));

   priceUpdate2 = (TextMessage)newYorkPriceUpdatesSubscriberB.receive(5000);

   System.out.println("Received forwarded price update on server 2: " + priceUpdate2.getText());
   System.out.println("Time of forward: " + priceUpdate2.getLongProperty("time_of_forward"));
  33. And finally, always remember to close your resources after use, in a finally block.
    34.            
   finally
   {
      if (initialContextLondon != null)
      {
         initialContextLondon.close();
      }
      if (initialContextNewYork != null)
      {
         initialContextNewYork.close();
      }
      if (connectionLondon != null)
      {
         connectionLondon.close();
      }
      if (connectionNewYork != null)
      {
         connectionNewYork.close();
      }
   }