Using the AMQ C++ Client

AMQ C++ supports the following industry-recognized standards and network protocols:

AMQ C++ supports the following OS and language versions:

AMQ C++ is supported in combination with the following AMQ components and versions:

For more information, see Red Hat AMQ Supported Configurations.

This section introduces the core API entities and describes how they operate together.

AMQ C++ sends and receives messages. Messages are transferred between connected peers over senders and receivers. Senders and receivers are established over sessions. Sessions are established over connections. Connections are established between two uniquely identified containers. Though a connection can have multiple sessions, often this is not needed. The API allows you to ignore sessions unless you require them.

A sending peer creates a sender to send messages. The sender has a target that identifies a queue or topic at the remote peer. A receiving peer creates a receiver to receive messages. The receiver has a source that identifies a queue or topic at the remote peer.

The sending of a message is called a delivery. The message is the content sent, including all metadata such as headers and annotations. The delivery is the protocol exchange associated with the transfer of that content.

To indicate that a delivery is complete, either the sender or the receiver settles it. When the other side learns that it has been settled, it will no longer communicate about that delivery. The receiver can also indicate whether it accepts or rejects the message.

This document uses the following conventions for the sudo command and file paths.

The sudo command

In this document, sudo is used for any command that requires root privileges. You should always exercise caution when using sudo, as any changes can affect the entire system.

For more information about using sudo, see The sudo Command.

About the use of file paths in this document

In this document, all file paths are valid for Linux, UNIX, and similar operating systems (for example, /home/...). If you are using Microsoft Windows, you should use the equivalent Microsoft Windows paths (for example, C:\Users\...).

To begin installation, use your subscription to access AMQ distribution files and repositories.

AMQ C++ is distributed as a set of RPM packages for Red Hat Enterprise Linux. Follow these steps to install them.

In order to compile programs using the API, you will also need to install gcc-c++, cmake, and make.

sudo yum install gcc-c++ cmake make

$ sudo yum install gcc-c++ cmake make

AMQ C++ is distributed as an SDK zip archive for use with Visual Studio. Follow these steps to install it.

The example programs require a running broker with a queue named examples. Follow these steps to define the queue and start the broker:

This section illustrates how to compile the example programs that come with the client API.

The compiled example programs use the broker we started earlier to queue the messages between sending and receiving.

Sending messages

Receiving messages

This client program connects to a server using <connection-url>, creates a sender for target <address>, sends a message containing <message-body>, closes the connection, and exits.

#include <proton/connection.hpp>
#include <proton/container.hpp>
#include <proton/message.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/sender.hpp>
#include <proton/target.hpp>

#include <iostream>
#include <string>

struct send_handler : public proton::messaging_handler {
    std::string conn_url_ {};
    std::string address_ {};
    std::string message_body_ {};

    void on_container_start(proton::container& cont) override {
        cont.connect(conn_url_);
    }

    void on_connection_open(proton::connection& conn) override {
        conn.open_sender(address_);
    }

    void on_sender_open(proton::sender& snd) override {
        std::cout << "SEND: Opened sender for target address '"
                  << snd.target().address() << "'\n";
    }

    void on_sendable(proton::sender& snd) override {
        proton::message msg {message_body_};
        snd.send(msg);

        std::cout << "SEND: Sent message '" << msg.body() << "'\n";

        snd.close();
        snd.connection().close();
    }
};

int main(int argc, char** argv) {
    if (argc != 4) {
        std::cerr << "Usage: send <connection-url> <address> <message-body>\n";
        return 1;
    }

    send_handler handler {};
    handler.conn_url_ = argv[1];
    handler.address_ = argv[2];
    handler.message_body_ = argv[3];

    proton::container cont {handler};

    try {
        cont.run();
    } catch (const std::exception& e) {
        std::cerr << e.what() << "\n";
        return 1;
    }

    return 0;
}

#include <proton/connection.hpp>
#include <proton/container.hpp>
#include <proton/message.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/sender.hpp>
#include <proton/target.hpp>

