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()[0]
    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