Using the AMQ C++ Client

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

AMQ C++ supports the OS and language versions listed below. For more information, see Red Hat AMQ 7 Supported Configurations.

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

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.

The sudo command

In this document, sudo is used for any command that requires root privileges. Exercise caution when using sudo because any changes can affect the entire system. For more information about sudo, see Using the sudo command.

File paths

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

Variable text

This document contains code blocks with variables that you must replace with values specific to your environment. Variable text is enclosed in arrow braces and styled as italic monospace. For example, in the following command, replace <project-dir> with the value for your environment:

$ cd <project-dir>

For more information about using packages, see Appendix B, Using Red Hat Enterprise Linux packages.

When you extract the contents of the .zip file, a directory named amq-clients-2.9.0-cpp-win is created. This is the top-level directory of the installation and is referred to as <install-dir> throughout this document.

The Hello World example creates a connection to the broker, sends a message containing a greeting to the examples queue, and receives it back. On success, it prints the received message to the console.

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

        // To connect with a user and password:
        //
        // proton::connection_options opts {};
        // opts.user("<user>");
        // opts.password("<password>");
        //
        // cont.connect(conn_url_, opts);
    }

    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. For more information, see Chapter 3, Getting started.

$ 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_);

        // To connect with a user and password:
        //
        // proton::connection_options opts {};
        // opts.user("<user>");
        // opts.password("<password>");
        //
        // cont.connect(conn_url_, opts);
    }

    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. For more information, see Chapter 3, Getting started.

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

AMQ C++ is an asynchronous event-driven API. To define how the application handles events, the user implements callback methods on the messaging_handler class. These methods are then called as network activity or timers trigger new events.

struct example_handler : public proton::messaging_handler {
    void on_container_start(proton::container& cont) override {
        std::cout << "The container has started\n";
    }

    void on_sendable(proton::sender& snd) override {
        std::cout << "A message can be sent\n";
    }

    void on_message(proton::delivery& dlv, proton::message& msg) override {
        std::cout << "A message is received\n";
    }
};

These are only a few common-case events. The full set is documented in the API reference.

The container is the top-level API object. It is the entry point for creating connections, and it is responsible for running the main event loop. It is often constructed with a global event handler.

int main() {
    example_handler handler {};
    proton::container cont {handler};
    cont.run();
}

Each container instance has a unique identity called the container ID. When AMQ C++ makes a connection, it sends the container ID to the remote peer. To set the container ID, pass it to the proton::container constructor.

proton::container cont {handler, "job-processor-3"};

If the user does not set the ID, the library will generate a UUID when the container is constucted.

Connection URLs encode the information used to establish new connections.

scheme://host[:port]

amqps://example.com
amqps://example.net:56720
amqp://127.0.0.1
amqp://[::1]:2000

To connect to a remote server, call the container::connect() method with a connection URL. This is typically done inside the messaging_handler::on_container_start() method.

class example_handler : public proton::messaging_handler {
    void on_container_start(proton::container& cont) override {
        cont.connect("amqp://example.com");
    }

    void on_connection_open(proton::connection& conn) override {
        std::cout << "The connection is open\n";
    }
};

For information about creating secure connections, see Chapter 7, Security.

Reconnect allows a client to recover from lost connections. It is used to ensure that the components in a distributed system reestablish communication after temporary network or component failures.

AMQ C++ disables reconnect by default. To enable it, set the reconnect connection option to an instance of the reconnect_options class.

proton::connection_options opts {};
proton::reconnect_options ropts {};

opts.reconnect(ropts);

container.connect("amqp://example.com", opts);

With reconnect enabled, if a connection is lost or a connection attempt fails, the client will try again after a brief delay. The delay increases exponentially for each new attempt.

To control the delays between connection attempts, set the delay, delay_multiplier, and max_delay options. All durations are specified in milliseconds.

To limit the number of reconnect attempts, set the max_attempts option. Setting it to 0 removes any limit.

proton::connection_options opts {};
proton::reconnect_options ropts {};

ropts.delay(proton::duration(10));
ropts.delay_multiplier(2.0);
ropts.max_delay(proton::duration::FOREVER);
ropts.max_attempts(0);

opts.reconnect(ropts);

