Chapter 2. Using the Qpid Messaging API

Table of Contents

2.1. A Simple Messaging Program in C++
2.2. A Simple Messaging Program in Python
2.3. A Simple Messaging Program in .NET C#
2.4. Addresses
2.4.1. Address Strings
2.4.2. Subjects
2.4.3. Address String Options
2.4.4. Address String Grammar
2.5. Sender Capacity and Replay
2.6. Receiver Capacity (Prefetch)
2.7. Acknowledging Received Messages
2.8. Receiving Messages from Multiple Sources
2.9. Transactions
2.10. Connection Options
2.11. Maps and Lists in Message Content
2.11.1. Qpid Maps and Lists in Python
2.11.2. Qpid Maps and Lists in C++
2.11.3. Qpid Maps and Lists in .NET
2.12. The Request / Response Pattern
2.13. Performance Tips
2.14. Cluster Failover
2.15. Logging
2.15.1. Logging in C++
2.15.2. Logging in Python
2.16. The AMQP 0-10 mapping
2.16.1. 0-10 Message Property Keys
2.17. Using Message Groups
2.17.1. Creating Message Group Queues
2.17.2. Sending Grouped Messages
2.17.3. Receiving Grouped Messages

The Qpid Messaging API is quite simple, consisting of only a handful of core classes.

The following sections show how to use these classes in a simple messaging program.

2.1. A Simple Messaging Program in C++

The following C++ program shows how to create a connection, create a session, send messages using a sender, and receive messages using a receiver.

Example 2.1. "Hello world!" in C++

	#include <qpid/messaging/Connection.h>
	#include <qpid/messaging/Message.h>
	#include <qpid/messaging/Receiver.h>
	#include <qpid/messaging/Sender.h>
	#include <qpid/messaging/Session.h>

	#include <iostream>

	using namespace qpid::messaging;

	int main(int argc, char** argv) {
	std::string broker = argc > 1 ? argv[1] : "localhost:5672";
	std::string address = argc > 2 ? argv[2] : "amq.topic";
	std::string connectionOptions = argc > 3 ? argv[3] : "";

	Connection connection(broker, connectionOptions);
	try {
        connection.open();  (1)
        Session session = connection.createSession(); (2)

        Receiver receiver = session.createReceiver(address); (3)
        Sender sender = session.createSender(address); (4)

        sender.send(Message("Hello world!"));

        Message message = receiver.fetch(Duration::SECOND * 1); (5)
        std::cout << message.getContent() << std::endl;
        session.acknowledge(); (6)

        connection.close(); (7)
        return 0;
	} catch(const std::exception& error) {
        std::cerr << error.what() << std::endl;
        connection.close();
        return 1;
	}
	}

(1)

Establishes the connection with the messaging broker.

(2)

Creates a session object on which messages will be sent and received.

(3)

Creates a receiver that receives messages from the given address.

(4)

Creates a sender that sends to the given address.

(5)

Receives the next message. The duration is optional, if omitted, will wait indefinitely for the next message.

(6)

Acknowledges receipt of all fetched messages on the session. This informs the broker that the messages were transferred and processed by the client successfully.

(7)

Closes the connection, all sessions managed by the connection, and all senders and receivers managed by each session.