Posts Tagged ‘AMQP’

Apache Qpid 0.5 is released!

Wednesday, May 27th, 2009

Apache Qpid 0.5 is now released! Download it here:

http://qpid.apache.org/download.html

Here is the text of the announcement:

http://mail-archives.apache.org/mod_mbox/qpid-users/200905.mbox/browser

The Apache Qpid community is pleased to announce the release of Apache Qpid 0.5

Apache Qpid (http://qpid.apache.org) is a cross platform enterprise messaging solution which implements the Advanced Message Queueing Protocol (http://www.amqp.org). It provides brokers written in C++ and Java and clients in C++, Java (including a JMS implementation), .Net, Python, and Ruby.

New features included in this release are:

C++ Broker

  • [QPID-1567] - Queue replication (asynchronous) between two sites
  • [QPID-1669] - Client connection management in the qpid-cluster CLI utility
  • [QPID-1673] - Dynamic Library Build on Windows (DLL)

Java Broker

  • [QPID-1583] - IP White/Black lists for virtual hosts
  • [QPID-1648] - Enable live reconfiguration of Log4J settings for
    the Java broker via JMX
  • [QPID-1699] - Reload security section in configuration files through JMX

C++ Client

  • [QPID-1673] - Dynamic Library Build on Windows (DLL)

Java Client

Ruby Client

Java Management : JMX Console

  • [QPID-1500] - Mac OS X Build
  • [QPID-1648] - Enable live reconfiguration of Log4J settings for the Java broker via JMX
  • [QPID-1691] - Linux x86-64 and Solaris builds

Java Management : QMan

It is available to download from:

http://qpid.apache.org/download.html

Complete release notes are available here:

https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12310520&styleName=Html&version=12313597

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");