Fix segfault when the handler is destructed from within a callback
This commit is contained in:
parent
25ce57818a
commit
89c2075a5f
|
|
@ -1,12 +1,12 @@
|
|||
/**
|
||||
* LibEV.h
|
||||
*
|
||||
*
|
||||
* Implementation for the AMQP::TcpHandler that is optimized for libev. You can
|
||||
* use this class instead of a AMQP::TcpHandler class, just pass the event loop
|
||||
* to the constructor and you're all set
|
||||
*
|
||||
*
|
||||
* Compile with: "g++ -std=c++11 libev.cpp -lamqpcpp -lev -lpthread"
|
||||
*
|
||||
*
|
||||
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
|
||||
* @copyright 2015 Copernica BV
|
||||
*/
|
||||
|
|
@ -43,7 +43,7 @@ private:
|
|||
* @var struct ev_loop
|
||||
*/
|
||||
struct ev_loop *_loop;
|
||||
|
||||
|
||||
/**
|
||||
* The actual watcher structure
|
||||
* @var struct ev_io
|
||||
|
|
@ -60,11 +60,11 @@ private:
|
|||
{
|
||||
// retrieve the connection
|
||||
TcpConnection *connection = static_cast<TcpConnection*>(watcher->data);
|
||||
|
||||
// tell the connection that its filedescriptor is active
|
||||
|
||||
// tell the connection that its filedescriptor is active
|
||||
connection->process(watcher->fd, revents);
|
||||
}
|
||||
|
||||
|
||||
public:
|
||||
/**
|
||||
* Constructor
|
||||
|
|
@ -77,14 +77,22 @@ private:
|
|||
{
|
||||
// initialize the libev structure
|
||||
ev_io_init(&_io, callback, fd, events);
|
||||
|
||||
|
||||
// store the connection in the data "void*"
|
||||
_io.data = connection;
|
||||
|
||||
|
||||
// start the watcher
|
||||
ev_io_start(_loop, &_io);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Watchers cannot be copied or moved
|
||||
*
|
||||
* @param that The object to not move or copy
|
||||
*/
|
||||
Watcher(Watcher &&that) = delete;
|
||||
Watcher(const Watcher &that) = delete;
|
||||
|
||||
/**
|
||||
* Destructor
|
||||
*/
|
||||
|
|
@ -93,7 +101,7 @@ private:
|
|||
// stop the watcher
|
||||
ev_io_stop(_loop, &_io);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Change the events for which the filedescriptor is monitored
|
||||
* @param events
|
||||
|
|
@ -102,10 +110,10 @@ private:
|
|||
{
|
||||
// stop the watcher if it was active
|
||||
ev_io_stop(_loop, &_io);
|
||||
|
||||
|
||||
// set the events
|
||||
ev_io_set(&_io, _io.fd, events);
|
||||
|
||||
|
||||
// and restart it
|
||||
ev_io_start(_loop, &_io);
|
||||
}
|
||||
|
|
@ -117,14 +125,14 @@ private:
|
|||
* @var struct ev_loop*
|
||||
*/
|
||||
struct ev_loop *_loop;
|
||||
|
||||
|
||||
/**
|
||||
* All I/O watchers that are active, indexed by their filedescriptor
|
||||
* @var std::map<int,Watcher>
|
||||
*/
|
||||
std::map<int,std::unique_ptr<Watcher>> _watchers;
|
||||
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Method that is called by AMQP-CPP to register a filedescriptor for readability or writability
|
||||
* @param connection The TCP connection object that is reporting
|
||||
|
|
@ -135,13 +143,13 @@ private:
|
|||
{
|
||||
// do we already have this filedescriptor
|
||||
auto iter = _watchers.find(fd);
|
||||
|
||||
|
||||
// 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;
|
||||
|
||||
|
||||
// construct a new watcher, and put it in the map
|
||||
_watchers[fd] = std::unique_ptr<Watcher>(new Watcher(_loop, connection, fd, flags));
|
||||
}
|
||||
|
|
@ -163,13 +171,13 @@ public:
|
|||
* @param loop The event loop to wrap
|
||||
*/
|
||||
LibEvHandler(struct ev_loop *loop) : _loop(loop) {}
|
||||
|
||||
|
||||
/**
|
||||
* Destructor
|
||||
*/
|
||||
virtual ~LibEvHandler() = default;
|
||||
};
|
||||
|
||||
|
||||
/**
|
||||
* End of namespace
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -26,7 +26,9 @@ class TcpState;
|
|||
/**
|
||||
* Class definition
|
||||
*/
|
||||
class TcpConnection : private ConnectionHandler
|
||||
class TcpConnection :
|
||||
private ConnectionHandler,
|
||||
private Watchable
|
||||
{
|
||||
private:
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -47,17 +47,23 @@ TcpConnection::~TcpConnection()
|
|||
*/
|
||||
void TcpConnection::process(int fd, int flags)
|
||||
{
|
||||
// monitor the object for destruction
|
||||
Monitor monitor{ this };
|
||||
|
||||
// pass on the the state, that returns a new impl
|
||||
auto *result = _state->process(fd, flags);
|
||||
|
||||
|
||||
// are we still valid
|
||||
if (!monitor.valid()) return;
|
||||
|
||||
// skip if the same state is continued to be used, or when the process()
|
||||
// method returns nullptr (which only happens when the object is destructed,
|
||||
// and "this" is no longer valid)
|
||||
if (!result || result == _state) return;
|
||||
|
||||
|
||||
// remove old state
|
||||
delete _state;
|
||||
|
||||
|
||||
// replace it with the new implementation
|
||||
_state = result;
|
||||
}
|
||||
|
|
@ -105,10 +111,10 @@ void TcpConnection::onError(Connection *connection, const char *message)
|
|||
{
|
||||
// current object is going to be removed, wrap it in a unique pointer to enforce that
|
||||
std::unique_ptr<TcpState> ptr(_state);
|
||||
|
||||
|
||||
// object is now in a closed state
|
||||
_state = new TcpClosed(_state);
|
||||
|
||||
|
||||
// tell the implementation to report the error
|
||||
ptr->reportError(message);
|
||||
}
|
||||
|
|
@ -131,10 +137,10 @@ void TcpConnection::onClosed(Connection *connection)
|
|||
{
|
||||
// current object is going to be removed, wrap it in a unique pointer to enforce that
|
||||
std::unique_ptr<TcpState> ptr(_state);
|
||||
|
||||
|
||||
// object is now in a closed state
|
||||
_state = new TcpClosed(_state);
|
||||
|
||||
|
||||
// tell the implementation to report the error
|
||||
ptr->reportClosed();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -177,7 +177,7 @@ public:
|
|||
{
|
||||
// only works if the incoming pipe is readable
|
||||
if (fd != _pipe.in() || !(flags & readable)) return this;
|
||||
|
||||
|
||||
// do we have a valid socket?
|
||||
if (_socket >= 0) return new TcpConnected(_connection, _socket, std::move(_buffer), _handler);
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue