From 40e17612ff0494ccd440a3a0a3789daabcfb3684 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Mon, 20 Oct 2008 13:58:23 +0000 Subject: cluster: DumpClient updates consumer notifyEnabled and blocked. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@706293 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/qpid/broker/SemanticState.cpp | 5 +++++ cpp/src/qpid/broker/SemanticState.h | 8 ++++++-- cpp/src/qpid/broker/SessionState.h | 1 + cpp/src/qpid/cluster/Connection.cpp | 8 ++++++++ cpp/src/qpid/cluster/Connection.h | 2 ++ cpp/src/qpid/cluster/DumpClient.cpp | 10 +++++++++- 6 files changed, 31 insertions(+), 3 deletions(-) (limited to 'cpp/src/qpid') diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp index 177157bbb6..4d735c9abc 100644 --- a/cpp/src/qpid/broker/SemanticState.cpp +++ b/cpp/src/qpid/broker/SemanticState.cpp @@ -616,6 +616,11 @@ void SemanticState::ConsumerImpl::disableNotify() notifyEnabled = false; } +bool SemanticState::ConsumerImpl::isNotifyEnabld() { + Mutex::ScopedLock l(lock); + return notifyEnabled; +} + void SemanticState::ConsumerImpl::notify() { //TODO: alter this, don't want to hold locks across external diff --git a/cpp/src/qpid/broker/SemanticState.h b/cpp/src/qpid/broker/SemanticState.h index 7b72921210..c2d8cc7d0b 100644 --- a/cpp/src/qpid/broker/SemanticState.h +++ b/cpp/src/qpid/broker/SemanticState.h @@ -100,6 +100,7 @@ class SemanticState : public sys::OutputTask, void disableNotify(); void enableNotify(); void notify(); + bool isNotifyEnabld(); void setWindowMode(); void setCreditMode(); @@ -109,7 +110,8 @@ class SemanticState : public sys::OutputTask, void stop(); void complete(DeliveryRecord&); Queue::shared_ptr getQueue() { return queue; } - bool isBlocked() const { return blocked; } + bool isBlocked() const { return blocked; } + bool setBlocked(bool set) { std::swap(set, blocked); return set; } bool hasOutput(); bool doOutput(); @@ -150,7 +152,7 @@ class SemanticState : public sys::OutputTask, void route(boost::intrusive_ptr msg, Deliverable& strategy); void record(const DeliveryRecord& delivery); void checkDtxTimeout(); - ConsumerImpl& find(const std::string& destination); + void complete(DeliveryRecord&); AckRange findRange(DeliveryId first, DeliveryId last); void requestDispatch(); @@ -162,6 +164,8 @@ class SemanticState : public sys::OutputTask, ~SemanticState(); SessionContext& getSession() { return session; } + + ConsumerImpl& find(const std::string& destination); /** * Get named queue, never returns 0. diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h index 66eba9d2a9..ba750f0cc6 100644 --- a/cpp/src/qpid/broker/SessionState.h +++ b/cpp/src/qpid/broker/SessionState.h @@ -101,6 +101,7 @@ class SessionState : public qpid::SessionState, void readyToSend(); template void eachConsumer(F f) { semanticState.eachConsumer(f); } + SemanticState::ConsumerImpl& getConsumer(const string& dest) { return semanticState.find(dest); } private: diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp index 2f1518f871..28391a5c78 100644 --- a/cpp/src/qpid/cluster/Connection.cpp +++ b/cpp/src/qpid/cluster/Connection.cpp @@ -22,6 +22,7 @@ #include "Cluster.h" #include "qpid/broker/SessionState.h" +#include "qpid/broker/SemanticState.h" #include "qpid/framing/AMQFrame.h" #include "qpid/framing/AllInvoker.h" #include "qpid/framing/ClusterConnectionDeliverCloseBody.h" @@ -159,6 +160,13 @@ void Connection::deliverBuffer(Buffer& buf) { delivered(mcastDecoder.frame); } +void Connection::consumerState(const string& name, bool blocked, bool notifyEnabled) { + broker::SessionHandler& h = connection.getChannel(currentChannel); + broker::SessionState* s = h.getSession(); + broker::SemanticState::ConsumerImpl& c = s->getConsumer(name); + c.setBlocked(blocked); + if (notifyEnabled) c.enableNotify(); else c.disableNotify(); +} void Connection::sessionState( const SequenceNumber& replayStart, diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h index b537470b41..3b5298a8a1 100644 --- a/cpp/src/qpid/cluster/Connection.h +++ b/cpp/src/qpid/cluster/Connection.h @@ -89,6 +89,8 @@ class Connection : void deliverBuffer(framing::Buffer&); void delivered(framing::AMQFrame&); + void consumerState(const std::string& name, bool blocked, bool notifyEnabled); + // ==== Used in catch-up mode to build initial state. // // State dump methods. diff --git a/cpp/src/qpid/cluster/DumpClient.cpp b/cpp/src/qpid/cluster/DumpClient.cpp index 58aa14655c..c262115f9f 100644 --- a/cpp/src/qpid/cluster/DumpClient.cpp +++ b/cpp/src/qpid/cluster/DumpClient.cpp @@ -35,6 +35,8 @@ #include "qpid/framing/MessageTransferBody.h" #include "qpid/framing/ClusterConnectionMembershipBody.h" #include "qpid/framing/ClusterConnectionShadowReadyBody.h" +#include "qpid/framing/ClusterConnectionSessionStateBody.h" +#include "qpid/framing/ClusterConnectionConsumerStateBody.h" #include "qpid/framing/enum.h" #include "qpid/framing/ProtocolVersion.h" #include "qpid/log/Statement.h" @@ -227,7 +229,13 @@ void DumpClient::dumpConsumer(broker::SemanticState::ConsumerImpl* ci) { shadowSession.messageSetFlowMode(ci->getName(), ci->isWindowing() ? FLOW_MODE_WINDOW : FLOW_MODE_CREDIT); shadowSession.messageFlow(ci->getName(), CREDIT_UNIT_MESSAGE, ci->getMsgCredit()); shadowSession.messageFlow(ci->getName(), CREDIT_UNIT_BYTE, ci->getByteCredit()); - // FIXME aconway 2008-09-23: need to replicate ConsumerImpl::blocked and notifyEnabled? + ClusterConnectionConsumerStateBody state( + ProtocolVersion(), + ci->getName(), + ci->isBlocked(), + ci->isNotifyEnabld() + ); + client::SessionBase_0_10Access(shadowSession).get()->send(state); QPID_LOG(debug, dumperId << " dumped consumer " << ci->getName() << " on " << shadowSession.getId()); } -- cgit v1.2.1