Posts Tagged ‘Qpid’

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

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