container.connect("amqp://example.com", opts);

AMQ C++ allows you to configure multiple connection endpoints. If connecting to one fails, the client attempts to connect to the next in the list. If the list is exhausted, the process starts over.

To specify alternate connection endpoints, set the failover_urls reconnect option to a list of connection URLs.

std::vector<std::string> failover_urls = {
    "amqp://backup1.example.com",
    "amqp://backup2.example.com"
};

proton::connection_options opts {};
proton::reconnect_options ropts {};

opts.reconnect(ropts);
ropts.failover_urls(failover_urls);

container.connect("amqp://primary.example.com", opts);

AMQ C++ can accept inbound network connections, enabling you to build custom messaging servers.

To start listening for connections, use the proton::container::listen() method with a URL containing the local host address and port to listen on.

class example_handler : public proton::messaging_handler {
    void on_container_start(proton::container& cont) override {
        cont.listen("0.0.0.0");
    }

    void on_connection_open(proton::connection& conn) override {
        std::cout << "New incoming connection\n";
    }
};

The special IP address 0.0.0.0 listens on all available IPv4 interfaces. To listen on all IPv6 interfaces, use [::0].

For more information, see the server receive.cpp example.

AMQ C++ uses SSL/TLS to encrypt communication between clients and servers.

To connect to a remote server with SSL/TLS, set the ssl_client_options connection option and use a connection URL with the amqps scheme. The ssl_client_options constructor takes the filename, directory, or database ID of a CA certificate.

proton::ssl_client_options sopts {"/etc/pki/ca-trust"};
proton::connection_options opts {};

opts.ssl_client_options(sopts);

container.connect("amqps://example.com", opts);

AMQ C++ can authenticate connections with a user and password.

To specify the credentials used for authentication, set the user and password options on the connect method.

proton::connection_options opts {};

opts.user("alice");
opts.password("secret");

container.connect("amqps://example.com", opts);

AMQ C++ uses the SASL protocol to perform authentication. SASL can use a number of different authentication mechanisms. When two network peers connect, they exchange their allowed mechanisms, and the strongest mechanism allowed by both is selected.

By default, AMQ C++ allows all of the mechanisms supported by the local SASL library configuration. To restrict the allowed mechanisms and thereby control what mechanisms can be negotiated, use the sasl_allowed_mechs connection option. This option accepts a string containing a space-separated list of mechanism names.

proton::connection_options opts {};

opts.sasl_allowed_mechs("ANONYMOUS");

container.connect("amqps://example.com", opts);

This example forces the connection to authenticate using the ANONYMOUS mechanism even if the server we connect to offers other options. Valid mechanisms include ANONYMOUS, PLAIN, SCRAM-SHA-256, SCRAM-SHA-1, GSSAPI, and EXTERNAL.

AMQ C++ enables SASL by default. To disable it, set the sasl_enabled connection option to false.

proton::connection_options opts {};

opts.sasl_enabled(false);

container.connect("amqps://example.com", opts);

Kerberos is a network protocol for centrally managed authentication based on the exchange of encrypted tickets. See Using Kerberos for more information.

Some message servers support on-demand creation of queues and topics. When a sender or receiver is attached, the server uses the sender target address or the receiver source address to create a queue or topic with a name matching the address.

The message server typically defaults to creating either a queue (for one-to-one message delivery) or a topic (for one-to-many message delivery). The client can indicate which it prefers by setting the queue or topic capability on the source or target.

To select queue or topic semantics, follow these steps:

void on_container_start(proton::container& cont) override {
    proton::connection conn = cont.connect("amqp://example.com");
    proton::sender_options opts {};
    proton::target_options topts {};

    topts.capabilities(std::vector<proton::symbol> { "queue" });
    opts.target(topts);

    conn.open_sender("jobs", opts);
}

void on_container_start(proton::container& cont) override {
    proton::connection conn = cont.connect("amqp://example.com");
    proton::receiver_options opts {};
    proton::source_options sopts {};

    sopts.capabilities(std::vector<proton::symbol> { "topic" });
    opts.source(sopts);

    conn.open_receiver("notifications", opts);
}

For more details, see the following examples:

