Archive for the ‘Enterprise Messaging’ Category

QpidComponents.org

Tuesday, July 14th, 2009

We have launched a site called QpidComponents.org for Apache Qpid components that can not be developed as part of Apache Qpid per se, generally because they use software components that use open source licenses other than the Apache license.

Currently, these components include persistence and management tools.

See http:://QpidComponents.org!

MRG Messaging: A Programmer’s Overview

Friday, June 26th, 2009

I am going to be giving a presentation on MRG Messaging for programmers at the Red Hat Summit in Chicago, Thursday, September 3, from 1:30 to 2:30 p.m.

You can see the schedule here:

http://www.redhat.com/promo/summit/2009/agenda/tracks/

Here’s the blurb for the talk:

MRG Messaging: A Programmer’s Overview
Jonathan Robie, Software Engineer, Red Hat

Red Hat Enterprise MRG is an open source messaging system based on Advanced Message Queuing Protocol (AMPQ), an open standard for enterprise messaging. Red Hat Enterprise MRG allows users to write applications for simple, high-speed messaging in multiple languages (Java JMS, C++, Python, Ruby) and it allows users to run the applications on multiple platforms (Linux, Unix, Macintosh, or Windows). Red Hat Enterprise MRG provides guaranteed delivery, speed, and security.

This presentation serves as a programmer’s overview of Red Hat Enterprise MRG and is designed to teach programmers what they need to know to get started writing messaging applications. This presentation will feature:

  • Overviews of Red Hat Enterprise MRG, Apache Qpid, C++, and Python Messaging APIs
  • Discussion of the AMQP messaging standard and Apache Qpid
  • API summary for C++ and Python
  • Description of how to use Red Hat Enterprise MRG with Java JMS
  • Demonstration of how to write applications based on common messaging paradigms (e.g., direct, fanout, publish-subscribe, request-response, and XML content-based routing)
  • Demonstration of how to write Java JMS programs that use Red Hat Enterprise MRG
  • Demonstration of how to implement persistence, messaging transactions, and failover for clients using a high-availability cluster
  • Demonstration of the tools that can be used to manage the messaging server, view of the configuration, and track the state of messages in the system

Tim Fox: AMQP “of strategic importance” for JBoss

Thursday, April 23rd, 2009

Tim Fox has announced that

JBoss Messaging 2.x will implement AMQP, as we consider this to be of strategic importance, along with JMS, and of course the full set of enterprise messaging functionality that you’d expect in any serious messaging broker.

And also that JBoss will be licensed under Apache 2.0:

JBoss Messaging will also be relicensing it’s code under the ASL 2.0 licence (moving away from LGPL) in order to be usable by the widest range of interested parties.

Using the XML Exchange

Thursday, March 26th, 2009

The XML Exchange allows routing decisions to be made using XQuery. It is used primarily for two things:

  • XML Content-based Routing - a queue can be bound to an exchange using an XQuery that identifies “interesting” messages. This is particularly useful when the message publisher does not know what information a message consumer may use to decide whether to route the message, since the publisher and consumer do not need to agree on a set of headers.
  • Queries on Headers - the XML Exchange can query headers using XQuery whether or not the message content is in XML, so this provides query capability analogous to Java JMS Selectors. And the same query may use data from both headers and message content.

In this post, I will focus on XML content-based routing. Suppose a message publisher sends weather report data to an XML Exchange using the routing key “weather”. It might use the National Weather Service’s XML format - here’s an example of a report in this format:

http://www.weather.gov/xml/current_obs/KRDU.xml

Here’s a simpler version that we will use in our sample program:

<weather>
    <station>Raleigh-Durham International Airport (KRDU)</station>
    <wind_speed_mph>16</wind_speed_mph>
    <temperature_f>70</temperature_f>
    <dewpoint>35</dewpoint>
</weather>

Now I want to write a message consumer that tells me when there is good weather for sailing. In this post I will show how to route messages to a queue - the complete program is one of the Apache Qpid examples, and is covered in Red Hat’s MRG Messaging Tutorial.

In my little sailboat, I don’t like wind above 20 mph, and it’s a lot more fun when there’s at least 7 mph wind. Rain isn’t a big issue, but it’s a little nicer when it’s not raining, so I’d like to be 5 degrees above dewpoint or so. And I don’t usually sail when it’s colder than 50 degrees, which makes me a bit of a wimp. Here is an XQuery that tests a message to see if it meets these criteria:

   let $w := ./weather
   return $w/station = 'Raleigh-Durham International Airport (KRDU)'
      and $w/temperature_f > 50
      and $w/temperature_f - $w/dewpoint > 5
      and $w/wind_speed_mph > 7
      and $w/wind_speed_mph < 20

