AMQP-CPP

AMQP-CPP is a C++ library for communicating with a RabbitMQ message broker. The library can be used to parse incoming data from, and generate frames to, a RabbitMQ server.

OVERVIEW

Are you upgrading from AMQP-CPP 3 to AMQP-CPP 4? Please read the upgrade instructions

Note for the reader: This readme file has a peculiar structure. We start explaining the pure and hard core low level interface in which you have to take care of opening socket connections yourself. In reality, you probably want to use the simpler TCP interface that is being described later on.

This library has a layered architecture, and allows you - if you like - to completely take care of the network layer. If you want to set up and manage the network connections yourself, the AMQP-CPP library will not make a connection to RabbitMQ by itself, nor will it create sockets and/or perform IO operations. As a user of this library, you create the socket connection and implement an interface defined by AMQP-CPP, passing it to the AMQP-CPP library to use for IO operations.

Intercepting this network layer is optional. The AMQP-CPP library comes with a predefined TCP and TLS module that can be used if you trust the AMQP library to take care of the network (and optional TLS) handling. In that case, the AMQP-CPP library does all the system and library calls to set up network connections, and send/receive the (possibly encrypted) data.

This layered architecture makes the library flexible and portable: it does not necessarily rely on operating system specific IO calls, and can be easily integrated into any kind of event loop. If you want to implement the AMQP protocol on top of some other unusual communication layer, this library can be used for that - but if you want to use it with regular TCP connections, setting it up is just as easy.

AMQP-CPP is fully asynchronous and does not do any blocking (system) calls, so it can be used in high performance applications without the need for threads.

The AMQP-CPP library uses C++17 features, so if you intend to use it, please make sure that your compiler is up-to-date and supports C++17.

TABLE OF CONTENTS

• Overview
• About
• How to install
• How to use AMQP-CPP
• Parsing incoming data
• TCP connections
• Secure connections
• Existing event loops
• Heartbeats
• Channels
• Channel callbacks
• Channel errors
• Flags and Tables
• Publishing messages
• Publisher confirms
• Consuming messages
• Upgrading
• Work in progress

ABOUT

Back to Table of Contents

This library is created and maintained by Copernica (www.copernica.com), and is used inside the MailerQ (www.mailerq.com) and Yothalot (www.yothalot.com) applications. MailerQ is a tool for sending large volumes of email, using AMQP message queues, and Yothalot is a big data processing map/reduce framework.

Do you appreciate our work and are you looking for high quality email solutions? Then check out our other commercial and open source solutions:

• Copernica Marketing Suite (www.copernica.com)
• MailerQ on-premise MTA (www.mailerq.com)
• Responsive Email web service (www.responsiveemail.com)
• SMTPeter cloud based SMTP server (www.smtpeter.com)
• PHP-CPP bridge between PHP and C++ (www.php-cpp.com)
• PHP-JS bridge between PHP and Javascript (www.php-js.com)
• Yothalot big data processor (www.yothalot.com)

HOW TO INSTALL

Back to Table of Contents

Start by cloning the repository and navigating to the AMQP-CPP directory.

git clone https://github.com/CopernicaMarketingSoftware/AMQP-CPP.git
cd AMQP-CPP

There are two methods to compile AMQP-CPP: CMake and Make. CMake is platform portable and works on all systems, while the Makefile only works on Linux. The two methods create both a shared and a static version of the AMQP-CPP library. Building of a shared library is currently not supported on Windows.

AMQP-CPP comes with an optional Linux-only TCP module that takes care of the network part required for the AMQP-CPP core library. If you use this module, you are required to link with pthread and dl.

After building there are two relevant files to #include when you use the library.

File	Include when?
amqpcpp.h	Always needed for the core features
amqpcpp/linux_tcp.h	If using the Linux-only TCP module

On Windows you are required to define NOMINMAX when compiling code that includes public AMQP-CPP header files.

Using cmake

The CMake file supports both building and installing. You can choose not to use the install functionality, and instead manually use the build output at build/bin/. Keep in mind that the TCP module is only supported for Linux. An example install method would be:

mkdir build
cd build
cmake .. [-DAMQP-CPP_BUILD_SHARED=ON] [-DAMQP-CPP_LINUX_TCP=ON]
cmake --build . --target install

Option	Default	Meaning
AMQP-CPP_BUILD_SHARED	OFF	OFF for static lib, ON for shared lib. Shared is not supported on Windows.
AMQP-CPP_LINUX_TCP	OFF	ON to build TCP module. TCP module is supported for Linux only.

Using make

Compiling and installing AMQP-CPP with make is as easy as running:

make
make install

This will install the full version of AMQP-CPP, including the system specific TCP module.

To install without the TCP module (so that you can handle network connection yourself), run:

make pure
make install

Compiling a program

When you compile an application that uses the AMQP-CPP library, remember to link with the library. For gcc and clang the linker flag is -lamqpcpp.

If you use the TCP module, you also need to pass the -lpthread and -ldl linker flags. The TCP module uses a thread for running an asynchronous and non-blocking DNS hostname lookup, and it must be linked with the dl library to allow dynamic lookups for functions from the openssl library if a secure connection to RabbitMQ has to be set up.

An example compilation command for an application using the TCP module:

