Tutorial

This is a brief tutorial that will walk you through the fundamentals of building messaging applications in incremental steps.

There are further examples, in addition the ones mentioned in the tutorial.

Proton provides an "event-driven" programming model, where you implement a subclass of `proton::messaging_handler` and override functions that react to various AMQP events (connections opening and closing, messages being delivered, and so on).

The examples below show how to implement handlers for clients and servers and how to run them using the `proton::default_container`, a portable, easy-to-use way to build single-threaded clients or servers.

Some of the examples require an AMQP broker that can receive, store, and send messages. broker.hpp and broker.cpp define a simple example broker. If run without arguments, it listens on `0.0.0.0:5672`, the standard AMQP port on all network interfaces. To use a different port or network interface:

    broker -a <host>:<port>

Instead of the example broker, you can use any AMQP 1.0-compliant broker. You must configure your broker to have a queue (or topic) named "examples".

The `helloworld` examples take an optional URL argument. The other examples take an option `-a URL`. A URL looks like this:

    HOST:PORT/ADDRESS

It usually defaults to `127.0.0.1:5672/examples`, but you can change this if your broker is on a different host or port, or you want to use a different queue or topic name (the ADDRESS part of the URL). URL details are at `proton::url`.

## Hello World!

Tradition dictates that we start with Hello World! This example demonstrates sending and receiving by sending a message to a broker and then receiving the same message back. In a realistic system the sender and receiver would normally be in different processes. The complete example is helloworld.cpp

We will include the following classes: `proton::default_container` (an implementation of `proton::container`) runs an event loop which dispatches events to a `proton::messaging_handler`. This allows a reactive* style of programming which is well suited to messaging applications. `proton::connection` and `proton::delivery` are AMQP entities used in the handler functions. `proton::url` is a simple parser for the URL format mentioned above.

#include <proton/connection.hpp>
#include <proton/default_container.hpp>
#include <proton/delivery.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/url.hpp>

We will define a class `hello_world` which is a subclass of `proton::messaging_handler` and overrides functions to handle the events of interest in sending and receiving a message.

class hello_world : public proton::messaging_handler {
private:
proton::url url;
public:
hello_world(const std::string& u) : url(u) {}

`proton::messaging_handler::on_container_start()` is called when the event loop first starts. We handle that by establishing a connection and creating a sender and a receiver.

void on_container_start(proton::container& c) OVERRIDE {
c.connect(url);
}
void on_connection_open(proton::connection& c) OVERRIDE {
c.open_receiver(url.path());
c.open_sender(url.path());
}

`proton::messaging_handler::on_sendable()` is called when the message can be transferred over the associated sender link to the remote peer. We create a `proton::message`, set the message body to `"Hello World!"`, and send the message. Then we close the sender, since we only want to send one message. Closing the sender will prevent further calls to `proton::messaging_handler::on_sendable()`.

void on_sendable(proton::sender &s) OVERRIDE {
proton::message m("Hello World!");
s.send(m);
s.close();
}

`proton::messaging_handler::on_message()` is called when a message is received. We just print the body of the message and close the connection, as we only want one message

void on_message(proton::delivery &d, proton::message &m) OVERRIDE {
std::cout << m.body() << std::endl;
d.connection().close();
}

The message body is a `proton::value`, see the documentation for more on how to extract the message body as type-safe C++ values.

Our `main` function creates an instance of the `hello_world` handler and a `proton::default_container` using that handler. Calling `proton::container::run` sets things in motion and returns when we close the connection as there is nothing further to do. It may throw an exception, which will be a subclass of `proton::error`. That in turn is a subclass of `std::exception`.

int main(int argc, char **argv) {
try {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:5672/examples";
hello_world hw(url);
proton::default_container(hw).run();
return 0;
} catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
}
return 1;
}

## Hello World, direct!

Though often used in conjunction with a broker, AMQP does not require* this. It also allows senders and receivers to communicate directly if desired.

We will modify our example to send a message directly to itself. This is a bit contrived but illustrates both sides of the direct send and receive scenario. The full code is at helloworld_direct.cpp.