Now I can take that query and bind it to the XML Exchange, asking that it be used to decide whether to route messages with the routing key “weather”. In Python, that looks like this:

session.queue_declare(queue="message_queue")

binding = {}
binding["xquery"] = """
   let $w := ./weather
   return $w/station = 'Raleigh-Durham International Airport (KRDU)'
      and $w/temperature_f > 50
      and $w/temperature_f - $w/dewpoint > 5
      and $w/wind_speed_mph > 7
      and $w/wind_speed_mph < 20 """

session.exchange_bind(exchange="xml", queue="message_queue", binding_key="weather", arguments=binding)

As you can see, the XML Exchange allows message consumers to use sophisticated queries to decide which messages are interesting, and the message producer does not need to do anything special to provide the data used for these routing decisions. And it’s very simple to use!

XML Exchange Patent

Wednesday, March 18th, 2009

I seem to have achieved my 15 minutes of fame as the Inventor in a patent application for the XML Exchange filed by Red Hat. Some people are upset to see Red Hat file a patent application related to open standards and open source software.

Red Hat works hard to make it easier to write innovative software despite the toxic software patent environment. We all would like to see this fixed (are you listening, President Obama?) In the meantime, defensive patents are a necessary evil.

Here’s an excerpt from a statement on this controversy from Rob Tiller, Vice President and Assistant General Counsel of Intellectual Property at Red Hat:

http://www.press.redhat.com/2009/03/17/discouraging-software-patent-lawsuits/:

Red Hat has worked hard to address the problems of our patent system. We believe there are serious problems with the existing system, particularly as it affects free and open source software. In just the last few months, we have supported new patent reform legislation, submitted a brief in the Bilski case arguing against patenting of software, and created an innovative patent settlement in the FireStar case that gave broad protection to the open source community. We are also proud of our work in helping establish the Open Invention Network, supporting the Peer-to-Patent program, and developing our Patent Promise.

We have also worked to build a patent portfolio. As we have explained many times, the purpose of this portfolio is defensive. This means that it is designed to discourage patent lawsuits by giving us the ability to retaliate against potential patent aggressors by asserting counterclaims as a defense. We believe it is important to have such a portfolio, because of the threats of companies that are hostile to FOSS and that have amassed large stockpiles of overbroad patents.

We have made a public commitment to this purely defensive approach with our Patent Promise, which is at http://www.redhat.com/legal/patent_policy.html. This promise is binding. It demonstrates our determination to use our patent portfolio only in the defense of free software.

In the next few days, I’ll write a post that describes the XML Exchange and shows how to use it in programs.

High Availability Messaging (Clustering)

Wednesday, March 11th, 2009

Apache Qpid has added support for High Availability, which is called “Clustering” in Qpid.

With this feature, any number of messaging servers can be run as one cluster. If one server goes down, a client that was using that server can “fail over” to another server in the cluster, and continue without any loss of messages. Servers can be added to or removed from a cluster while it is in use.

Failover is not transparent, but clients that support failover are easy to write (and transparent failover is planned in a future version). For details, look at the code for the Qpid failover example , or read Handing Failover in C++ Connections in the MRG Messaging Tutorial .

If you have logging enabled, and are using persistence, you can restore the messaging state by restarting a message server if the server you are using crashes, but you have to wait for the server to restart. Clustering allows you to instantly connect to another server and continue work. However, there is a cost: in order to allow each server in a cluster to continue the work of any other server, a cluster must replicate state for all brokers in the cluster. Therefore, using multiple servers in a cluster is somewhat slower than using a single server without clustering. This may be counter-intuitive for people who use clustering in the context of High Performance Computing or High Throughput Computing, where clustering increases performance or throughput. To minimize the overhead of clustering, the server uses the the Application Interface Specification (AIS) to communicate changes in state across the cluster.

For information on configuring OpenAIS and viewing the state with qpidtool, see the MRG Messaging User Guide chapter on clustering. For information on configuring clustering in MRG Messaging, see the clustering chapter in the MRG Messaging Installation Guide.

Submitting jobs with AMQP to a Condor based Grid

Monday, March 9th, 2009

Matt F. of Spinning Matt has written a tutorial on MRG Grid’s Low Latency scheduling, which allows Condor jobs to be submitted using AMQP messaging.

