A Sinfonia on Messaging with txAMQP, Part IV

A Sinfonia on Messaging:

1. The Voice of Business
   
2. The Voice of Architecture
3. A RabbitMQ and txAMQP Interlude
4. The Voice of Code
   

Finally, we reach the foundation, the fundament, the bass line of our sinfonia . First we looked at the business need (however contrived), then an architecture that fit this need. We had a little play with txAMQP and RabbitMQ – we know how to connect to RabbitMQ, create exchanges and queues and then bind them together.

Now, with a basic understanding under our belts of messaging systems and some practical experience with Twisted AMQP consumers and producers, we can tackle unified orders. In this post we will learn how to extend our knowledge by creating some messaging pattern implementations for MacGredder's kilt store.


Architecture Redress

The outcome we expect for this part of the implementation is a single message representing a single order (with a single item per order, for simplicity) that could have come from any of the three order sources at the kilt store: web site, phone call, or sales guy. By adding a little complexity in creating a single message abstraction for the order, we will be able to simplify the system as a whole: the rest of the code we write will only have to know about one order message, not three.

After closer inspection of the architecture, I've made some changes to the "unified orders" EIP model. In particular, I've been more specific about the patterns which we will be using, not treating any as implicit. Here's the latest:

[diagram of new model]

Here's a list of the patterns we've identified for use in MaGrudder's kilt store:

• Messaging (53)
• Message Broker (322)
  
• Message (66)
• Message Endpoint (95)
  
• Channel Adapter (127)
  
• Message Channel (60)
• Normalizer (352)
  
• Message Translator (85)
• Point-to-Point Channel (103)

The numbers in parentheses correspond to the page number in Enterprise Integration Patterns, where you will find each pattern described. Some of these may seem rather obvious, but I don't want to take anything for granted. By keeping them in mind we don't loose sight of the big picture, and we get to appreciate the work that others have done for us!

For instance, some of these patterns we get for free with our software:

• Messaging - provided by RabbitMQ and txAMQP
• Message Broker - provided by RabbitMQ
• Message - specified by the AMQ protocol
• Message Channel - specified by the AMQ protocol
• Point-to-Point Channel - specified by the AMQ protocol

This leaves the following that we need to actually implement:

• Message Endpoint - connect an application to a messaging channel
  
• Channel Adapter - access an application's API/data, publish messages to a channel and receive messages
• Message Translator - translate one data format into another
• Normalizer - rewire our messaging to handle message translations better
  

For now, let us ignore RabbitMQ, message routing, and the rest of MacGrudder's infrastructure. All we're going to do is create message from three possible sources and ensure that a single format is used in the messaging system. We'll worry about the rest later :-)

Let's look at how orders are created:

• Web store - user hits the "place order" button
• Sales guy - creates an order in the CRM
• Phone call - someone in the shop writes down an order on a piece of paper
  

We're going to need to have web girl do the following:

• add a new component to the web store code that creates a message when an order is placed by a customer (Message Endpoint)
  
• sales guy's CRM offers an XML-RPC API, so web girl will add a layer of code combining an XML-RPC client and a messaging client (Channel Adatper)
  
• either build a custom form for internal use (custom orders) or add an admin mode to the web store so that employees can create orders on behalf of non-web customers (in both cases, Message Endpoint)
  

The Voice of Code

The web store component will be very straight-forward: we need to intercept the data during the order creation process, right after the data is successfully saved to the database.

Web Store and Message Endpoint

Let's say we have a super-simple web store like the o ne here. At the point where the data is saved to a database, we want to also publish a message on our messaging server. We can do that by updating the ProcessOrder.render method with a call to our endpoint:

[new render method]

Based on previous analysis, we know that we need to create a message and then send it. We create a new endpoint module with the following code:

[endpoint code]

As is expected with the Message Endpoint pattern, we had to modify our application code, but so far, the changes have been minimal. For applications where we don't have access to the source code, we need to create a channel adapter.

CRM and Channel Adapter

Fortunately, sales guy's CRM has an XML-RPC API. To connect the CRM to our messaging system, we need to do something similar to what we did to our web store:

• get new orders (with an XML-RPC client)
  
• create a new order message
• send the message

For testing purposes, I've supplied a fake CRM XML-RPC server. Since we've already written the code for sending messages, adding support for sending a message based on the data we get from the CRM's XML-RPC is simplicity itself:

[code for adapter]

With our adapter in place, we now notice that the data we get back fron the web server and the data provided by the CRM are actually not the same. These need to be correlated with a Message Translator.

Message Translator

For the message translator, I chose to put the code in a separate module and not integrate the changes directly into the two places where messages are currently created (the Message Adapter and the Message Endpoint implementations). For the purposes of this blog post, that decision keeps the code and concepts clear, however it has another excellent benefit: when we choose to split up the message queues (and we will!), the translator will integrate easily.

With these thoughts in mind, I chose to write the translator using the Twisted component architecture. You might want to take a quick look at that link to get a sense of it (especially the bit about creating the plug socket adapters and plugging in the hairdryer). As cumbersome as a few folks may find this, it really is a good match: we've got one type of message (the unified format one) and we want to adapt other message formats to the unified format one. I did a quick mock-up of a toy implementation (fully-running code) that expresses this clearly: view it here.

The implementation that we use for our orders isn't all that much different. Besides implementing actually useful getMessage methods, it provides a convenience function, translate, for doing the adaptation. To save space, I'll just show the two adapter classes:

[adapter classes]

This is the fatal piece of code that finally forces us to think more about how messaging should really work in this scenario. Do we follow old tendencies? Do we cobble something together and glue a little messaging in there? Or do we take full advantage of messages and queues. I bet you can guess which :-)

Normalizer

There's really no reason we can't continue with the messaging setup we used in the example from the last blog post. It would work. But let's look at how we would wire our messaging server such that we could keep our code loosely coupled.

If we do so and keep our translator code separate from the message-creation code, we're going to need to think of a way for the original messages to be transformed into translated messages. So, how about this:

• add a routing key for each type of source message (e.g., one for web orders, one for CRM orders, etc.)
• route each received message to the appropriate translator
• send unified format messages to a new queue
• point other parts of the application that need order messages at this new queue
  

What we haven't done in this code is create a queue for every Message Translator; all


Running the Code

If you would like to see all of this in action, here's what you need to do:

Download the example code. Ensure that you have Twisted, txAMQP, and simplejson installed. Open several terminal windows. In one of them, add the kilt store vhost to RabbitMQ and map the guest user to it:

$ rabbitmqctl add_vhost kilt-store
$ rabbitmqctl mapuservhost guest kilt-store

Start the fake CRM XML-RPC server in another terminal window:

$ twistd -noy ./bin/crm-xmlrpc.tac –pidfile=crm.pid

Start the fake web store in another terminal window:

$ twistd -noy ./bin/webstore.tac –pidfile=web.pid

In another terminal window, start the fake consumer that will listen on the unified message channel. All messages that have been normalised to the unified message format we've defined will be publised here:

$ python ./bin/fakeconsumer.py

In yet another window, start the CRM XML-RPC Message Adapter. This will start pulling orders off the fake CRM order queue:

$ python adapter.py


XXX With multiple processes connecting to our queue, we've got a new problem: the messages are consumed in a round-robin fashion, but what we actually want is for all the consumers to be able to read off the queue.

With a single message, we will want to continue with our business process: order validation and billing. But that will have to wait until the next post.


References

• Enterprise Integration Patterns
  
• The async Python library for AMQP
• The AMQP Spec
• RabbitMQ
• Twisted Components : Interfaces and Adapters