diff --git a/README.md b/README.md index cb1b0e16..dad41025 100644 --- a/README.md +++ b/README.md @@ -710,7 +710,7 @@ it has a couple of open issues. | TCP Handler Impl | Header File Location | Sample File Location | | ----------------------- | ---------------------- | ------------------------- | -| Boost asio (io_service) | include/libboostasio.h | examples/libboostasio.cpp | +| Boost asio | include/libboostasio.h | examples/libboostasio.cpp | | libev | include/libev.h | examples/libev.cpp | | libevent | include/libevent.h | examples/libevent.cpp | | libuv | include/libuv.h | examples/libuv.cpp | @@ -770,7 +770,7 @@ class MyTcpHandler : public AMQP::TcpHandler ```` If you enable heartbeats, it is your own responsibility to ensure that the -```connection->heartbeat()``` method is called at least once during this period, +`connection->heartbeat()` method is called at least once during this period, or that you call one of the other channel or connection methods to send data over the connection. Heartbeats are sent by the server too, RabbitMQ also ensures that _some data_ is sent over the connection from the server to the client diff --git a/examples/libboostasio.cpp b/examples/libboostasio.cpp index d290382d..57e88d01 100644 --- a/examples/libboostasio.cpp +++ b/examples/libboostasio.cpp @@ -5,15 +5,14 @@ * * @author Gavin Smith * - * Compile with g++ -std=c++14 libboostasio.cpp -o boost_test -lpthread -lboost_system -lamqpcpp + * Compile with + * g++ -std=c++17 libboostasio.cpp -o boost_test -lpthread -ldl -lboost_system -lamqpcpp */ /** * Dependencies */ -#include -#include -#include +#include #include @@ -27,30 +26,41 @@ int main() { // access to the boost asio handler - // note: we suggest use of 2 threads - normally one is fin (we are simply demonstrating thread safety). - boost::asio::io_service service(4); + boost::asio::io_context context; + + // handler for libboostasio + AMQP::LibBoostAsioHandler handler(context); - // handler for libev - AMQP::LibBoostAsioHandler handler(service); - // make a connection AMQP::TcpConnection connection(&handler, AMQP::Address("amqp://guest:guest@localhost/")); - + // we need a channel too AMQP::TcpChannel channel(&connection); - + + channel.onReady([]() { + std::cout << "channel ready" << std::endl; + }); + channel.onError([](const char *message) { + std::cout << "channel error: " << message << std::endl; + }); + // create a temporary queue channel.declareQueue(AMQP::exclusive).onSuccess([&connection](const std::string &name, uint32_t messagecount, uint32_t consumercount) { - + // report the name of the temporary queue std::cout << "declared queue " << name << std::endl; - + // now we can close the connection connection.close(); + + }).onError([](const char *message) { + + // something went wrong creating the queue (or even before) + std::cerr << "declareQueue error: " << message << std::endl; }); - + // run the handler - // a t the moment, one will need SIGINT to stop. In time, should add signal handling through boost API. - return service.run(); + // at the moment, one will need SIGINT to stop. In time, should add signal handling through boost API. + return context.run(); } diff --git a/include/amqpcpp/libboostasio.h b/include/amqpcpp/libboostasio.h index 0df9011f..a76808e3 100644 --- a/include/amqpcpp/libboostasio.h +++ b/include/amqpcpp/libboostasio.h @@ -1,17 +1,18 @@ /** * LibBoostAsio.h * - * Implementation for the AMQP::TcpHandler for boost::asio. You can use this class - * instead of a AMQP::TcpHandler class, just pass the boost asio service to the - * constructor and you're all set. See tests/libboostasio.cpp for example. + * Implementation for the AMQP::TcpHandler for boost::asio. You can use this class + * instead of a AMQP::TcpHandler class, just pass the boost asio io_context to the + * constructor and you're all set. See examples/libboostasio.cpp for an example. * - * Watch out: this class was not implemented or reviewed by the original author of + * Watch out: this class was not implemented or reviewed by the original author of * AMQP-CPP. However, we do get a lot of questions and issues from users of this class, * so we cannot guarantee its quality. If you run into such issues too, it might be * better to implement your own handler that interact with boost. * * * @author Gavin Smith + * @author Paolo Pastori */ @@ -24,24 +25,18 @@ * Dependencies */ #include +#include +#include +#include #include -#include -#include +#include +#include #include #include -#include -#include #include "amqpcpp/linux_tcp.h" -// C++17 has 'weak_from_this()' support. -#if __cplusplus >= 201701L -#define PTR_FROM_THIS(T) weak_from_this() -#else -#define PTR_FROM_THIS(T) std::weak_ptr(shared_from_this()) -#endif - /** * Set up namespace */ @@ -51,28 +46,22 @@ namespace AMQP { * Class definition * @note Because of a limitation on Windows, this will only work on POSIX based systems - see https://github.com/chriskohlhoff/asio/issues/70 */ -class LibBoostAsioHandler : public virtual TcpHandler +class LibBoostAsioHandler : public TcpHandler { protected: /** * Helper class that wraps a boost io_context socket monitor. */ - class Watcher : public virtual std::enable_shared_from_this + class Watcher : public std::enable_shared_from_this { private: - /** - * The boost asio io_context which is responsible for detecting events. - * @var class boost::asio::io_context& - */ - boost::asio::io_context & _iocontext; - using strand_weak_ptr = std::weak_ptr; /** * The boost asio io_context::strand managed pointer. - * @var class std::shared_ptr + * @var std::weak_ptr */ strand_weak_ptr _wpstrand; @@ -84,241 +73,278 @@ class LibBoostAsioHandler : public virtual TcpHandler boost::asio::posix::stream_descriptor _socket; /** - * The boost asynchronous deadline timer. - * @var class boost::asio::deadline_timer + * The boost asynchronous timer. + * + * Used for monitoring of both server connection timeout condition + * (at initial connection stage) and heartbeat (client and server). + * @var class boost::asio::steady_timer */ - boost::asio::deadline_timer _timer; + boost::asio::steady_timer _timer; /** - * A boolean that indicates if the watcher is monitoring for read events. - * @var _read True if reads are being monitored else false. + * Timeout after which the connection is no longer considered alive. + * A heartbeat must be sent every _timeout / 2 seconds. + * Value zero means heartbeats are disabled, or not yet negotiated. + * @var uint16_t */ - bool _read{false}; + uint16_t _timeout = 0; /** - * A boolean that indicates if the watcher has a pending read event. - * @var _read True if read is pending else false. + * AMQP Server connection timeout setting (seconds). + * @var uint16_t */ - bool _read_pending{false}; + uint16_t _connection_timeout = 0; + + using steady_time_point = std::chrono::time_point; /** - * A boolean that indicates if the watcher is monitoring for write events. - * @var _read True if writes are being monitored else false. + * When should we send the next heartbeat? + * @var std::chrono::time_point */ - bool _write{false}; + steady_time_point _next; /** - * A boolean that indicates if the watcher has a pending write event. - * @var _read True if read is pending else false. + * When does the connection expire / was the server idle for too long? + * @var std::chrono::time_point */ - bool _write_pending{false}; + steady_time_point _expire; - using handler_cb = boost::function; - using io_handler = boost::function; - using timer_handler = boost::function; + /** + * The events for which the socket filedescriptor is actually monitored. + * @var int + */ + int _events = 0; /** - * Builds a io handler callback that executes the io callback in a strand. - * @param io_handler The handler callback to dispatch - * @return handler_cb A function wrapping the execution of the handler function in a io_context::strand. + * A boolean that indicates if the watcher is monitoring for read events. + * @var _read True if reads are being monitored else false. */ - handler_cb get_dispatch_wrapper(io_handler fn) - { - const strand_weak_ptr wpstrand = _wpstrand; + bool _read = false; - return [fn, wpstrand](const boost::system::error_code &ec, const std::size_t bytes_transferred) - { - const strand_shared_ptr strand = wpstrand.lock(); - if (!strand) - { - fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled), std::size_t{0}); - return; - } - boost::asio::dispatch(strand->context().get_executor(), boost::bind(fn, ec, bytes_transferred)); - }; - } + /** + * A boolean that indicates if the watcher has a pending read event. + * @var _read_pending True if read is pending else false. + */ + bool _read_pending = false; /** - * Binds and returns a read handler for the io operation. - * @param connection The connection being watched. - * @param fd The file descripter being watched. - * @return handler callback + * A boolean that indicates if the watcher is monitoring for write events. + * @var _write True if writes are being monitored else false. */ - handler_cb get_read_handler(TcpConnection *const connection, const int fd) - { - auto fn = boost::bind(&Watcher::read_handler, - this, - boost::placeholders::_1, - boost::placeholders::_2, - PTR_FROM_THIS(Watcher), - connection, - fd); - return get_dispatch_wrapper(fn); - } + bool _write = false; /** - * Binds and returns a read handler for the io operation. - * @param connection The connection being watched. - * @param fd The file descripter being watched. - * @return handler callback + * A boolean that indicates if the watcher has a pending write event. + * @var _write_pending True if read is pending else false. */ - handler_cb get_write_handler(TcpConnection *const connection, const int fd) - { - auto fn = boost::bind(&Watcher::write_handler, - this, - boost::placeholders::_1, - boost::placeholders::_2, - PTR_FROM_THIS(Watcher), - connection, - fd); - return get_dispatch_wrapper(fn); - } + bool _write_pending = false; + + using handler_cb = std::function; /** - * Binds and returns a lamba function handler for the io operation. - * @param connection The connection being watched. - * @param timeout The file descripter being watched. - * @return handler callback + * Make a generic handler callback. + * @param mmfn The wrapping member pointer to be invoked by the handler. + * @param connection The connection being watched. + * @param fd The file descriptor being watched. + * @return handler_cb */ - timer_handler get_timer_handler(TcpConnection *const connection, const uint16_t timeout) + template + handler_cb make_handler(M &&mmfn, TcpConnection *const connection, const int fd) { - const auto fn = boost::bind(&Watcher::timeout_handler, - this, - boost::placeholders::_1, - PTR_FROM_THIS(Watcher), - connection, - timeout); - +#if __cplusplus >= 201701L + // C++17 has weak_from_this() + std::weak_ptr wpthis = weak_from_this(); +#else + std::weak_ptr wpthis(shared_from_this()); +#endif const strand_weak_ptr wpstrand = _wpstrand; - return [fn, wpstrand](const boost::system::error_code &ec) - { - const strand_shared_ptr strand = wpstrand.lock(); - if (!strand) + return +#if __cplusplus >= 201402L + // C++14 lambda has init capture + [wpthis=std::move(wpthis), wpstrand=std::move(wpstrand), mmfn=std::forward(mmfn), + connection, fd] (const boost::system::error_code& ec) +#else + [wpthis, wpstrand, mmfn, connection, fd] (const boost::system::error_code& ec) +#endif { - fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled)); - return; - } - boost::asio::dispatch(strand->context().get_executor(), boost::bind(fn, ec)); - }; + const std::shared_ptr spwatcher = wpthis.lock(); + // is the watcher still here? + if (!spwatcher) return; + + const strand_shared_ptr strand = wpstrand.lock(); + // is the strand still here? + if (!strand) return; + + boost::asio::dispatch(*strand, + // moving spwatcher into the bind ensures that the watcher + // will not be destroyed for the duration of the callback + std::bind(std::move(mmfn), std::move(spwatcher), ec, connection, fd)); + }; } /** * Handler method that is called by boost's io_context when the socket pumps a read event. * @param ec The status of the callback. - * @param bytes_transferred The number of bytes transferred. - * @param awpWatcher A weak pointer to this object. * @param connection The connection being watched. * @param fd The file descriptor being watched. - * @note The handler will get called if a read is cancelled. */ void read_handler(const boost::system::error_code &ec, - const std::size_t bytes_transferred, - const std::weak_ptr awpWatcher, TcpConnection *const connection, const int fd) { - // Resolve any potential problems with dangling pointers - // (remember we are using async). - const std::shared_ptr apWatcher = awpWatcher.lock(); - if (!apWatcher) { return; } - _read_pending = false; - if ((!ec || ec == boost::asio::error::would_block) && _read) + if (!ec && _read) { + if (_timeout) + { + // the server is sending data, update the _expire time + _expire = std::chrono::steady_clock::now() + + std::chrono::seconds(_timeout + (_timeout >> 1) + 1); + } + connection->process(fd, AMQP::readable); - _read_pending = true; + // still we need monitoring read? + if (_socket.is_open()) + { + _read_pending = true; - _socket.async_read_some( - boost::asio::null_buffers(), - get_read_handler(connection, fd)); + _socket.async_wait( + boost::asio::posix::stream_descriptor::wait_read, + get_read_handler(connection, fd)); + } } } + /** + * Make a handler callback to invoke the read_handler method. + */ + handler_cb get_read_handler(TcpConnection *const connection, const int fd) + { + return make_handler(std::mem_fn(&Watcher::read_handler), connection, fd); + } /** * Handler method that is called by boost's io_context when the socket pumps a write event. * @param ec The status of the callback. - * @param bytes_transferred The number of bytes transferred. - * @param awpWatcher A weak pointer to this object. * @param connection The connection being watched. * @param fd The file descriptor being watched. - * @note The handler will get called if a write is cancelled. */ void write_handler(const boost::system::error_code ec, - const std::size_t bytes_transferred, - const std::weak_ptr awpWatcher, TcpConnection *const connection, const int fd) { - // Resolve any potential problems with dangling pointers - // (remember we are using async). - const std::shared_ptr apWatcher = awpWatcher.lock(); - if (!apWatcher) { return; } - _write_pending = false; - if ((!ec || ec == boost::asio::error::would_block) && _write) + if (!ec && _write) { + if (_timeout) + { + // the client is sending data, update the _next time + _next = std::chrono::steady_clock::now() + + std::chrono::seconds((_timeout >> 1) + 1); + } + connection->process(fd, AMQP::writable); - _write_pending = true; + // still we need monitoring write? + if (_socket.is_open()) + { + _write_pending = true; - _socket.async_write_some( - boost::asio::null_buffers(), - get_write_handler(connection, fd)); + _socket.async_wait( + boost::asio::posix::stream_descriptor::wait_write, + get_write_handler(connection, fd)); + } } } - /** - * Callback method that is called by libev when the timer expires - * @param ec error code returned from loop - * @param loop The loop in which the event was triggered - * @param connection - * @param timeout + * Make a handler callback to invoke the write_handler method. */ - void timeout_handler(const boost::system::error_code &ec, - std::weak_ptr awpThis, - TcpConnection *const connection, - const uint16_t timeout) + handler_cb get_write_handler(TcpConnection *const connection, const int fd) { - // Resolve any potential problems with dangling pointers - // (remember we are using async). - const std::shared_ptr apTimer = awpThis.lock(); - if (!apTimer) { return; } + return make_handler(std::mem_fn(&Watcher::write_handler), connection, fd); + } + /** + * Handler method that is called by boost's io_context when the timer expires. + * @param ec The status of the callback. + * @param connection The connection being watched. + * @param fd The file descriptor being watched. + */ + void timer_handler(const boost::system::error_code &ec, + TcpConnection *const connection, + const int fd) + { if (!ec) { - if (connection) + steady_time_point now = std::chrono::steady_clock::now(); + + if (_timeout == 0) + { + // this can happen in three situations: + // 1. a connection timeout, + // 2. user space has overidden onNegotiate to reject heartbeats + // 3. AMQP server does not want heartbeats + // in either case we're no longer going to run further timers. + + // if we have an initialized connection, user space must have overidden + // the onNegotiate method, so we keep using the connection + if (connection->initialized()) return; + + // this is a connection timeout, close the connection with immediate effect + return (void) connection->close(true); + } + else if (now >= _expire) + { + // the server was inactive for a too long period of time, + // close the connection with immediate effect + _timeout = 0; + return (void) connection->close(true); + } + else if (now >= _next) { // send the heartbeat connection->heartbeat(); + + // when we should send out the next one + _next = now + std::chrono::seconds((_timeout >> 1) + 1); } - // Reschedule the timer for the future: - _timer.expires_at(_timer.expires_at() + boost::posix_time::seconds(timeout)); + // reschedule the timer + _timer.expires_at(_next); // Posts the timer event - _timer.async_wait(get_timer_handler(connection, timeout)); + _timer.async_wait(get_timer_handler(connection, fd)); } } + /** + * Make a handler callback to invoke the timer_handler method. + */ + handler_cb get_timer_handler(TcpConnection *const connection, const int fd) + { + return make_handler(std::mem_fn(&Watcher::timer_handler), connection, fd); + } public: + /** - * Constructor- initialises the watcher and assigns the filedescriptor to + * Constructor - initialises the watcher and assigns the filedescriptor to * a boost socket for monitoring. - * @param io_context The boost io_context - * @param wpstrand A weak pointer to a io_context::strand instance. - * @param fd The filedescriptor being watched + * @param io_context The boost io_context + * @param wpstrand A weak pointer to a io_context::strand instance. + * @param fd The filedescriptor being watched + * @param connection_timeout The AMQP server connection timeout */ Watcher(boost::asio::io_context &io_context, const strand_weak_ptr wpstrand, - const int fd) : - _iocontext(io_context), + const int fd, + uint16_t connection_timeout) : _wpstrand(wpstrand), _socket(io_context), - _timer(io_context) + _timer(io_context), + _connection_timeout(connection_timeout) { _socket.assign(fd); @@ -334,64 +360,109 @@ class LibBoostAsioHandler : public virtual TcpHandler Watcher(const Watcher &that) = delete; /** - * Destructor + * Destructors */ - ~Watcher() + void close() { - _read = false; - _write = false; + // release ownership of filedescriptor and cancel pending io callbacks _socket.release(); + // cancel any pending callback waiting on the timer stop_timer(); } + ~Watcher() { close(); } + /** * Change the events for which the filedescriptor is monitored + * @param connection The connection being watched. + * @param fd The file descripter being watched. * @param events */ - void events(TcpConnection *connection, int fd, int events) + void events(TcpConnection *const connection, int fd, int events) { - // 1. Handle reads? - _read = ((events & AMQP::readable) != 0); - - // Read requsted but no read pending? - if (_read && !_read_pending) + if (events != _events) { - _read_pending = true; + // cancel pending io callback + _socket.cancel(); - _socket.async_read_some( - boost::asio::null_buffers(), - get_read_handler(connection, fd)); - } + // handle reads? + _read = ((events & AMQP::readable) != 0); + + if (_read) + { + _read_pending = true; + + _socket.async_wait( + boost::asio::posix::stream_descriptor::wait_read, + get_read_handler(connection, fd)); + } - // 2. Handle writes? - _write = ((events & AMQP::writable) != 0); + // handle writes? + _write = ((events & AMQP::writable) != 0); - // Write requested but no write pending? - if (_write && !_write_pending) + if (_write) + { + _write_pending = true; + + _socket.async_wait( + boost::asio::posix::stream_descriptor::wait_write, + get_write_handler(connection, fd)); + } + + // remember current events + _events = events; + } + } + + /** + * Apply AMQP server connection timeout watching. + * @param connection The connection being watched. + * @param fd The file descripter being watched. + */ + void apply_connection_timeout(TcpConnection *const connection, const int fd) + { + if (_connection_timeout && !_timeout) { - _write_pending = true; + // schedule the timer + _timer.expires_after(std::chrono::seconds(_connection_timeout)); - _socket.async_write_some( - boost::asio::null_buffers(), - get_write_handler(connection, fd)); + // Posts the timer event + _timer.async_wait(get_timer_handler(connection, fd)); } } /** - * Change the expire time - * @param connection - * @param timeout + * Configure the heartbeat interval to use + * @param timeout The timeout value (seconds, 0 to disable heartbeat monitor.) */ - void set_timer(TcpConnection *connection, uint16_t timeout) + void set_heartbeat(uint16_t timeout) + { + _timeout = timeout; + } + + /** + * Schedule the timer + * @param connection The connection being watched. + * @param fd The file descripter being watched. + */ + void set_timer(TcpConnection *const connection, const int fd) { // stop timer in case it was already set stop_timer(); - // Reschedule the timer for the future: - _timer.expires_from_now(boost::posix_time::seconds(timeout)); + if (_timeout) + { + // when we should send out the next heartbeat + _next = std::chrono::steady_clock::now() + std::chrono::seconds((_timeout >> 1) + 1); + // by when we should expect some server activity + _expire = _next + std::chrono::seconds(_timeout); + + // schedule the timer + _timer.expires_at(_next); - // Posts the timer event - _timer.async_wait(get_timer_handler(connection, timeout)); + // Posts the timer event + _timer.async_wait(get_timer_handler(connection, fd)); + } } /** @@ -414,25 +485,29 @@ class LibBoostAsioHandler : public virtual TcpHandler /** * The boost asio io_context::strand managed pointer. - * @var class std::shared_ptr + * @var std::shared_ptr */ strand_shared_ptr _strand; /** - * All I/O watchers that are active, indexed by their filedescriptor - * @var std::map + * Active I/O watchers, indexed by their filedescriptor. + * @var std::map */ std::map > _watchers; + /** + * AMQP Server connection timeout setting (seconds). + * @var uint16_t + */ + uint16_t _connection_timeout; + /** * Method that is called by AMQP-CPP to register a filedescriptor for readability or writability * @param connection The TCP connection object that is reporting * @param fd The filedescriptor to be monitored * @param flags Should the object be monitored for readability or writability? */ - void monitor(TcpConnection *const connection, - const int fd, - const int flags) override + void monitor(TcpConnection *connection, int fd, int flags) override { // do we already have this filedescriptor auto iter = _watchers.find(fd); @@ -440,73 +515,82 @@ class LibBoostAsioHandler : public virtual TcpHandler // was it found? if (iter == _watchers.end()) { - // we did not yet have this watcher - but that is ok if no filedescriptor was registered - if (flags == 0){ return; } + // a new watcher is required + assert(flags != 0); // a watcher should not be dead on arrival + + // construct a new watcher + const std::shared_ptr spwatcher = + std::make_shared(_iocontext, _strand, fd, _connection_timeout); - // construct a new pair (watcher/timer), and put it in the map - const std::shared_ptr apWatcher = - std::make_shared(_iocontext, _strand, fd); + // register as active + _watchers[fd] = spwatcher; - _watchers[fd] = apWatcher; + // apply server connection timeout monitor + spwatcher->apply_connection_timeout(connection, fd); // explicitly set the events to monitor - apWatcher->events(connection, fd, flags); + spwatcher->events(connection, fd, flags); } else if (flags == 0) { - // the watcher does already exist, but we no longer have to watch this watcher + // the watcher does not need anymore, unregister _watchers.erase(iter); + + // have to release the filedescriptor immediately, cannot rely on the + // dtor for that since must wait for all the callback to terminate + iter->second->close(); } else { - // Change the events on which to act. - iter->second->events(connection,fd,flags); + // reconfigure the events to monitor + iter->second->events(connection, fd, flags); } } protected: + /** - * Method that is called when the heartbeat frequency is negotiated between the server and the client. - * @param connection The connection that suggested a heartbeat interval - * @param interval The suggested interval from the server - * @return uint16_t The interval to use + * Method that is called when the heartbeat timeout is negotiated between the server and the client. + * @param connection The connection that suggested a heartbeat timeout + * @param timeout The suggested timeout from the server (seconds) + * @return uint16_t The timeout to use */ - virtual uint16_t onNegotiate(TcpConnection *connection, uint16_t interval) override + uint16_t onNegotiate(TcpConnection *connection, uint16_t timeout) override { // skip if no heartbeats are needed - if (interval == 0) return 0; + if (timeout == 0) return 0; - const auto fd = connection->fileno(); + const int fd = connection->fileno(); auto iter = _watchers.find(fd); - if (iter == _watchers.end()) return 0; + assert(iter != _watchers.end()); // a watcher must exist when negotiating the heartbeat - // set the timer - iter->second->set_timer(connection, interval); + // apply heartbeat monitor + iter->second->set_heartbeat(timeout); + iter->second->set_timer(connection, fd); - // we agree with the interval - return interval; + // we agree with the timeout + return timeout; } public: /** * Handler cannot be default constructed. - * - * @param that The object to not move or copy */ LibBoostAsioHandler() = delete; /** * Constructor - * @param io_context The boost io_context to wrap + * @param io_context The boost io_context to wrap + * @param connection_timeout The AMQP server connection timeout (seconds) */ - explicit LibBoostAsioHandler(boost::asio::io_context &io_context) : + explicit LibBoostAsioHandler(boost::asio::io_context &io_context, + uint16_t connection_timeout = 60) : _iocontext(io_context), - _strand(std::make_shared(_iocontext)) - //_timer(std::make_shared(_iocontext,_strand)) + _strand(std::make_shared(_iocontext)), + _connection_timeout(connection_timeout) { - } /**