#include <iostream>
#include <string>

struct send_handler : public proton::messaging_handler {
    std::string conn_url_ {};
    std::string address_ {};
    std::string message_body_ {};

    void on_container_start(proton::container& cont) override {
        cont.connect(conn_url_);
    }

    void on_connection_open(proton::connection& conn) override {
        conn.open_sender(address_);
    }

    void on_sender_open(proton::sender& snd) override {
        std::cout << "SEND: Opened sender for target address '"
                  << snd.target().address() << "'\n";
    }

    void on_sendable(proton::sender& snd) override {
        proton::message msg {message_body_};
        snd.send(msg);

        std::cout << "SEND: Sent message '" << msg.body() << "'\n";

        snd.close();
        snd.connection().close();
    }
};

int main(int argc, char** argv) {
    if (argc != 4) {
        std::cerr << "Usage: send <connection-url> <address> <message-body>\n";
        return 1;
    }

    send_handler handler {};
    handler.conn_url_ = argv[1];
    handler.address_ = argv[2];
    handler.message_body_ = argv[3];

    proton::container cont {handler};

    try {
        cont.run();
    } catch (const std::exception& e) {
        std::cerr << e.what() << "\n";
        return 1;
    }

    return 0;
}

Running the example

To run the example program, copy it to a local file, compile it, and execute it from the command line.

g++ send.cpp -o send -std=c++11 -lstdc++ -lqpid-proton-cpp
./send amqp://localhost queue1 hello

$ g++ send.cpp -o send -std=c++11 -lstdc++ -lqpid-proton-cpp
$ ./send amqp://localhost queue1 hello

This client program connects to a server using <connection-url>, creates a receiver for source <address>, and receives messages until it is terminated or it reaches <count> messages.

#include <proton/connection.hpp>
#include <proton/container.hpp>
#include <proton/delivery.hpp>
#include <proton/message.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/receiver.hpp>
#include <proton/source.hpp>

#include <iostream>
#include <string>

struct receive_handler : public proton::messaging_handler {
    std::string conn_url_ {};
    std::string address_ {};
    int desired_ {0};
    int received_ {0};

    void on_container_start(proton::container& cont) override {
        cont.connect(conn_url_);
    }

    void on_connection_open(proton::connection& conn) override {
        conn.open_receiver(address_);
    }

    void on_receiver_open(proton::receiver& rcv) override {
        std::cout << "RECEIVE: Opened receiver for source address '"
                  << rcv.source().address() << "'\n";
    }

    void on_message(proton::delivery& dlv, proton::message& msg) override {
        std::cout << "RECEIVE: Received message '" << msg.body() << "'\n";

        received_++;

        if (received_ == desired_) {
            dlv.receiver().close();
            dlv.connection().close();
        }
    }
};

int main(int argc, char** argv) {
    if (argc != 3 && argc != 4) {
        std::cerr << "Usage: receive <connection-url> <address> [<message-count>]\n";
        return 1;
    }

    receive_handler handler {};
    handler.conn_url_ = argv[1];
    handler.address_ = argv[2];

    if (argc == 4) {
        handler.desired_ = std::stoi(argv[3]);
    }

    proton::container cont {handler};

    try {
        cont.run();
    } catch (const std::exception& e) {
        std::cerr << e.what() << "\n";
        return 1;
    }

    return 0;
}

#include <proton/connection.hpp>
#include <proton/container.hpp>
#include <proton/delivery.hpp>
#include <proton/message.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/receiver.hpp>
#include <proton/source.hpp>

#include <iostream>
#include <string>

struct receive_handler : public proton::messaging_handler {
    std::string conn_url_ {};
    std::string address_ {};
    int desired_ {0};
    int received_ {0};

    void on_container_start(proton::container& cont) override {
        cont.connect(conn_url_);
    }

    void on_connection_open(proton::connection& conn) override {
        conn.open_receiver(address_);
    }

