diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/qpid/broker/DeliveryRecord.cpp | 6 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/SemanticState.cpp | 2 |
2 files changed, 6 insertions, 2 deletions
diff --git a/cpp/src/qpid/broker/DeliveryRecord.cpp b/cpp/src/qpid/broker/DeliveryRecord.cpp index 0cfa2391fd..22ec5e86a0 100644 --- a/cpp/src/qpid/broker/DeliveryRecord.cpp +++ b/cpp/src/qpid/broker/DeliveryRecord.cpp @@ -43,7 +43,7 @@ DeliveryRecord::DeliveryRecord(const QueuedMessage& _msg, acceptExpected(!accepted), cancelled(false), completed(false), - ended(accepted), + ended(accepted && acquired), windowing(_windowing), credit(msg.payload ? msg.payload->getRequiredCredit() : _credit) {} @@ -150,6 +150,10 @@ void DeliveryRecord::acquire(DeliveryIds& results) { if (queue->acquire(msg)) { acquired = true; results.push_back(id); + if (!acceptExpected) { + if (ended) { QPID_LOG(error, "Can't dequeue ended message"); } + else { queue->dequeue(0, msg); setEnded(); } + } } else { QPID_LOG(info, "Message already acquired " << id.getValue()); } diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp index 9680ada936..4502ff9f32 100644 --- a/cpp/src/qpid/broker/SemanticState.cpp +++ b/cpp/src/qpid/broker/SemanticState.cpp @@ -311,7 +311,7 @@ bool SemanticState::ConsumerImpl::deliver(QueuedMessage& msg) bool sync = syncFrequency && ++deliveryCount >= syncFrequency; if (sync) deliveryCount = 0;//reset parent->deliver(record, sync); - if (!ackExpected) record.setEnded();//allows message to be released now its been delivered + if (!ackExpected && acquire) record.setEnded();//allows message to be released now its been delivered if (windowing || ackExpected || !acquire) { parent->record(record); } |