The first difference is that, rather than creating a receiver on the same connection as our sender, we listen for incoming connections by invoking the `proton::container::listen()` method on the container.

void on_container_start(proton::container &c) OVERRIDE {
listener = c.listen(url);
c.open_sender(url);
}

As we only need then to initiate one link, the sender, we can do that by passing in a url rather than an existing connection, and the connection will also be automatically established for us.

We send the message in response to the `proton::messaging_handler::on_sendable()` callback and print the message out in response to the `proton::messaging_handler::on_message()` callback exactly as before.

void on_sendable(proton::sender &s) OVERRIDE {
proton::message m("Hello World!");
s.send(m);
s.close();
}
void on_message(proton::delivery &, proton::message &m) OVERRIDE {
std::cout << m.body() << std::endl;
}

However, we also handle two new events. We now close the connection from the sender's side once the message has been accepted. The acceptance of the message is an indication of successful transfer to the peer. We are notified of that event through the `proton::messaging_handler::on_tracker_accept()` callback.

void on_tracker_accept(proton::tracker &t) OVERRIDE {
t.connection().close();
}

Then, once the connection has been closed, of which we are notified through the `proton::messaging_handler::on_connection_close()` callback, we stop accepting incoming connections. A that point there is no work to be done, the event loop exits, and the `proton::container::run()` method returns.

void on_connection_close(proton::connection&) OVERRIDE {
listener.stop();
}

So now we have our example working without a broker involved!

Note that for this example we pick an "unusual" port 8888 since we are talking to ourselves rather than a broker.

std::string url = argc > 1 ? argv[1] : "127.0.0.1:8888/examples";

## Asynchronous send and receive

Of course, these `Hello World!` examples are very artificial, communicating as they do over a network connection but with the same process. A more realistic example involves communication between separate processes, which could indeed be running on completely separate machines.

Let's separate the sender from the receiver, and transfer more than a single message between them.

We'll start with a simple sender, simple_send.cpp.

As with the previous example, we define the application logic in a class that handles events. Because we are transferring more than one message, we need to keep track of how many we have sent. We'll use a `sent` member variable for that. The `total` member variable will hold the number of messages we want to send.

