when a consumer is cancelled, it is also removed from the map of active consumers in the the ChannelImpl object
This commit is contained in:
parent
3b78247363
commit
ae7a32a8bf
|
|
@ -493,8 +493,6 @@ public:
|
|||
// we are going to call callbacks that could destruct the channel
|
||||
Monitor monitor(this);
|
||||
|
||||
// @todo should this be a std::string parameter?
|
||||
|
||||
// inform handler
|
||||
if (_errorCallback) _errorCallback(message);
|
||||
|
||||
|
|
@ -528,6 +526,16 @@ public:
|
|||
else _consumers.erase(consumertag);
|
||||
}
|
||||
|
||||
/**
|
||||
* Uninstall a consumer callback
|
||||
* @param consumertag The consumer tag
|
||||
*/
|
||||
void uninstall(const std::string &consumertag)
|
||||
{
|
||||
// erase the callback
|
||||
_consumers.erase(consumertag);
|
||||
}
|
||||
|
||||
/**
|
||||
* Report that a message was received
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -19,6 +19,12 @@ namespace AMQP {
|
|||
class DeferredCancel : public Deferred
|
||||
{
|
||||
private:
|
||||
/**
|
||||
* Pointer to the channel
|
||||
* @var ChannelImpl
|
||||
*/
|
||||
ChannelImpl *_channel;
|
||||
|
||||
/**
|
||||
* Callback to execute when the instruction is completed
|
||||
* @var CancelCallback
|
||||
|
|
@ -30,20 +36,7 @@ private:
|
|||
* @param name Consumer tag that is cancelled
|
||||
* @return Deferred
|
||||
*/
|
||||
virtual Deferred *reportSuccess(const std::string &name) const override
|
||||
{
|
||||
// skip if no special callback was installed
|
||||
if (!_cancelCallback) return Deferred::reportSuccess();
|
||||
|
||||
// call the callback
|
||||
_cancelCallback(name);
|
||||
|
||||
// call finalize callback
|
||||
if (_finalizeCallback) _finalizeCallback();
|
||||
|
||||
// return next object
|
||||
return _next;
|
||||
}
|
||||
virtual Deferred *reportSuccess(const std::string &name) const override;
|
||||
|
||||
/**
|
||||
* The channel implementation may call our
|
||||
|
|
@ -57,9 +50,11 @@ protected:
|
|||
* Protected constructor that can only be called
|
||||
* from within the channel implementation
|
||||
*
|
||||
* @param boolean are we already failed?
|
||||
* @param channel Pointer to the channel
|
||||
* @param failed Are we already failed?
|
||||
*/
|
||||
DeferredCancel(bool failed = false) : Deferred(failed) {}
|
||||
DeferredCancel(ChannelImpl *channel, bool failed = false) :
|
||||
Deferred(failed), _channel(channel) {}
|
||||
|
||||
public:
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -524,7 +524,7 @@ DeferredCancel &ChannelImpl::cancel(const std::string &tag, int flags)
|
|||
BasicCancelFrame frame(_id, tag, flags & nowait);
|
||||
|
||||
// send the frame, and create deferred object
|
||||
auto *deferred = new DeferredCancel(send(frame));
|
||||
auto *deferred = new DeferredCancel(this, send(frame));
|
||||
|
||||
// push to list
|
||||
push(deferred, "Cannot send basic cancel frame");
|
||||
|
|
|
|||
|
|
@ -0,0 +1,42 @@
|
|||
/**
|
||||
* DeferredCancel.cpp
|
||||
*
|
||||
* Implementation file for the DeferredCancel class
|
||||
*
|
||||
* @copyright 2014 Copernica BV
|
||||
*/
|
||||
#include "includes.h"
|
||||
|
||||
/**
|
||||
* Namespace
|
||||
*/
|
||||
namespace AMQP {
|
||||
|
||||
/**
|
||||
* Report success for frames that report cancel operations
|
||||
* @param name Consumer tag that is cancelled
|
||||
* @return Deferred
|
||||
*/
|
||||
Deferred *DeferredCancel::reportSuccess(const std::string &name) const
|
||||
{
|
||||
// in the channel, we should uninstall the consumer
|
||||
_channel->uninstall(name);
|
||||
|
||||
// skip if no special callback was installed
|
||||
if (!_cancelCallback) return Deferred::reportSuccess();
|
||||
|
||||
// call the callback
|
||||
_cancelCallback(name);
|
||||
|
||||
// call finalize callback
|
||||
if (_finalizeCallback) _finalizeCallback();
|
||||
|
||||
// return next object
|
||||
return _next;
|
||||
}
|
||||
|
||||
/**
|
||||
* End namespace
|
||||
*/
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue