diff options
Diffstat (limited to 'cpp/src/qpid')
| -rw-r--r-- | cpp/src/qpid/broker/IncompleteMessageList.cpp | 6 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/IncompleteMessageList.h | 3 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/SessionState.h | 4 | ||||
| -rw-r--r-- | cpp/src/qpid/client/SessionBase_0_10.cpp | 2 | ||||
| -rw-r--r-- | cpp/src/qpid/client/SessionBase_0_10.h | 3 | ||||
| -rw-r--r-- | cpp/src/qpid/cluster/ClusterPlugin.cpp | 2 | ||||
| -rw-r--r-- | cpp/src/qpid/cluster/DumpClient.cpp | 26 |
7 files changed, 33 insertions, 13 deletions
diff --git a/cpp/src/qpid/broker/IncompleteMessageList.cpp b/cpp/src/qpid/broker/IncompleteMessageList.cpp index dd7bbfc067..edb3721a40 100644 --- a/cpp/src/qpid/broker/IncompleteMessageList.cpp +++ b/cpp/src/qpid/broker/IncompleteMessageList.cpp @@ -30,7 +30,7 @@ void IncompleteMessageList::add(boost::intrusive_ptr<Message> msg) incomplete.push_back(msg); } -void IncompleteMessageList::process(CompletionListener l, bool sync) +void IncompleteMessageList::process(const CompletionListener& l, bool sync) { while (!incomplete.empty()) { boost::intrusive_ptr<Message>& msg = incomplete.front(); @@ -48,4 +48,8 @@ void IncompleteMessageList::process(CompletionListener l, bool sync) } } +void IncompleteMessageList::each(const CompletionListener& l) { + std::for_each(incomplete.begin(), incomplete.end(), l); +} + }} diff --git a/cpp/src/qpid/broker/IncompleteMessageList.h b/cpp/src/qpid/broker/IncompleteMessageList.h index 2cfd7bfee5..36cc1b4bf5 100644 --- a/cpp/src/qpid/broker/IncompleteMessageList.h +++ b/cpp/src/qpid/broker/IncompleteMessageList.h @@ -39,7 +39,8 @@ public: typedef boost::function<void(boost::intrusive_ptr<Message>)> CompletionListener; void add(boost::intrusive_ptr<Message> msg); - void process(CompletionListener l, bool sync); + void process(const CompletionListener& l, bool sync); + void each(const CompletionListener& l); }; diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h index ba750f0cc6..4c5a7a66b7 100644 --- a/cpp/src/qpid/broker/SessionState.h +++ b/cpp/src/qpid/broker/SessionState.h @@ -102,7 +102,9 @@ class SessionState : public qpid::SessionState, template <class F> void eachConsumer(F f) { semanticState.eachConsumer(f); } SemanticState::ConsumerImpl& getConsumer(const string& dest) { return semanticState.find(dest); } - + + boost::intrusive_ptr<Message> getMessageInProgress() { return msgBuilder.getMessage(); } + private: void handleCommand(framing::AMQMethodBody* method, const framing::SequenceNumber& id); diff --git a/cpp/src/qpid/client/SessionBase_0_10.cpp b/cpp/src/qpid/client/SessionBase_0_10.cpp index 50cfb4b09d..701acaf7d4 100644 --- a/cpp/src/qpid/client/SessionBase_0_10.cpp +++ b/cpp/src/qpid/client/SessionBase_0_10.cpp @@ -59,6 +59,8 @@ void SessionBase_0_10::sendCompletion() impl->sendCompletion(); } +uint16_t SessionBase_0_10::getChannel() const { return impl->getChannel(); } + void SessionBase_0_10::suspend() { impl->suspend(); } void SessionBase_0_10::resume(Connection c) { impl->resume(c.impl); } uint32_t SessionBase_0_10::timeout(uint32_t seconds) { return impl->setTimeout(seconds); } diff --git a/cpp/src/qpid/client/SessionBase_0_10.h b/cpp/src/qpid/client/SessionBase_0_10.h index 429f684424..2d1586d042 100644 --- a/cpp/src/qpid/client/SessionBase_0_10.h +++ b/cpp/src/qpid/client/SessionBase_0_10.h @@ -99,6 +99,9 @@ class SessionBase_0_10 { /** Resume a suspended session with a new connection */ void resume(Connection); + /** Get the channel associated with this session */ + uint16_t getChannel() const; + Execution& getExecution(); void flush(); void markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer); diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp index 14a666a1c6..9526a33ac6 100644 --- a/cpp/src/qpid/cluster/ClusterPlugin.cpp +++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp @@ -78,7 +78,7 @@ struct ClusterPlugin : public Plugin { if (values.name.empty()) return; // Only if --cluster-name option was specified. Broker* broker = dynamic_cast<Broker*>(&target); if (!broker) return; - cluster = new Cluster(values.name, values.getUrl(broker->getPort()), *broker); + cluster = new Cluster(values.name, values.getUrl(broker->getPort(Broker::TCP_TRANSPORT)), *broker); broker->setConnectionFactory( boost::shared_ptr<sys::ConnectionCodec::Factory>( new ConnectionCodec::Factory(broker->getConnectionFactory(), *cluster))); diff --git a/cpp/src/qpid/cluster/DumpClient.cpp b/cpp/src/qpid/cluster/DumpClient.cpp index c262115f9f..4bc001b4c6 100644 --- a/cpp/src/qpid/cluster/DumpClient.cpp +++ b/cpp/src/qpid/cluster/DumpClient.cpp @@ -166,11 +166,9 @@ void DumpClient::dumpConnection(const boost::intrusive_ptr<Connection>& dumpConn shadowConnection = catchUpConnection(); broker::Connection& bc = dumpConnection->getBrokerConnection(); - // FIXME aconway 2008-09-19: Open with identical settings to dumpConnection: password, vhost, frame size, - // authentication etc. See ConnectionSettings. - shadowConnection.open(dumpeeUrl, bc.getUserId()); - - dumpConnection->getBrokerConnection().eachSessionHandler(boost::bind(&DumpClient::dumpSession, this, _1)); + // FIXME aconway 2008-10-20: What authentication info to reconnect? + shadowConnection.open(dumpeeUrl, bc.getUserId(), ""/*password*/, "/"/*vhost*/, bc.getFrameMax()); + bc.eachSessionHandler(boost::bind(&DumpClient::dumpSession, this, _1)); ClusterConnectionProxy(shadowConnection).shadowReady( dumpConnection->getId().getMember(), reinterpret_cast<uint64_t>(dumpConnection->getId().getPointer())); @@ -194,20 +192,30 @@ void DumpClient::dumpSession(broker::SessionHandler& sh) { // For reasons unknown, boost::bind does not work here with boost 1.33. ss->eachConsumer(std::bind1st(std::mem_fun(&DumpClient::dumpConsumer),this)); - - // FIXME aconway 2008-09-19: update remaining session state. + + boost::intrusive_ptr<Message> inProgress = ss->getMessageInProgress(); + + // Adjust for message in progress, will be sent after state update. + SequenceNumber received = ss->receiverGetReceived().command; + if (inProgress) + --received; // Reset command-sequence state. proxy.sessionState( ss->senderGetReplayPoint().command, ss->senderGetCommandPoint().command, ss->senderGetIncomplete(), - ss->receiverGetExpected().command, - ss->receiverGetReceived().command, + std::max(received, ss->receiverGetExpected().command), + received, ss->receiverGetUnknownComplete(), ss->receiverGetIncomplete() ); + // Send frames for partial message in progress. + if (inProgress) { + inProgress->getFrames().map(simpl->out); + } + // FIXME aconway 2008-09-23: update session replay list. QPID_LOG(debug, dumperId << " dumped session " << sh.getSession()->getId()); |