class simple_send : public proton::messaging_handler {
private:
std::string url;
std::string user;
std::string password;
proton::sender sender;
int sent;
int confirmed;
int total;

As before, we use the `proton::messaging_handler::on_container_start()` event to establish our sender link over which we will transfer messages.

void on_container_start(proton::container &c) OVERRIDE {
proton::connection_options co;
if (!user.empty()) co.user(user);
if (!password.empty()) co.password(password);
sender = c.open_sender(url, co);
}

AMQP defines a credit-based flow-control mechanism. Flow control allows the receiver to control how many messages it is prepared to receive at a given time and thus prevents any component being overwhelmed by the number of messages it is sent.

In the `proton::messaging_handler::on_sendable()` callback, we check that our sender has credit before sending messages. We also check that we haven't already sent the required number of messages.

void on_sendable(proton::sender &s) OVERRIDE {
while (s.credit() && sent < total) {
proton::message msg;
std::map<std::string, int> m;
m["sequence"] = sent + 1;
msg.id(sent + 1);
msg.body(m);
s.send(msg);
sent++;
}
}

The `proton::sender::send()` call above is asynchronous. When it returns, the message has not yet actually been transferred across the network to the receiver. By handling the `proton::messaging_handler::on_tracker_accept()` event, we can get notified when the receiver has received and accepted the message. In our example we use this event to track the confirmation of the messages we have sent. We only close the connection and exit when the receiver has received all the messages we wanted to send.

void on_tracker_accept(proton::tracker &t) OVERRIDE {
confirmed++;
if (confirmed == total) {
std::cout << "all messages confirmed" << std::endl;
t.connection().close();
}
}

If we are disconnected after a message is sent and before it has been confirmed by the receiver, it is said to be "in doubt". We don't know whether or not it was received. In this example, we will handle that by resending any in-doubt messages. This is known as an "at-least-once" guarantee, since each message should eventually be received at least once, though a given message may be received more than once (i.e., duplicates are possible). In the `proton::messaging_handler::on_transport_close()` callback, we reset the sent count to reflect only those that have been confirmed. The library will automatically try to reconnect for us, and when our sender is sendable again, we can restart from the point we know the receiver got to.

void on_transport_close(proton::transport &) OVERRIDE {
sent = confirmed;
}

Now let's look at the corresponding receiver, simple_recv.cpp.

This time we'll use an `expected` member variable for for the number of messages we expect and a `received` variable to count how many we have received so far.

class simple_recv : public proton::messaging_handler {
private:
std::string url;
std::string user;
std::string password;
proton::receiver receiver;
uint64_t expected;
uint64_t received;

We handle `proton::messaging_handler::on_container_start()` by creating our receiver, much like we did for the sender.

void on_container_start(proton::container &c) OVERRIDE {
proton::connection_options co;
if (!user.empty()) co.user(user);
if (!password.empty()) co.password(password);
receiver = c.open_receiver(url, co);
std::cout << "simple_recv listening on " << url << std::endl;
}

We also handle the `proton::messaging_handler::on_message()` event for received messages and print the message out as in the `Hello World!` examples. However, we add some logic to allow the receiver to wait for a given number of messages and then close the connection and exit. We also add some logic to check for and ignore duplicates, using a simple sequential ID scheme.

void on_message(proton::delivery &d, proton::message &msg) OVERRIDE {
if (proton::get<uint64_t>(msg.id()) < received) {
return; // Ignore duplicate
}

## Direct send and receive

Sending between these two examples requires an intermediary broker since neither accepts incoming connections. AMQP allows us to send messages directly between two processes. In that case, one or other of the processes needs to accept incoming connections. Let's create a modified version of the receiving example that does this with direct_recv.cpp.

There are only two differences here. Instead of initiating a link (and implicitly a connection), we listen for incoming connections.

void on_container_start(proton::container &c) OVERRIDE {
listener = c.listen(url);
std::cout << "direct_recv listening on " << url << std::endl;
}

When we have received all the expected messages, we then stop listening for incoming connections by calling `proton::listener::stop()`

void on_message(proton::delivery &d, proton::message &msg) OVERRIDE {
if (proton::coerce<uint64_t>(msg.id()) < received) {
return; // Ignore duplicate
}
if (expected == 0 || received < expected) {
std::cout << msg.body() << std::endl;
received++;
}
if (received == expected) {
d.receiver().close();
d.connection().close();
listener.stop();
}
}

You can use the simple_send.cpp example to send to this receiver directly. (Note: you will need to stop any broker that is listening on the 5672 port, or else change the port used by specifying a different address to each example via the `-a` command-line switch).

We can also modify the sender to allow the original receiver to connect to it, in direct_send.cpp. Again, that requires just two modifications:

As with the modified receiver, instead of initiating establishment of a link, we listen for incoming connections.

void on_container_start(proton::container &c) OVERRIDE {
listener = c.listen(url);
std::cout << "direct_send listening on " << url << std::endl;
}

When we have received confirmation of all the messages we sent, we call `container::listener::stop()` to exit.

void on_tracker_accept(proton::tracker &t) OVERRIDE {
confirmed++;
if (confirmed == total) {
std::cout << "all messages confirmed" << std::endl;
t.connection().close();
listener.stop();
}
}

To try this modified sender, run the original simple_recv.cpp against it.

The symmetry in the underlying AMQP wire protocol that enables this is quite unique and elegant, and in reflecting this the Proton API provides a flexible toolkit for implementing all sorts of interesting intermediaries.

## Request and response

A common pattern is to send a request message and expect a response message in return. AMQP has special support for this pattern. Let's have a look at a simple example. We'll start with server.cpp, the program that will process the request and send the response. Note that we are still using a broker in this example.

Our server will provide a very simple service: it will respond with the body of the request converted to uppercase.

class server : public proton::messaging_handler {
private:
typedef std::map<std::string, proton::sender> sender_map;
proton::url url;
proton::connection connection;
sender_map senders;
public:
server(const std::string &u) : url(u) {}
void on_container_start(proton::container &c) OVERRIDE {
connection = c.connect(url);
connection.open_receiver(url.path());
std::cout << "server connected to " << url << std::endl;
}
std::string to_upper(const std::string &s) {
std::string uc(s);
size_t l = uc.size();
for (size_t i=0; i<l; i++)
uc[i] = static_cast<char>(std::toupper(uc[i]));
return uc;
}
void on_message(proton::delivery &, proton::message &m) OVERRIDE {
std::cout << "Received " << m.body() << std::endl;
std::string reply_to = m.reply_to();
proton::message reply;
reply.to(reply_to);
reply.body(to_upper(proton::get<std::string>(m.body())));
reply.correlation_id(m.correlation_id());
if (!senders[reply_to]) {
senders[reply_to] = connection.open_sender(reply_to);
}
senders[reply_to].send(reply);
}
};

The code here is not too different from the simple receiver example. However, when we receive a request in `proton::messaging_handler::on_message`, we look at the `proton::message::reply_to` address and create a sender with that address for the response. We'll cache the senders in case we get further requests with the same `reply_to`.

Now let's create a simple client.cpp to test this service out.

Our client takes a list of strings to send as requests.

client(const std::string &u, const std::vector<std::string>& r) : url(u), requests(r) {}

Since we will be sending and receiving, we create a sender and a receiver in `proton::messaging_handler::on_container_start`. Our receiver has a blank address and sets the `dynamic` flag to true, which means we expect the remote end (the broker or server) to assign a unique address for us.

void on_container_start(proton::container &c) OVERRIDE {
sender = c.open_sender(url);
// Create a receiver requesting a dynamically created queue
// for the message source.
receiver_options opts = receiver_options().source(source_options().dynamic(true));
receiver = sender.connection().open_receiver("", opts);
}

Now we need a function to send the next request from our list of requests. We set the `reply_to` address to be the dynamically assigned address of our receiver.

void send_request() {
proton::message req;
req.body(requests.front());
req.reply_to(receiver.source().address());
sender.send(req);
}

We need to use the address assigned by the broker as the `reply_to` address of our requests, so we can't send them until our receiver has been set up. To do that, we add an `proton::messaging_handler::on_receiver_open()` method to our handler class and use that as the trigger to send our first request.

void on_receiver_open(proton::receiver &) OVERRIDE {
send_request();
}

When we receive a reply, we send the next request.

void on_message(proton::delivery &d, proton::message &response) OVERRIDE {
if (requests.empty()) return; // Spurious extra message!
std::cout << requests.front() << " => " << response.body() << std::endl;
requests.erase(requests.begin());
if (!requests.empty()) {
send_request();
} else {
d.connection().close();
}
}

## Direct request and response

We can avoid the intermediary process by writing a server that accepts connections directly, server_direct.cpp. It involves the following changes to our original server:

Our server must generate unique `reply-to` addresses for links from the client that request a dynamic address (previously this was done by the broker). We use a simple counter.

std::string generate_address() {
std::ostringstream addr;
addr << "server" << address_counter++;
return addr.str();
}

Next we need to handle incoming requests for links with dynamic addresses from the client. We give the link a unique address and record it in our `senders` map.

void on_sender_open(proton::sender &sender) OVERRIDE {
if (sender.source().dynamic()) {
std::string addr = generate_address();
sender.open(proton::sender_options().source(proton::source_options().address(addr)));
senders[addr] = sender;
}

Note that we are interested in sender links above because we are implementing the server. A receiver link created on the client corresponds to a sender link on the server.

Finally when we receive a message we look up its `reply_to` in our senders map and send the reply.

void on_message(proton::delivery &, proton::message &m) OVERRIDE {
std::cout << "Received " << m.body() << std::endl;
std::string reply_to = m.reply_to();
sender_map::iterator it = senders.find(reply_to);
if (it == senders.end()) {
std::cout << "No link for reply_to: " << reply_to << std::endl;
} else {
proton::sender sender = it->second;
proton::message reply;
reply.to(reply_to);
reply.body(to_upper(proton::get<std::string>(m.body())));
reply.correlation_id(m.correlation_id());
sender.send(reply);
}
}