Nicely written, with code examples.

Looks like his first blog post. Hope we’ll see more.

Python Messaging API for Apache Qpid

Friday, March 6th, 2009

I posted a summary of the C++ Messaging API for Qpid yesterday, here’s a summary of the Python Messaging API. You can find the full Python Messaging API on the Apache Qpid web site.

I have not yet worked this summary into that documentation.

Includes and Namespaces

import qpid
import sys
import os
from qpid.util import connect
from qpid.connection import Connection
from qpid.datatypes import Message, RangedSet, uuid4
from qpid.queue import Empty

Opening and closing connections and sessions

host="127.0.0.1"
port=5672
user="guest"
password="guest"

socket = connect(host, port)
connection = Connection (sock=socket, username=user, password=password)
connection.start()
session = connection.session(str(uuid4()))
...
session.close(timeout=10)

Declaring and binding queues:

session.queue_declare(queue="message_queue")
session.exchange_bind(exchange="amq.direct", queue="message_queue", binding_key="routing_key")

Sending a message:

props = session.delivery_properties(routing_key="routing_key")
session.message_transfer(destination="amq.direct", message=Message(props,"Hi, Mom!")))

Replying to a message:

Message request, response

message_properties = request.get("message_properties")
reply_to = message_properties.reply_to
if reply_to == None:
   raise Exception("This message is missing the 'reply_to' property, which is required")   

props = session.delivery_properties(routing_key=reply_to["routing_key"])
session.message_transfer(destination=reply_to["exchange"], message=Message(props,request.body.upper()))

A message listener:

#----- Message Receive Handler -----------------------------
class Receiver:
  def __init__ (self):
    self.finalReceived = False

  def isFinal (self):
    return self.finalReceived

  def Handler (self, message):
    content = message.body
    session.message_accept(RangedSet(message.id))
    print content
    if content == "That's all, folks!":
      self.finalReceived = True

# Call message_subscribe() to tell the broker to deliver messages
# from the AMQP queue to this local client queue. The broker will
# start delivering messages as soon as message_subscribe() is called.

session.message_subscribe(queue="message_queue", destination=local_queue_name)
queue.start()

# Register a message listener with the queue

receiver = Receiver()
queue.listen (receiver.Handler)

while not receiver.isFinal() :
  sleep (1)

Getting and setting message contents

Getting Content

content = ""	# Content of the last message read

message = None
while content != final:
	message = queue.get(timeout=10)
	content = message.body
        session.message_accept(RangedSet(message.id))
	print content

Setting Content

# Set content in the message constructor
props = session.delivery_properties(routing_key="routing_key")
message = Message(props,"Hi, Mom!")

Getting and Setting Delivery Properties

Look up the delivery properties in the Python Messaging API Reference under qpid.session.Session.delivery_properties.

delivery_properties = session.delivery_properties()
delivery_properties.routing_key="routing_key"
delivery_properties.delivery_mode="persistent"
delivery_properties.ttl=100
delivery_properties.priority=9

message=Message(delivery_properties, "Hi, Mom!")

Getting and Setting Message Properties

Look up the message properties in the Python Messaging API Reference
under qpid.session.Session.message_properties.

# Setting reply_to

Message message
message_properties = message.get("message_properties")
message_properties.reply_to = session.reply_to("amq.direct", reply_to)
message=Message(props, "Hi, Mom!")

# Getting reply_to

Message message
message_properties = request.get("message_properties")
reply_to = message_properties.reply_to
props = session.delivery_properties(routing_key=reply_to["routing_key"]) 

# Setting Mime type and Encoding
message_properties = request.get("message_properties")
message_properties.content_type = "text/plain"
message_properties.encoding = "UTF-8"

Getting and Setting Application Headers

message_properties = message.get("message_properties")
message_properties.application_headers["control"] = "continue"

C++ Messaging API for Apache Qpid

Thursday, March 5th, 2009

Here’s a summary I wrote for the C++ Messaging API reference. You can find the complete API reference at the Apache Qpid site or the Red Hat MRG Messaging site:

Includes and Namespaces

#include <qpid/client/Connection.h>
#include <qpid/client/Session.h>
#include <qpid/client/Message.h>

#include <qpid/client/SubscriptionManager.h<>

using namespace qpid::client;
using namespace qpid::framing;

Opening and closing connections and sessions

Connection connection;
try {
    connection.open(host, port);
    Session session =  connection.newSession();
    ...
    connection.close();
    return 0;
} catch(const std::exception& error) {
    std::cout << error.what() << std::endl;
}
return 1;

