Wednesday, September 5, 2012

Using XQuery for AMQP XML Subscriptions

I was trying to find some examples about how to use xquery to setup an x-binding for subscribing to specific XML messages. I didn’t find too much and the documentation for the Qpid Client libraries are scarce. So here's what I did.

Setting up the XML Exchange
First off, I didn't have the XML Exchange setup on my server. I'm running the Qpid C++ broker on CentOS 6.3. To setup the XML exchange I needed to install the XML exchange package.
# yum install qpid-cpp-server-xml.x86_64
Then modify the /etc/qpidd.conf file and add the following so that the Qpid Broker loads the modules at startup.
module-dir /usr/lib64/qpid/daemon
Restart the Qpid daemon.
 # service qpidd restart
Now you're ready to send some messages.

Filtering Messages w/ XQuery
Creating x-bindings provides greater control on filtering messages using XQuery. Your queries can filter on both XML content or messages properties contained in the body of the message.

Before I get to a more advanced example I'll explain how the XML Exchange works. For example, let's say we have the following message.
 
<Employee> 
  <EmpId>9289</EmpId> 
  <FirstName>John</FirstName> 
  <LastName>Doe</LastName> 
</Employee>

To query on an employee's last name you can create the following XQuery.
let $m := ./Employee
return $m/LastName = 'Doe'
The query can be combined with the address string and provided to the Receiver for a subscription.

Here's a C++ example of querying off multiple last names.

std::string query = "let $m := ./Employee "
        "return (";
for (size_t i = 0; i < names.size(); ++i)
{
    if (i == 0)
        query += "(";
    else
        query += " or ";
    query += "matches($m/LastName, '" + names[i] + "')";
}
query += "))";
std::string xmladdr;
xmladdr = "xml; {"
        " create: always, "
        " node: { type: topic, x-declare: {type: xml } }, "
        " link: { "
        "  x-bindings: [{ exchange: xml, key: test, arguments: { xquery:\""
        + query +
        "\"} }]"
        " } "
        "}";
qpid::messaging::Receiver r = conn_p.session.createReceiver(xmladdr);

No comments:

Post a Comment