g++ -g -Wall -lamqcpp -lpthread -ldl my-amqp-cpp.c -o my-amqp-cpp

HOW TO USE AMQP-CPP

Back to Table of Contents

AMQP-CPP operates in a network-agnostic fashion. It does not do IO by itself. An object must be provided that defines the IO operations. We have provided the ConnectionHandler base class for you to extend from and create your own object. This class defines a number of methods called by the library any time it wants to send data, or if it wants to notify you an error has occurred.

#include <amqpcpp.h>

// You'll need to extend the ConnectionHandler class and make your own, like this
class MyConnectionHandler : public AMQP::ConnectionHandler
{
    /**
     *  Method that is called by the AMQP library every time it has data
     *  available that should be sent to RabbitMQ.
     *  @param  connection  pointer to the main connection object
     *  @param  data        memory buffer with the data that should be sent to RabbitMQ
     *  @param  size        size of the buffer
     */
    virtual void onData(AMQP::Connection *connection, const char *data, size_t size)
    {
        // @todo
        //  Add your own implementation, for example by doing a call to the
        //  send() system call. But be aware that the send() call may not
        //  send all data at once, so you also need to take care of buffering
        //  the bytes that could not immediately be sent, and try to send
        //  them again when the socket becomes writable again
    }

    /**
     *  Method that is called by the AMQP library when the login attempt
     *  succeeded. After this method has been called, the connection is ready
     *  to use.
     *  @param  connection      The connection that can now be used
     */
    virtual void onReady(AMQP::Connection *connection)
    {
        // @todo
        //  add your own implementation, for example by creating a channel
        //  instance, and start publishing or consuming
    }

    /**
     *  Method that is called by the AMQP library when a fatal error occurs
     *  on the connection, for example because data received from RabbitMQ
     *  could not be recognized.
     *  @param  connection      The connection on which the error occurred
     *  @param  message         A human readable error message
     */
    virtual void onError(AMQP::Connection *connection, const char *message)
    {
        // @todo
        //  add your own implementation, for example by reporting the error
        //  to the user of your program, log the error, and destruct the
        //  connection object because it is no longer in a usable state
    }

    /**
     *  Method that is called when the connection was closed. This is the
     *  counter part of a call to Connection::close() and it confirms that the
     *  AMQP connection was correctly closed.
     *
     *  @param  connection      The connection that was closed and that is now unusable
     */
    virtual void onClosed(AMQP::Connection *connection) 
    {
        // @todo
        //  add your own implementation, for example by closing down the
        //  underlying TCP connection too
    }


};

Even though ConnectionHandler methods are not marked noexcept explicitly, you are not expected to throw from them, and the behaviour is undefined if you do so.

After you've implemented the ConnectionHandler class the way you like, you can start using the library by creating a Connection object, and one or more Channel objects:

// create an instance of your own connection handler
MyConnectionHandler myHandler;

// create a AMQP connection object
AMQP::Connection connection(&myHandler, AMQP::Login("guest","guest"), "/");

// and create a channel
AMQP::Channel channel(&connection);

// use the channel object to call the AMQP method you like
channel.declareExchange("my-exchange", AMQP::fanout);
channel.declareQueue("my-queue");
channel.bindQueue("my-exchange", "my-queue", "my-routing-key");

A number of remarks about the example above. You may notice that we've created all objects on the stack. You are also free to create them on the heap with the C++ operator new. That works just as well, and in a real project you will likely want to keep your Handler, Connection, and Channel objects around for a longer time.

But more importantly, you can see in the example above that we instantiated the Channel object directly after we made the Connection object, and we also started declaring exchanges and queues right away.

However, under the hood, a handshake protocol is executed between the server and the client when the Connection object is first created. During this handshake procedure other operations are not permitted (like opening a channel or declaring a queue).

It would be better practice to wait for the connection to be ready (implementing the MyConnectionHandler::OnReady() method) and creating the Channel object only then.

However, this is not strictly necessary. Methods called during a handshake are cached by the AMQP library, and will be executed the moment the handshake is completed and the connection becomes ready for use.

PARSING INCOMING DATA

Back to Table of Contents

The ConnectionHandler class has a method onData() that is called by the library every time that it wants to send data. The onData method is implemented by you, the user. For example, you might make system calls to send() or write() to send data to the RabbitMQ server. But what about data in the other direction? How does the library receive data from RabbitMQ?

In this raw setup, the AMQP-CPP library does not do IO by itself, and so does not receive any data from a socket. You will have to create a socket that connects to the RabbitMQ server yourself.

Inside your event loop, after checking the socket is readable, you should read out that socket (for example by using the recv() system call), and pass the received bytes to the AMQP-CPP library. This is done by calling the parse() method in the Connection object.

The Connection::parse() method gets two parameters, a pointer to a buffer of data that you just read from the socket, and a parameter that holds the size of this buffer. The code snippet below comes from the Connection.h C++ header file.

````c++ /** * Parse data that was received from RabbitMQ * * Every time that data comes in from RabbitMQ, you should call this method to parse * the incoming data, and let it handle by the AMQP-CPP library. This method returns * the number of bytes that were processed. * * If not all bytes could be processed because it only contained a partial frame, * you should call this same method later on when more data is available. The * AMQP-CPP library does not do any buffering, so it is up to the caller to ensur