Declaring and binding queues:

session.queueDeclare(arg::queue="message_queue");
session.exchangeBind(arg::exchange="amq.direct", arg::queue="message_queue", arg::bindingKey="routing_key");

Sending a message:

message.getDeliveryProperties().setRoutingKey("routing_key");
message.setData("Hi, Mom!");
session.messageTransfer(arg::content=message,  arg::destination="amq.direct");

Sending a message (asynchronous):

#include <qpid/client/AsyncSession.h>
async(session).messageTransfer(arg::content=message,  arg::destination="amq.direct");
...
session.sync();

Replying to a message:

Message request, response;
...
if (request.getMessageProperties().hasReplyTo()) {
   string routingKey = request.getMessageProperties().getReplyTo().getRoutingKey();
   string exchange = request.getMessageProperties().getReplyTo().getExchange();
   response.getDeliveryProperties().setRoutingKey(routingKey);
   messageTransfer(arg::content=response, arg::destination=exchange);
}

A message listener:

class Listener : public MessageListener{
  private:
    SubscriptionManager& subscriptions;
  public:
    Listener(SubscriptionManager& subscriptions);
    virtual void received(Message& message);
};
void Listener::received(Message& message) {
    std::cout << "Message: " << message.getData() << std::endl;
    if (endCondition(message)) {
       subscriptions.cancel(message.getDestination());
    }
}

Using a message listener with a subscription manager:

SubscriptionManager subscriptions(session);
Listener listener(subscriptions);
subscriptions.subscribe(listener, "message_queue");
subscriptions.run();

Using a LocalQueue with a subscription manager:

SubscriptionManager subscriptions(session);
LocalQueue local_queue;
subscriptions.subscribe(local_queue, string("message_queue"));
Message message;
for (int i=0; i<10; i++) {
    local_queue.get(message, 10000);
    std::cout << message.getData() << std::endl;
}

Getting and setting message contents

getData()

std::cout << "Response: " << message.getData() << std::endl;

setData()

message.setData("That's all, folks!");

appendData()

message.appendData(" ... let's add a bit more ...");

Getting and Setting Delivery Properties

getDeliveryProperties()

message.getDeliveryProperties().setRoutingKey("control");
message.getDeliveryProperties().setDeliveryMode(PERSISTENT);
message.getDeliveryProperties().setPriority(9);
message.getDeliveryProperties().setTtl(100);

hasDeliveryProperties()

if (! message.hasDeliveryProperties()) {
  ...
}

Getting and Setting Message Properties

getMessageProperties()

request.getMessageProperties().setReplyTo(ReplyTo("amq.direct", response_queue.str()));
routingKey = request.getMessageProperties().getReplyTo().getRoutingKey();
exchange = request.getMessageProperties().getReplyTo().getExchange();
message.getMessageProperties().setContentType("text/plain");
message.getMessageProperties().setContentEncoding("text/plain");

hasMessageProperties()

request.getMessageProperties().hasReplyTo();

Getting and Setting Application Headers

getHeaders()

message.getHeaders().getString("control");
message.getHeaders().setString("control","continue");

Apache Qpid is now a Top Level Project

Thursday, March 5th, 2009

Apache Qpid (http://qpid.apache.org) has graduated from an Apache Incubator to an Apache Top Level Project. Here’s the press release, it includes quotes from Carl Trieloff of Red Hat, John O’Hara of J.P. Morgan, and Sam Ramji of Microsoft:


The Apache Software Foundation Names Qpid a Top-Level Project

Here’s a description of Apache Qpid from http://qpid.apache.org:

Apache Qpid: Open Source AMQP Messaging

Enterprise Messaging systems let programs communicate by exchanging messages, much as people communicate by exchanging email. Unlike email, enterprise messaging systems provide guaranteed delivery, speed, security, and freedom from spam. Until recently, there was no open standard for Enterprise Messaging systems, so programmers either wrote their own, or used expensive proprietary systems.

AMQP Advanced Message Queuing Protocol is the first open standard for Enterprise Messaging. It is designed to support messaging for just about any distributed or business application. Routing can be configured flexibly, easily supporting common messaging paradigms like point-to-point, fanout, publish-subscribe, and request-response.

Apache Qpid implements the latest AMQP specification, providing transaction management, queuing, distribution, security, management, clustering, federation and heterogeneous multi-platform support and a lot more. And Apache Qpid is extremely fast. Apache Qpid aims to be 100% AMQP Compliant.