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

bugfix; base handler was no longer (always) properly called

This commit is contained in:
Michael van der Werve 2020-10-14 11:21:02 +02:00
parent 8f85ad7818
commit b368f9e635
1 changed files with 39 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
src/confirmed.cpp
View File

@ -36,7 +36,7 @@ void Confirmed::onAck(uint64_t deliveryTag, bool multiple)
// we did not find it (this should not be possible, unless somebody explicitly called) // we did not find it (this should not be possible, unless somebody explicitly called)
// the base-class publish methods for some reason. // the base-class publish methods for some reason.
if (iter == _handlers.end()) return; if (iter == _handlers.end()) return Throttle::onAck(deliveryTag, multiple);
// call the ack handler // call the ack handler
iter->second->reportAck(); iter->second->reportAck();
@ -46,26 +46,30 @@ void Confirmed::onAck(uint64_t deliveryTag, bool multiple)
// erase it from the map // erase it from the map
_handlers.erase(iter); _handlers.erase(iter);
// we are done
return;
} }
// find the last element, inclusive // do multiple at once
auto upper = _handlers.upper_bound(deliveryTag); else
// call the handlers
for (auto iter = _handlers.begin(); iter != upper; iter++)
{ {
// call the handler // find the last element, inclusive
iter->second->reportAck(); auto upper = _handlers.upper_bound(deliveryTag);
// if we were destructed in the meantime, we leap out // call the handlers
if (!monitor) return; for (auto iter = _handlers.begin(); iter != upper; iter++)
{
// call the handler
iter->second->reportAck();
// if we were destructed in the meantime, we leap out
if (!monitor) return;
}
// erase all acknowledged items
_handlers.erase(_handlers.begin(), upper);
} }
// erase all acknowledged items // make sure the object is still valid
_handlers.erase(_handlers.begin(), upper); if (!monitor) return;
// call base handler, will advance on the throttle if needed. we call this _after_ we're // call base handler, will advance on the throttle if needed. we call this _after_ we're
// done processing the callbacks, since one of the callbacks might close the channel, or publish // done processing the callbacks, since one of the callbacks might close the channel, or publish
@ -93,7 +97,7 @@ void Confirmed::onNack(uint64_t deliveryTag, bool multiple)
// we did not find it (this should not be possible, unless somebody explicitly called) // we did not find it (this should not be possible, unless somebody explicitly called)
// the base-class publish methods for some reason. // the base-class publish methods for some reason.
if (iter == _handlers.end()) return; if (iter == _handlers.end()) return Throttle::onNack(deliveryTag, multiple);
// call the ack handler // call the ack handler
iter->second->reportNack(); iter->second->reportNack();
@ -103,27 +107,31 @@ void Confirmed::onNack(uint64_t deliveryTag, bool multiple)
// erase it from the map // erase it from the map
_handlers.erase(iter); _handlers.erase(iter);
// we are done
return;
} }
// find the last element, inclusive // nack multiple elements
auto upper = _handlers.upper_bound(deliveryTag); else
// call the handlers
for (auto iter = _handlers.begin(); iter != upper; iter++)
{ {
// call the handler // find the last element, inclusive
iter->second->reportNack(); auto upper = _handlers.upper_bound(deliveryTag);
// if we were destructed in the meantime, we leap out // call the handlers
if (!monitor) return; for (auto iter = _handlers.begin(); iter != upper; iter++)
{
// call the handler
iter->second->reportNack();
// if we were destructed in the meantime, we leap out
if (!monitor) return;
}
// erase all acknowledged items
_handlers.erase(_handlers.begin(), upper);
} }
// erase all acknowledged items // make sure the object is still valid
_handlers.erase(_handlers.begin(), upper); if (!monitor) return;
// call base handler, will advance on the throttle if needed. we call this _after_ we're // call base handler, will advance on the throttle if needed. we call this _after_ we're
// done processing the callbacks, since one of the callbacks might close the channel, or publish // done processing the callbacks, since one of the callbacks might close the channel, or publish
// more stuff. additionally, if it does destroy the channel, we are doing a lot of extra publishing // more stuff. additionally, if it does destroy the channel, we are doing a lot of extra publishing