    void on_receiver_open(proton::receiver& rcv) override {
        std::cout << "RECEIVE: Opened receiver for source address '"
                  << rcv.source().address() << "'\n";
    }

    void on_message(proton::delivery& dlv, proton::message& msg) override {
        std::cout << "RECEIVE: Received message '" << msg.body() << "'\n";

        received_++;

        if (received_ == desired_) {
            dlv.receiver().close();
            dlv.connection().close();
        }
    }
};

int main(int argc, char** argv) {
    if (argc != 3 && argc != 4) {
        std::cerr << "Usage: receive <connection-url> <address> [<message-count>]\n";
        return 1;
    }

    receive_handler handler {};
    handler.conn_url_ = argv[1];
    handler.address_ = argv[2];

    if (argc == 4) {
        handler.desired_ = std::stoi(argv[3]);
    }

    proton::container cont {handler};

    try {
        cont.run();
    } catch (const std::exception& e) {
        std::cerr << e.what() << "\n";
        return 1;
    }

    return 0;
}

Running the example

To run the example program, copy it to a local file, compile it, and execute it from the command line.

g++ receive.cpp -o receive -std=c++11 -lstdc++ -lqpid-proton-cpp
./receive amqp://localhost queue1

$ g++ receive.cpp -o receive -std=c++11 -lstdc++ -lqpid-proton-cpp
$ ./receive amqp://localhost queue1

Errors in AMQ C++ can be handled in two different ways:

Catching exceptions

Catching exceptions is the most basic, but least granular, way to handle errors. If an error is not handled using an override in a handler routine, an exception will be thrown and can be caught and handled. An exception thrown in this way will be thrown by the container’s run method.

All of the exceptions that can be thrown by AMQ C++ are descended from proton::error, which in turn is a subclass of std::runtime_error (which is a subclass of std::exception).

The code example below illustrates how a block could be written to catch any exception thrown from AMQ C++.

try {
    // Something that might throw an exception
} catch (proton::error& e) {
    // Handle Proton-specific problems here
} catch (std::exception& e) {
    // Handle more general problems here
}

try {
    // Something that might throw an exception
} catch (proton::error& e) {
    // Handle Proton-specific problems here
} catch (std::exception& e) {
    // Handle more general problems here
}

If you require no API-specific exception handling, you only need to catch std::exception since proton::error descends from it.

int main() {
    try {
        // Something that might throw an exception
    } catch (std::exception& e) {
        std::cerr << "Caught exception: " << e.what() << std::endl;
    }
}

int main() {
    try {
        // Something that might throw an exception
    } catch (std::exception& e) {
        std::cerr << "Caught exception: " << e.what() << std::endl;
    }
}

Handling connection and protocol errors

You can handle protocol-level errors by overriding the following messaging_handler methods:

These event handling routines are called whenever there is an error condition with the specific object that is in the event. After calling the error handler, the appropriate close handler will also be called.

If not overridden the default error handler will be called with an indication of the error condition that occurred.

There is also a default error handler:

If one of the more specific error handlers is not overridden, this will be called.

AMQ C++ has the ability to execute code after a delay. You can use this to implement time-based behaviors in your application, such as periodically scheduled work or timeouts.

void on_sender_open(proton::sender& snd) override {
    proton::duration interval {5 * proton::duration::SECOND};
    snd.work_queue().schedule(interval, [=] { send(snd); });
}

void send(proton::sender snd) {
    if (snd.credit() > 0) {
        proton::message msg {"hello"};
        snd.send(msg);
    }
}

void on_sender_open(proton::sender& snd) override {
    proton::duration interval {5 * proton::duration::SECOND};
    snd.work_queue().schedule(interval, [=] { send(snd); });
}

void send(proton::sender snd) {
    if (snd.credit() > 0) {
        proton::message msg {"hello"};
        snd.send(msg);
    }
}

For more information, see the API reference.

If set, AMQ C++ uses the value of the MESSAGING_CONNECT_FILE environment variable to locate the configuration file.

