Advanced Message Queuing Protocol (AMQP)
Listing 2. Server-Side Python Program
#!/usr/bin/python import subprocess import os from qpid.util import connect, ssl from qpid.connection import Connection, sslwrap from qpid.datatypes import Message, RangedSet, uuid4 from qpid.queue import Empty from qpid.spec import load from qpid.queue import Empty from qpid.session import SessionException # processRequest: this is what actually does the work. def processRequest(requestMessage): print "Servicing Request" proc = subprocess.Popen('rpm -qa', shell=True, stdout=subprocess.PIPE, ) stdout_value = proc.communicate() myPid = os.getpid() ret_value = "From Server PID " \ + str(myPid) + ":\n" + stdout_value \ + "---------------------------------\n" return ret_value # First, load the correct specification file. locSpec = load('/usr/local/share/qpid/specs/amqp.0-10.xml') # Now, connect to the server. socket = connect("localhost", 5672) connection = Connection (sock=socket, spec=locSpec, username = "guest", password = "guest") connection.start() session = connection.session(str(uuid4())) # Declare the listening server queue and connect to server queue. # Create server queue if it does not exist. myPid = os.getpid() serverQueueName = "serverListenQueue" + str(myPid) localQueueName = "serverListenLocal_" + session.name session.queue_declare(queue=serverQueueName, exclusive=True) session.exchange_bind(exchange="amq.topic", queue=serverQueueName, binding_key="SERVER_STATUS") session.message_subscribe(queue=serverQueueName, destination=localQueueName) localQueue = session.incoming(localQueueName) localQueue.start() # Now, start an event loop. while True: try: requestObj = localQueue.get(timeout=60) session.message_accept(RangedSet(requestObj.id)) requestStr = requestObj.body requestProperties = requestObj.get("message_properties") replyTo = requestProperties.reply_to if replyTo == None: raise Exception("This message is missing " \ + "the 'reply_to'" property, " \ + "which is required") responseMessage = processRequest(requestStr) props = session.delivery_properties( routing_key=replyTo["routing_key"]) session.message_transfer(destination=replyTo["exchange"], message=Message(props, responseMessage)) except Empty: continue
In the event loop, the server first receives a request from the local queue. If there is no request within the timeout value (60 seconds), the get() method will raise an Empty exception. Because the server needs to serve requests continually, the program catches the Empty exception and simply continues. When a message arrives, the server runs the processRequest method and constructs data with the method's return values. The reply message takes exchange and routing key information from the original message's reply-to field and then is delivered to the exchange.
With AMQP, it is possible to construct a queuing system that allows a server farm to respond to multiple different kinds of requests. This example considers weather prediction models. Here, there are different server clusters, with one cluster serving each state. In such a case, it would be extremely handy to be able to send requests to each farm from an arbitrary location.
This example requires three processes. The first process (the client) delivers requests, and it is fundamentally the same as the client in the previous example. It is different only in that it loops over a list to deliver ten weather requests for Ohio and ten requests for Virginia. On the receiving end, there are two servers: one for Ohio and one for Virginia. Each server subscribes to the amq.topic exchange with the routing key #.ohio or #.virginia. Furthermore, each server has the ability to subscribe to existing server queues or create those that do not exist.
These routing keys contain wild cards. When the routing key contains a hash mark in place of text, the exchange will match any text where the hash mark resides. In this way, the weather predicting dæmons using #.ohio also would respond to requests for topic news.ohio and sports.ohio. Likewise, if a sports reporting dæmon had invaded the cluster and was listening for sports.#, the subscriptions for both the sports dæmon and the weather reporting dæmon for Ohio would match sports.ohio.
Listing 3 contains the client, and Listing 4 contains the server for Ohio. Create the server for Virginia by duplicating the server for Ohio and replacing all occurrences of Ohio with Virginia. (When you do so, make sure all routing keys have all lowercase characters.)