A durable subscription is a piece of state on the remote server representing a message receiver. Ordinarily, message receivers are discarded when a client closes. However, because durable subscriptions are persistent, clients can detach from them and then re-attach later. Any messages received while detached are available when the client re-attaches.

Durable subscriptions are uniquely identified by combining the client container ID and receiver name to form a subscription ID. These must have stable values so that the subscription can be recovered.

To create a durable subscription, follow these steps:

To detach from a subscription, use the proton::receiver::detach() method. To terminate the subscription, use the proton::receiver::close() method.

For more information, see the durable-subscribe.cpp example.

A shared subscription is a piece of state on the remote server representing one or more message receivers. Because it is shared, multiple clients can consume from the same stream of messages.

The client configures a shared subscription by setting the shared capability on the receiver source.

Shared subscriptions are uniquely identified by combining the client container ID and receiver name to form a subscription ID. These must have stable values so that multiple client processes can locate the same subscription. If the global capability is set in addition to shared, the receiver name alone is used to identify the subscription.

To create a durable subscription, follow these steps:

To detach from a subscription, use the proton::receiver::detach() method. To terminate the subscription, use the proton::receiver::close() method.

For more information, see the shared-subscribe.cpp example.

To send a message, override the on_sendable event handler and call the sender::send() method. The sendable event fires when the proton::sender has enough credit to send at least one message.

struct example_handler : public proton::messaging_handler {
    void on_container_start(proton::container& cont) override {
        proton::connection conn = cont.connect("amqp://example.com");
        conn.open_sender("jobs");
    }

    void on_sendable(proton::sender& snd) override {
        proton::message msg {"job-1"};
        snd.send(msg);
    }
};

When a message is sent, the sender can keep a reference to the tracker object representing the transfer. The receiver accepts or rejects each message that is delivered. The sender is notified of the outcome for each tracked delivery.

To monitor the outcome of a sent message, override the on_tracker_accept and on_tracker_reject event handlers and map the delivery state update to the tracker returned from send().

void on_sendable(proton::sender& snd) override {
    proton::message msg {"job-1"};
    proton::tracker trk = snd.send(msg);
}

void on_tracker_accept(proton::tracker& trk) override {
    std::cout << "Delivery for " << trk << " is accepted\n";
}

void on_tracker_reject(proton::tracker& trk) override {
    std::cout << "Delivery for " << trk << " is rejected\n";
}

To receive messages, create a receiver and override the on_message event handler.

struct example_handler : public proton::messaging_handler {
    void on_container_start(proton::container& cont) override {
        proton::connection conn = cont.connect("amqp://example.com");
        conn.open_receiver("jobs");
    }

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

To explicitly accept or reject a delivery, use the delivery::accept() or delivery::reject() methods in the on_message event handler.

void on_message(proton::delivery& dlv, proton::message& msg) override {
    try {
        process_message(msg);
        dlv.accept();
    } catch (std::exception& e) {
        dlv.reject();
    }
}

By default, if you do not explicity acknowledge a delivery, then the library accepts it after on_message returns. To disable this behavior, set the auto_accept receiver option to false.

If an error is not handled using an override in an event-handling function, an exception is thrown by the container’s run method.

All of the exceptions that AMQ C++ throws inherit from the proton::error class, which in turn inherits from the std::runtime_error and std::exception classes.

The following example illustrates how 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
}

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

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 is also called.

If one of the more specific error handlers is not overridden, the default error handler is called:

The client can log AMQP protocol frames to the console. This data is often critical when diagnosing problems.

To enable protocol logging, set the PN_TRACE_FRM environment variable to 1:

$ export PN_TRACE_FRM=1
$ <your-client-program>

To disable protocol logging, unset the PN_TRACE_FRM environment variable.

The container object can handle multiple connections concurrently. When 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. It is recommended to create a separate handler for each connection so that 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 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 is 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.

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.

To defer work for a fixed amount of time, use the schedule method to set the delay and register a function defining the work.

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

This example uses the schedule method on the work queue of the sender in order to establish it as the execution context for the work.

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.

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.

If no connect.json file is found, the library uses default values for all options.

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

All of the configuration attributes are optional or have default values, so a simple example need only provide a few details:

{
    "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"
    }
}

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

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: