diff --git a/include/amqpcpp/libboostasio.h b/include/amqpcpp/libboostasio.h index 0df9011f..da1eb1da 100644 --- a/include/amqpcpp/libboostasio.h +++ b/include/amqpcpp/libboostasio.h @@ -23,6 +23,8 @@ /** * Dependencies */ +#include +#include #include #include @@ -30,18 +32,9 @@ #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 */ @@ -112,10 +105,16 @@ class LibBoostAsioHandler : public virtual TcpHandler * @var _read True if read is pending else false. */ bool _write_pending{false}; + + /** + * The socket descriptor that is being monitored. + * @var _fd The OS's underlying socket descriptor. + */ + int _fd; - using handler_cb = boost::function; - using io_handler = boost::function; - using timer_handler = boost::function; + using handler_cb = std::function; + using io_handler = std::function; + using timer_handler = std::function; /** * Builds a io handler callback that executes the io callback in a strand. @@ -124,82 +123,88 @@ class LibBoostAsioHandler : public virtual TcpHandler */ handler_cb get_dispatch_wrapper(io_handler fn) { + // Use weak_ptr to strand to prevent circular dependencies or leaks const strand_weak_ptr wpstrand = _wpstrand; - return [fn, wpstrand](const boost::system::error_code &ec, const std::size_t bytes_transferred) + return [fn{std::move(fn)}, wpstrand](const boost::system::error_code &ec, const std::size_t bytes_transferred) { - const strand_shared_ptr strand = wpstrand.lock(); + const auto strand = wpstrand.lock(); if (!strand) { - fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled), std::size_t{0}); + // If the strand is gone, the handler is shutting down. + // Execute the function with operation_canceled to allow cleanup if needed, + // or just return (depending on library semantics). + fn(boost::asio::error::operation_aborted, 0); return; } - boost::asio::dispatch(strand->context().get_executor(), boost::bind(fn, ec, bytes_transferred)); + + boost::asio::dispatch(strand->context().get_executor(), + [fn, ec, bytes_transferred]() { + fn(ec, bytes_transferred); + }); }; } /** * 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 */ - handler_cb get_read_handler(TcpConnection *const connection, const int fd) + handler_cb get_read_handler(TcpConnection *const connection) { - 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); + auto self = weak_from_this(); + auto fn = [this, self{std::move(self)}, connection](const boost::system::error_code &ec, const std::size_t bytes) { + this->read_handler(ec, bytes, self, connection); + }; + return get_dispatch_wrapper(std::move(fn)); } /** * 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 */ - handler_cb get_write_handler(TcpConnection *const connection, const int fd) + handler_cb get_write_handler(TcpConnection *const connection) { - 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); + auto self = weak_from_this(); + auto fn = [this, self{std::move(self)}, connection](const boost::system::error_code &ec, const std::size_t bytes) { + this->write_handler(ec, bytes, self, connection); + }; + return get_dispatch_wrapper(std::move(fn)); } /** - * Binds and returns a lamba function handler for the io operation. + * Binds and returns a lambda function handler for the io operation. * @param connection The connection being watched. * @param timeout The file descripter being watched. * @return handler callback */ timer_handler get_timer_handler(TcpConnection *const connection, const uint16_t timeout) { - const auto fn = boost::bind(&Watcher::timeout_handler, - this, - boost::placeholders::_1, - PTR_FROM_THIS(Watcher), - connection, - timeout); + auto self = weak_from_this(); + + // The actual logic that runs when timer fires + auto fn = [this, self{std::move(self)}, connection, timeout](const boost::system::error_code &ec) { + this->timeout_handler(ec, self, connection, timeout); + }; const strand_weak_ptr wpstrand = _wpstrand; - return [fn, wpstrand](const boost::system::error_code &ec) + // The dispatch wrapper + return [fn{std::move(fn)}, wpstrand](const boost::system::error_code &ec) { - const strand_shared_ptr strand = wpstrand.lock(); + auto strand = wpstrand.lock(); if (!strand) { - fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled)); + // If strand is dead, we cannot safely execute the callback return; } - boost::asio::dispatch(strand->context().get_executor(), boost::bind(fn, ec)); + + // Dispatch ensuring thread safety via strand + boost::asio::dispatch(strand->context().get_executor(), + [fn, ec]() { + fn(ec); + }); }; } @@ -209,14 +214,12 @@ class LibBoostAsioHandler : public virtual TcpHandler * @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::size_t /*bytes_transferred*/, // Stops: -Wunused-parameter const std::weak_ptr awpWatcher, - TcpConnection *const connection, - const int fd) + TcpConnection *const connection) { // Resolve any potential problems with dangling pointers // (remember we are using async). @@ -227,13 +230,13 @@ class LibBoostAsioHandler : public virtual TcpHandler if ((!ec || ec == boost::asio::error::would_block) && _read) { - connection->process(fd, AMQP::readable); + connection->process(_fd, AMQP::readable); _read_pending = true; _socket.async_read_some( boost::asio::null_buffers(), - get_read_handler(connection, fd)); + get_read_handler(connection)); } } @@ -243,14 +246,12 @@ class LibBoostAsioHandler : public virtual TcpHandler * @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::size_t /*bytes_transferred*/, // Stops: -Wunused-parameter const std::weak_ptr awpWatcher, - TcpConnection *const connection, - const int fd) + TcpConnection *const connection) { // Resolve any potential problems with dangling pointers // (remember we are using async). @@ -261,13 +262,13 @@ class LibBoostAsioHandler : public virtual TcpHandler if ((!ec || ec == boost::asio::error::would_block) && _write) { - connection->process(fd, AMQP::writable); + connection->process(_fd, AMQP::writable); _write_pending = true; _socket.async_write_some( boost::asio::null_buffers(), - get_write_handler(connection, fd)); + get_write_handler(connection)); } } @@ -318,7 +319,8 @@ class LibBoostAsioHandler : public virtual TcpHandler _iocontext(io_context), _wpstrand(wpstrand), _socket(io_context), - _timer(io_context) + _timer(io_context), + _fd(fd) { _socket.assign(fd); @@ -360,7 +362,7 @@ class LibBoostAsioHandler : public virtual TcpHandler _socket.async_read_some( boost::asio::null_buffers(), - get_read_handler(connection, fd)); + get_read_handler(connection)); } // 2. Handle writes? @@ -373,7 +375,7 @@ class LibBoostAsioHandler : public virtual TcpHandler _socket.async_write_some( boost::asio::null_buffers(), - get_write_handler(connection, fd)); + get_write_handler(connection)); } }