diff options
Diffstat (limited to 'qpid/cpp')
| -rw-r--r-- | qpid/cpp/src/qpid/broker/DeliveryRecord.cpp | 24 | ||||
| -rw-r--r-- | qpid/cpp/src/qpid/broker/SemanticState.cpp | 7 | ||||
| -rw-r--r-- | qpid/cpp/src/tests/MessagingSessionTests.cpp | 41 |
3 files changed, 59 insertions, 13 deletions
diff --git a/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp b/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp index 9443eb6ea5..64760bea36 100644 --- a/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp +++ b/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp @@ -131,18 +131,20 @@ void DeliveryRecord::committed() const{ void DeliveryRecord::reject() { - Exchange::shared_ptr alternate = queue->getAlternateExchange(); - if (alternate) { - DeliverableMessage delivery(msg.payload); - alternate->route(delivery, msg.payload->getRoutingKey(), msg.payload->getApplicationHeaders()); - QPID_LOG(info, "Routed rejected message from " << queue->getName() << " to " - << alternate->getName()); - } else { - //just drop it - QPID_LOG(info, "Dropping rejected message from " << queue->getName()); + if (acquired && !ended) { + Exchange::shared_ptr alternate = queue->getAlternateExchange(); + if (alternate) { + DeliverableMessage delivery(msg.payload); + alternate->route(delivery, msg.payload->getRoutingKey(), msg.payload->getApplicationHeaders()); + QPID_LOG(info, "Routed rejected message from " << queue->getName() << " to " + << alternate->getName()); + } else { + //just drop it + QPID_LOG(info, "Dropping rejected message from " << queue->getName()); + } + dequeue(); + setEnded(); } - - dequeue(); } uint32_t DeliveryRecord::getCredit() const diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp index c91cfba2f8..cfc379f47c 100644 --- a/qpid/cpp/src/qpid/broker/SemanticState.cpp +++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp @@ -697,8 +697,11 @@ void SemanticState::reject(DeliveryId first, DeliveryId last) { AckRange range = findRange(first, last); for_each(range.start, range.end, mem_fun_ref(&DeliveryRecord::reject)); - //need to remove the delivery records as well - unacked.erase(range.start, range.end); + //may need to remove the delivery records as well + for (DeliveryRecords::iterator i = range.start; i != unacked.end() && i->getId() <= last; ) { + if (i->isRedundant()) i = unacked.erase(i); + else i++; + } } bool SemanticState::ConsumerImpl::doOutput() diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp index f9a8b0e4c1..20ed16dece 100644 --- a/qpid/cpp/src/tests/MessagingSessionTests.cpp +++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp @@ -937,6 +937,47 @@ QPID_AUTO_TEST_CASE(testQmfCreateAndDelete) } } +QPID_AUTO_TEST_CASE(testRejectAndCredit) +{ + //Ensure credit is restored on completing rejected messages + QueueFixture fix; + Sender sender = fix.session.createSender(fix.queue); + Receiver receiver = fix.session.createReceiver(fix.queue); + + const uint count(10); + receiver.setCapacity(count); + for (uint i = 0; i < count; i++) { + sender.send(Message((boost::format("Message_%1%") % (i+1)).str())); + } + + Message in; + for (uint i = 0; i < count; ++i) { + if (receiver.fetch(in, Duration::SECOND)) { + BOOST_CHECK_EQUAL(in.getContent(), (boost::format("Message_%1%") % (i+1)).str()); + fix.session.reject(in); + } else { + BOOST_FAIL((boost::format("Message_%1% not received as expected") % (i+1)).str()); + break; + } + } + //send another batch of messages + for (uint i = 0; i < count; i++) { + sender.send(Message((boost::format("Message_%1%") % (i+count)).str())); + } + + for (uint i = 0; i < count; ++i) { + if (receiver.fetch(in, Duration::SECOND)) { + BOOST_CHECK_EQUAL(in.getContent(), (boost::format("Message_%1%") % (i+count)).str()); + } else { + BOOST_FAIL((boost::format("Message_%1% not received as expected") % (i+count)).str()); + break; + } + } + fix.session.acknowledge(); + receiver.close(); + sender.close(); +} + QPID_AUTO_TEST_SUITE_END() }} // namespace qpid::tests |