jessequ
/
AMQP-CPP
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #148 from zerodefect/add_strand_to_boost_handler 

Add strand to boost handler
This commit is contained in:
Emiel Bruijntjes 2017-10-23 12:45:25 +02:00 committed by GitHub
parent 9559dfb880 034e72bbc2
commit 228daeee08
2 changed files with 215 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

210
include/libboostasio.h
View File

@ -3,9 +3,11 @@
* *
* Implementation for the AMQP::TcpHandler that is optimized for boost::asio. You can * Implementation for the AMQP::TcpHandler that is optimized for boost::asio. You can
* use this class instead of a AMQP::TcpHandler class, just pass the boost asio service * use this class instead of a AMQP::TcpHandler class, just pass the boost asio service
* to the constructor and you're all set * to the constructor and you're all set. See tests/libboostasio.cpp for example.
* *
* @author Gavin Smith <gavin.smith@coralbay.tv> * @author Gavin Smith <gavin.smith@coralbay.tv>
*
*
*/ */
/** /**
@ -16,11 +18,44 @@
/** /**
* Dependencies * Dependencies
*/ */
#include <memory>
#include <boost/asio/io_service.hpp> #include <boost/asio/io_service.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/posix/stream_descriptor.hpp> #include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/bind.hpp> #include <boost/bind.hpp>
///////////////////////////////////////////////////////////////////
#define STRAND_SOCKET_HANDLER(_fn) \
[fn = _fn, strand = _strand](const boost::system::error_code &ec, \
const std::size_t bytes_transferred) \
{ \
const std::shared_ptr<boost::asio::io_service::strand> apStrand = strand.lock(); \
if (!apStrand) \
{ \
fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled),std::size_t{0}); \
return; \
} \
\
apStrand->dispatch(boost::bind(fn,ec,bytes_transferred)); \
}
///////////////////////////////////////////////////////////////////
#define STRAND_TIMER_HANDLER(_fn) \
[fn = _fn, strand = _strand](const boost::system::error_code &ec) \
{ \
const std::shared_ptr<boost::asio::io_service::strand> apStrand = strand.lock(); \
if (!apStrand) \
{ \
fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled)); \
return; \
} \
\
apStrand->dispatch(boost::bind(fn,ec)); \
}
/** /**
* Set up namespace * Set up namespace
*/ */
@ -47,6 +82,12 @@ private:
*/ */
boost::asio::io_service & _ioservice; boost::asio::io_service & _ioservice;
/**
* The boost asio io_service::strand managed pointer.
* @var class std::shared_ptr<boost::asio::io_service>
*/
std::weak_ptr<boost::asio::io_service::strand> _strand;
/** /**
* The boost tcp socket. * The boost tcp socket.
* @var class boost::asio::ip::tcp::socket * @var class boost::asio::ip::tcp::socket
@ -82,15 +123,17 @@ private:
/** /**
* Handler method that is called by boost's io_service when the socket pumps a read event. * Handler method that is called by boost's io_service when the socket pumps a read event.
* @param ec The status of the callback. * @param ec The status of the callback.
* @param bytes_transferred The number of bytes transferred.
* @param awpWatcher A weak pointer to this object. * @param awpWatcher A weak pointer to this object.
* @param connection The connection being watched. * @param connection The connection being watched.
* @param fd The file descriptor being watched. * @param fd The file descriptor being watched.
* @note The handler will get called if a read is cancelled. * @note The handler will get called if a read is cancelled.
*/ */
void read_handler(boost::system::error_code ec, void read_handler(const boost::system::error_code &ec,
std::weak_ptr<Watcher> awpWatcher, const std::size_t bytes_transferred,
TcpConnection *connection, const std::weak_ptr<Watcher> awpWatcher,
int fd) TcpConnection *const connection,
const int fd)
{ {
// Resolve any potential problems with dangling pointers // Resolve any potential problems with dangling pointers
// (remember we are using async). // (remember we are using async).
@ -106,32 +149,36 @@ private:
_read_pending = true; _read_pending = true;
_socket.async_read_some(boost::asio::null_buffers(), _socket.async_read_some(boost::asio::null_buffers(),
boost::bind(&Watcher::read_handler, STRAND_SOCKET_HANDLER(
this, boost::bind(&Watcher::read_handler,
boost::placeholders::_1, this,
boost::placeholders::_1,
boost::placeholders::_2,
// C++17 has 'weak_from_this()' support. // C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L #if __cplusplus >= 201701L
weak_from_this() weak_from_this()
#else #else
shared_from_this(), shared_from_this(),
#endif #endif
connection, connection,
fd)); fd)));
} }
} }
/** /**
* Handler method that is called by boost's io_service when the socket pumps a write event. * Handler method that is called by boost's io_service when the socket pumps a write event.
* @param ec The status of the callback. * @param ec The status of the callback.
* @param bytes_transferred The number of bytes transferred.
* @param awpWatcher A weak pointer to this object. * @param awpWatcher A weak pointer to this object.
* @param connection The connection being watched. * @param connection The connection being watched.
* @param fd The file descriptor being watched. * @param fd The file descriptor being watched.
* @note The handler will get called if a write is cancelled. * @note The handler will get called if a write is cancelled.
*/ */
void write_handler(boost::system::error_code ec, void write_handler(const boost::system::error_code ec,
std::weak_ptr<Watcher> awpWatcher, const std::size_t bytes_transferred,
TcpConnection *connection, const std::weak_ptr<Watcher> awpWatcher,
int fd) TcpConnection *const connection,
const int fd)
{ {
// Resolve any potential problems with dangling pointers // Resolve any potential problems with dangling pointers
// (remember we are using async). // (remember we are using async).
@ -147,17 +194,19 @@ private:
_write_pending = true; _write_pending = true;
_socket.async_write_some(boost::asio::null_buffers(), _socket.async_write_some(boost::asio::null_buffers(),
boost::bind(&Watcher::write_handler, STRAND_SOCKET_HANDLER(
this, boost::bind(&Watcher::write_handler,
boost::placeholders::_1, this,
boost::placeholders::_1,
boost::placeholders::_2,
// C++17 has 'weak_from_this()' support. // C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L #if __cplusplus >= 201701L
weak_from_this() weak_from_this()
#else #else
shared_from_this(), shared_from_this(),
#endif #endif
connection, connection,
fd)); fd)));
} }
} }
@ -166,16 +215,19 @@ private:
* Constructor- initialises the watcher and assigns the filedescriptor to * Constructor- initialises the watcher and assigns the filedescriptor to
* a boost socket for monitoring. * a boost socket for monitoring.
* @param io_service The boost io_service * @param io_service The boost io_service
* @param strand A weak pointer to a io_service::strand instance.
* @param fd The filedescriptor being watched * @param fd The filedescriptor being watched
*/ */
Watcher(boost::asio::io_service &io_service, int fd) : Watcher(boost::asio::io_service &io_service,
const std::weak_ptr<boost::asio::io_service::strand> strand,
const int fd) :
_ioservice(io_service), _ioservice(io_service),
_strand(strand),
_socket(_ioservice) _socket(_ioservice)
{ {
_socket.assign(fd); _socket.assign(fd);
_socket.non_blocking(true); _socket.non_blocking(true);
} }
/** /**
@ -205,50 +257,56 @@ private:
// 1. Handle reads? // 1. Handle reads?
_read = ((events & AMQP::readable) != 0); _read = ((events & AMQP::readable) != 0);
// Read requsted but no read pending?
if (_read && !_read_pending) if (_read && !_read_pending)
{ {
_read_pending = true; _read_pending = true;
_socket.async_read_some(boost::asio::null_buffers(), _socket.async_read_some(boost::asio::null_buffers(),
boost::bind(&Watcher::read_handler, STRAND_SOCKET_HANDLER(
this, boost::bind(&Watcher::read_handler,
boost::placeholders::_1, this,
boost::placeholders::_1,
boost::placeholders::_2,
// C++17 has 'weak_from_this()' support. // C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L #if __cplusplus >= 201701L
weak_from_this() weak_from_this()
#else #else
shared_from_this(), shared_from_this(),
#endif #endif
connection, connection,
fd)); fd)));
} }
// 2. Handle writes? // 2. Handle writes?
_write = ((events & AMQP::writable) != 0); _write = ((events & AMQP::writable) != 0);
// Write requested but no write pending?
if (_write && !_write_pending) if (_write && !_write_pending)
{ {
_write_pending = true; _write_pending = true;
_socket.async_write_some(boost::asio::null_buffers(), _socket.async_write_some(boost::asio::null_buffers(),
boost::bind(&Watcher::write_handler, STRAND_SOCKET_HANDLER(
this, boost::bind(&Watcher::write_handler,
boost::placeholders::_1, this,
boost::placeholders::_1,
boost::placeholders::_2,
// C++17 has 'weak_from_this()' support. // C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L #if __cplusplus >= 201701L
weak_from_this() weak_from_this()
#else #else
shared_from_this(), shared_from_this(),
#endif #endif
connection, connection,
fd)); fd)));
} }
} }
}; };
/** /**
* Timer class to periodically fire a heartbeat * Timer class to periodically fire a heartbeat
*/ */
class Timer : public std::enable_shared_from_this<Timer> class Timer : public std::enable_shared_from_this<Timer>
{ {
private: private:
@ -259,6 +317,12 @@ private:
*/ */
boost::asio::io_service & _ioservice; boost::asio::io_service & _ioservice;
/**
* The boost asio io_service::strand managed pointer.
* @var class std::shared_ptr<boost::asio::io_service>
*/
std::weak_ptr<boost::asio::io_service::strand> _strand;
/** /**
* The boost asynchronous deadline timer. * The boost asynchronous deadline timer.
* @var class boost::asio::deadline_timer * @var class boost::asio::deadline_timer
@ -273,8 +337,8 @@ private:
*/ */
void timeout(const boost::system::error_code &ec, void timeout(const boost::system::error_code &ec,
std::weak_ptr<Timer> awpThis, std::weak_ptr<Timer> awpThis,
TcpConnection *connection, TcpConnection *const connection,
uint16_t timeout) const uint16_t timeout)
{ {
// Resolve any potential problems with dangling pointers // Resolve any potential problems with dangling pointers
// (remember we are using async). // (remember we are using async).
@ -289,11 +353,22 @@ private:
connection->heartbeat(); connection->heartbeat();
} }
// Reschedule the timer for 1 second in the future: // Reschedule the timer for the future:
_timer.expires_at(_timer.expires_at() + boost::posix_time::seconds(timeout)); _timer.expires_at(_timer.expires_at() + boost::posix_time::seconds(timeout));
// Posts the timer event // Posts the timer event
_timer.async_wait(boost::bind(&Timer::timeout,this,boost::placeholders::_1, shared_from_this(), connection, timeout)); _timer.async_wait(STRAND_TIMER_HANDLER(
boost::bind(&Timer::timeout,
this,
boost::placeholders::_1,
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this()
#else
shared_from_this(),
#endif
connection,
timeout)));
} }
} }
@ -309,10 +384,13 @@ private:
public: public:
/** /**
* Constructor * Constructor
* @param loop The current event loop * @param io_service The boost asio io_service.
* @param strand A weak pointer to a io_service::strand instance.
*/ */
Timer(boost::asio::io_service &io_service) : Timer(boost::asio::io_service &io_service,
const std::weak_ptr<boost::asio::io_service::strand> strand) :
_ioservice(io_service), _ioservice(io_service),
_strand(strand),
_timer(_ioservice) _timer(_ioservice)
{ {
@ -345,9 +423,19 @@ private:
// stop timer in case it was already set // stop timer in case it was already set
stop(); stop();
_timer.expires_from_now(boost::posix_time::seconds(timeout)); _timer.expires_from_now(boost::posix_time::seconds(timeout));
_timer.async_wait(boost::bind(&Timer::timeout,this,boost::placeholders::_1, shared_from_this(),connection, timeout)); _timer.async_wait(STRAND_TIMER_HANDLER(
boost::bind(&Timer::timeout,
this,
boost::placeholders::_1,
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this()
#else
shared_from_this(),
#endif
connection,
timeout)));
} }
}; };
@ -357,6 +445,12 @@ private:
*/ */
boost::asio::io_service & _ioservice; boost::asio::io_service & _ioservice;
/**
* The boost asio io_service::strand managed pointer.
* @var class std::shared_ptr<boost::asio::io_service>
*/
std::shared_ptr<boost::asio::io_service::strand> _strand;
/** /**
* All I/O watchers that are active, indexed by their filedescriptor * All I/O watchers that are active, indexed by their filedescriptor
@ -374,7 +468,9 @@ private:
* @param fd The filedescriptor to be monitored * @param fd The filedescriptor to be monitored
* @param flags Should the object be monitored for readability or writability? * @param flags Should the object be monitored for readability or writability?
*/ */
void monitor(TcpConnection *connection, int fd, int flags) override void monitor(TcpConnection *const connection,
const int fd,
const int flags) override
{ {
// do we already have this filedescriptor // do we already have this filedescriptor
auto iter = _watchers.find(fd); auto iter = _watchers.find(fd);
@ -386,8 +482,13 @@ private:
if (flags == 0){ return; } if (flags == 0){ return; }
// construct a new pair (watcher/timer), and put it in the map // construct a new pair (watcher/timer), and put it in the map
_watchers[fd] = std::make_shared<Watcher>(_ioservice, fd); const std::shared_ptr<Watcher> apWatcher =
_watchers[fd]->events(connection, fd, flags); std::make_shared<Watcher>(_ioservice, _strand, fd);
_watchers[fd] = apWatcher;
// explicitly set the events to monitor
apWatcher->events(connection, fd, flags);
} }
else if (flags == 0) else if (flags == 0)
{ {
@ -431,11 +532,12 @@ public:
/** /**
* Constructor * Constructor
* @param loop The boost io_service to wrap * @param io_service The boost io_service to wrap
*/ */
explicit LibBoostAsioHandler(boost::asio::io_service &io_service) : explicit LibBoostAsioHandler(boost::asio::io_service &io_service) :
_ioservice(io_service), _ioservice(io_service),
_timer(std::make_shared<Timer>(_ioservice)) _strand(std::make_shared<boost::asio::io_service::strand>(_ioservice)),
_timer(std::make_shared<Timer>(_ioservice,_strand))
{ {
} }

59
tests/libboostasio.cpp Normal file
View File

@ -0,0 +1,59 @@
/**
* LibBoostAsio.cpp
*
* Test program to check AMQP functionality based on Boost's asio io_service.
*
* @author Gavin Smith <gavin.smith@coralbay.tv>
*
* Compile with g++ libboostasio.cpp -o boost_test -lpthread -lboost_system -lamqpcpp
*/
/**
* Dependencies
*/
#include <boost/asio/io_service.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <amqpcpp.h>
#include <amqpcpp/libboostasio.h>
/**
* Main program
* @return int
*/
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);
// create a work object to process our events.
boost::asio::io_service::work work(service);
// 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);
// 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();
});
// 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();
}