Author: gsim
Date: Fri Oct 5 10:28:48 2007
New Revision: 582353
URL: http://svn.apache.org/viewvc?rev=582353&view=rev
Log:
Don't recover messages for cancelled subscriptions.
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Message.h
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp?rev=582353&r1=582352&r2=582353&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp Fri Oct 5
10:28:48 2007
@@ -39,18 +39,21 @@
id(_id),
acquired(_acquired),
confirmed(_confirmed),
- pull(false)
+ pull(false),
+
cancelled(false)
{
}
DeliveryRecord::DeliveryRecord(const QueuedMessage& _msg,
Queue::shared_ptr _queue,
const DeliveryId _id) : msg(_msg),
- queue(_queue),
- id(_id),
- acquired(true),
-
confirmed(false),
- pull(true){}
+ queue(_queue),
+ id(_id),
+ acquired(true),
+ confirmed(false),
+ pull(true),
+ cancelled(false)
+{}
void DeliveryRecord::dequeue(TransactionContext* ctxt) const{
@@ -76,7 +79,7 @@
}
void DeliveryRecord::redeliver(SemanticState* const session) {
- if (!confirmed) {
+ if (!confirmed && !cancelled) {
if(pull){
//if message was originally sent as response to get, we must
requeue it
requeue();
@@ -147,6 +150,12 @@
} else {
QPID_LOG(info, "Message already acquired " << id.getValue());
}
+}
+
+void DeliveryRecord::cancel(const std::string& cancelledTag)
+{
+ if (tag == cancelledTag)
+ cancelled = true;
}
namespace qpid {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.h?rev=582353&r1=582352&r2=582353&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.h Fri Oct 5
10:28:48 2007
@@ -49,6 +49,7 @@
bool acquired;
const bool confirmed;
const bool pull;
+ bool cancelled;
public:
DeliveryRecord(const QueuedMessage& msg, Queue::shared_ptr queue, const
std::string tag, DeliveryToken::shared_ptr token,
@@ -63,6 +64,7 @@
void requeue() const;
void release();
void reject();
+ void cancel(const std::string& tag);
void redeliver(SemanticState* const);
void updateByteCredit(uint32_t& credit) const;
void addTo(Prefetch&) const;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=582353&r1=582352&r2=582353&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Fri Oct 5
10:28:48 2007
@@ -95,8 +95,14 @@
// consumers is a ptr_map so erase will delete the consumer
// which will call cancel.
ConsumerImplMap::iterator i = consumers.find(tag);
- if (i != consumers.end())
+ if (i != consumers.end()) {
consumers.erase(i);
+ //should cancel all unacked messages for this consumer so that
+ //they are not redelivered on recovery
+ Mutex::ScopedLock locker(deliveryLock);
+ for_each(unacked.begin(), unacked.end(),
boost::bind(mem_fun_ref(&DeliveryRecord::cancel), _1, tag));
+
+ }
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Message.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Message.h?rev=582353&r1=582352&r2=582353&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Message.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Message.h Fri Oct 5 10:28:48
2007
@@ -75,6 +75,11 @@
return method;
}
+ const framing::SequenceNumber& getId() const
+ {
+ return id;
+ }
+
private:
//method and id are only set for received messages:
const framing::MessageTransferBody method;
