From 3db445447d3e1060f9ac27dbc7578b9a227ae2ef Mon Sep 17 00:00:00 2001 From: Gavin Smith <26778249+zerodefect@users.noreply.github.com> Date: Sun, 7 Dec 2025 21:49:44 +0000 Subject: [PATCH 1/2] Refactor: Modernize LibBoostAsioHandler to C++17 syntax This commit updates the Boost Asio handler implementation to remove dependency on legacy Boost libraries in favor of modern C++17 standard features. Changes include: - Replaced all instances of `boost::bind` with native C++ lambdas. - Removed `boost/bind/bind.hpp` dependency. - Replaced `PTR_FROM_THIS` macro with standard `weak_from_this()`. - Standardized callback types to use `std::function` instead of `boost::function`. - Use `std::move()` semantics to reduce memory allocation(s). --- include/amqpcpp/libboostasio.h | 91 +++++++++++++++++----------------- 1 file changed, 46 insertions(+), 45 deletions(-) diff --git a/include/amqpcpp/libboostasio.h b/include/amqpcpp/libboostasio.h index 0df9011f..57001195 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 */ @@ -113,9 +106,9 @@ class LibBoostAsioHandler : public virtual TcpHandler */ bool _write_pending{false}; - 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,17 +117,25 @@ 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); + }); }; } @@ -146,14 +147,11 @@ class LibBoostAsioHandler : public virtual TcpHandler */ 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); + auto self = weak_from_this(); + auto fn = [this, self{std::move(self)}, connection, fd](const boost::system::error_code &ec, const std::size_t bytes) { + this->read_handler(ec, bytes, self, connection, fd); + }; + return get_dispatch_wrapper(std::move(fn)); } /** @@ -164,42 +162,45 @@ class LibBoostAsioHandler : public virtual TcpHandler */ 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); + auto self = weak_from_this(); + auto fn = [this, self{std::move(self)}, connection, fd](const boost::system::error_code &ec, const std::size_t bytes) { + this->write_handler(ec, bytes, self, connection, fd); + }; + 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); + }); }; } @@ -213,7 +214,7 @@ class LibBoostAsioHandler : public virtual TcpHandler * @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) @@ -247,7 +248,7 @@ class LibBoostAsioHandler : public virtual TcpHandler * @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) From 3e128544d25ea432abf4a2cf7e89b54f8283f738 Mon Sep 17 00:00:00 2001 From: Gavin Smith <26778249+zerodefect@users.noreply.github.com> Date: Tue, 9 Dec 2025 12:06:41 +0000 Subject: [PATCH 2/2] Refactor: Encapsulate file descriptor in Watcher class This commit moves the file descriptor (`fd`) into a member variable `_fd` within the `Watcher` class. Previously, `fd` was passed as an argument through every internal handler and captured in every lambda. Since the `Watcher` owns the socket, passing the `fd` explicitly was redundant. This change simplifies the function signatures and reduces the lambda capture overhead. --- include/amqpcpp/libboostasio.h | 43 +++++++++++++++++----------------- 1 file changed, 22 insertions(+), 21 deletions(-) diff --git a/include/amqpcpp/libboostasio.h b/include/amqpcpp/libboostasio.h index 57001195..da1eb1da 100644 --- a/include/amqpcpp/libboostasio.h +++ b/include/amqpcpp/libboostasio.h @@ -105,6 +105,12 @@ 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 = std::function; using io_handler = std::function; @@ -142,14 +148,13 @@ class LibBoostAsioHandler : public virtual TcpHandler /** * 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 self = weak_from_this(); - auto fn = [this, self{std::move(self)}, connection, fd](const boost::system::error_code &ec, const std::size_t bytes) { - this->read_handler(ec, bytes, self, connection, fd); + 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)); } @@ -157,14 +162,13 @@ class LibBoostAsioHandler : public virtual TcpHandler /** * 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 self = weak_from_this(); - auto fn = [this, self{std::move(self)}, connection, fd](const boost::system::error_code &ec, const std::size_t bytes) { - this->write_handler(ec, bytes, self, connection, fd); + 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)); } @@ -210,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*/, // 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). @@ -228,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)); } } @@ -244,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*/, // 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). @@ -262,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)); } } @@ -319,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); @@ -361,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? @@ -374,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)); } }