If MESSAGING_CONNECT_FILE is not set, AMQ C++ searches for a file named connect.json at the following locations and in the order shown. It stops at the first match it encounters.

The connect.json file contains JSON data, with additional support for JavaScript comments.

Many of the options have default values, so a simple example need only provide a few details:

{
    "host": "example.com",
    "user": "alice",
    "password": "secret"
}

{
    "host": "example.com",
    "user": "alice",
    "password": "secret"
}

SASL and SSL/TLS options are nested under "sasl" and "tls" namespaces:

{
    "host": "example.com",
    "user": "ortega",
    "password": "secret",
    "sasl": {
        "mechanisms": ["SCRAM-SHA-1", "SCRAM-SHA-256"]
    },
    "tls": {
        "cert": "/home/ortega/cert.pem",
        "key": "/home/ortega/key.pem"
    }
}

{
    "host": "example.com",
    "user": "ortega",
    "password": "secret",
    "sasl": {
        "mechanisms": ["SCRAM-SHA-1", "SCRAM-SHA-256"]
    },
    "tls": {
        "cert": "/home/ortega/cert.pem",
        "key": "/home/ortega/key.pem"
    }
}

The option keys containing a dot (.) represent attributes nested inside a namespace.

The container object can handle multiple connections concurrently. As AMQP events occur on connections, the container calls messaging_handler callback functions. Callbacks for any one connection are serialized (not called concurrently), but callbacks for different connections can be safely executed in parallel.

You can assign a handler to a connection in container::connect() or listen_handler::on_accept() using the handler connection option. We recommend creating a separate handler for each connection. That way the handler does not need locks or other synchronization to protect it against concurrent use by library threads. If any non-library threads use the handler concurrently, then you will need synchronization.

The connection, session, sender, receiver, tracker, and delivery objects are not thread-safe and are subject to the following rules.

The message object is a value type with the same threading constraints as a standard C++ built-in type. It cannot be concurrently modified.

The work_queue interface provides a safe way to communicate between different connection handlers or between non-library threads and connection handlers.

When the library calls the work function, it will be serialized safely so that you can treat the work function like an event callback and safely access the handler and AMQ C++ objects stored on it.

The connection::wake() method allows any thread to prompt activity on a connection by triggering an on_connection_wake() callback. This is the only thread-safe method on connection.

wake() is a lightweight, low-level primitive for signaling between threads.

The semantics of wake() are similar to std::condition_variable::notify_one(). There will be a wakeup, but there must be some shared application state to determine why the wakeup occurred and what, if anything, to do about it.

Work queues are easier to use in many instances, but wake() may be useful if you already have your own external thread-safe queues and need an efficient way to wake a connection to check them for data.

Before C++11 there was no standard support for threading in C++. You can use AMQ C++ with threads but with the following limitations.

The container::schedule() and work_queue APIs accept C++11 lambda functions to define units of work. If you are using a version of C++ that does not support lambdas, you must use the make_work() function instead.

AMQP messages are composed using the AMQP type system. This common format is one of the reasons AMQP clients in different languages are able to interoperate with each other.

When sending messages, AMQ C++ automatically converts language-native types to AMQP-encoded data. When receiving messages, the reverse conversion takes place.

AMQP defines a standard mapping to the JMS messaging model. This section discusses the various aspects of that mapping. For more information, see the AMQ JMS Interoperability chapter.

JMS message types

AMQ C++ provides a single message type whose body type can vary. By contrast, the JMS API uses different message types to represent different kinds of data. The table below indicates how particular body types map to JMS message types.

For more explicit control of the resulting JMS message type, you can set the x-opt-jms-msg-type message annotation. See the AMQ JMS Interoperability chapter for more information.

AMQ Broker is designed to interoperate with AMQP 1.0 clients. Check the following to ensure the broker is configured for AMQP messaging:

AMQ Interconnect works with any AMQP 1.0 client. Check the following to ensure the components are configured correctly: