Chapter 5. Using the API

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

Setting the container identity

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

See the Section 5.5, “Security” section for information about creating secure connections.

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

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. After the message is delivered, the receiver accepts or rejects it. 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.

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

To connect to a remote server with SSL/TLS, use a connection URL with the amqps scheme.

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

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. It takes 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.

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.