Skip to content

Refactor: Encapsulate file descriptor in LibBoostAsioHandler Watcher class (pt 3) #554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zerodefect wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Refactor: Encapsulate file descriptor in LibBoostAsioHandler Watcher class (pt 3) #554

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3db4454
Refactor: Modernize LibBoostAsioHandler to C++17 syntax
zerodefect Dec 7, 2025
3e12854
Refactor: Encapsulate file descriptor in Watcher class
zerodefect Dec 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 64 additions & 62 deletions include/amqpcpp/libboostasio.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,25 +23,18 @@
/**
* Dependencies
*/
#include <functional>
#include <map>
#include <memory>

#include <boost/asio/io_context.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/bind/bind.hpp>
#include <boost/function.hpp>

#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<T>(shared_from_this())
#endif

/**
* Set up namespace
*/
Expand Down Expand Up @@ -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<void(boost::system::error_code,std::size_t)>;
using io_handler = boost::function<void(const boost::system::error_code&, const std::size_t)>;
using timer_handler = boost::function<void(boost::system::error_code)>;
using handler_cb = std::function<void(boost::system::error_code, std::size_t)>;
using io_handler = std::function<void(const boost::system::error_code&, const std::size_t)>;
using timer_handler = std::function<void(boost::system::error_code)>;

/**
* Builds a io handler callback that executes the io callback in a strand.
Expand All @@ -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);
});
};
}

Expand All @@ -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<Watcher> awpWatcher,
TcpConnection *const connection,
const int fd)
TcpConnection *const connection)
{
// Resolve any potential problems with dangling pointers
// (remember we are using async).
Expand All @@ -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));
}
}

Expand All @@ -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<Watcher> awpWatcher,
TcpConnection *const connection,
const int fd)
TcpConnection *const connection)
{
// Resolve any potential problems with dangling pointers
// (remember we are using async).
Expand All @@ -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));
}
}

Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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?
Expand All @@ -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));
}
}

Expand Down