You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2009/01/08 18:33:09 UTC
svn commit: r732768 - in /qpid/trunk/qpid/cpp/src/qpid/cluster: Cluster.cpp
ClusterPlugin.cpp Multicaster.cpp Multicaster.h
Author: aconway
Date: Thu Jan 8 09:33:08 2009
New Revision: 732768
URL: http://svn.apache.org/viewvc?rev=732768&view=rev
Log:
cluster: handle multicast errors.
Modified:
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=732768&r1=732767&r2=732768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Thu Jan 8 09:33:08 2009
@@ -98,7 +98,7 @@
0, // write
boost::bind(&Cluster::disconnect, this, _1) // disconnect
),
- mcast(cpg, poller),
+ mcast(cpg, poller, boost::bind(&Cluster::leave, this)),
mgmtObject(0),
deliverQueue(boost::bind(&Cluster::delivered, this, _1), poller),
state(INIT),
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp?rev=732768&r1=732767&r2=732768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp Thu Jan 8 09:33:08 2009
@@ -69,9 +69,9 @@
("cluster-cman", optValue(values.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
#endif
("cluster-read-max", optValue(values.readMax,"N"),
- "Throttle read rate from client connections.")
+ "Experimental: Throttle read rate from client connections.")
("cluster-write-estimate", optValue(values.writeEstimate, "Kb"),
- "Estimate connection write rate per multicast cycle")
+ "Experimental: initial estimate for connection write per multicast cycle")
;
}
};
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp?rev=732768&r1=732767&r2=732768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.cpp Thu Jan 8 09:33:08 2009
@@ -28,8 +28,9 @@
namespace qpid {
namespace cluster {
-Multicaster::Multicaster(Cpg& cpg_, const boost::shared_ptr<sys::Poller>& poller) :
- cpg(cpg_), queue(boost::bind(&Multicaster::sendMcast, this, _1), poller),
+Multicaster::Multicaster(Cpg& cpg_, const boost::shared_ptr<sys::Poller>& poller, boost::function<void()> onError_) :
+ onError(onError_), cpg(cpg_),
+ queue(boost::bind(&Multicaster::sendMcast, this, _1), poller),
holding(true)
{
queue.start();
@@ -70,7 +71,9 @@
values.erase(values.begin(), i);
}
catch (const std::exception& e) {
- throw ClusterLeaveException(e.what());
+ QPID_LOG(critical, "Multicast error: " << e.what());
+ queue.stop();
+ onError();
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h?rev=732768&r1=732767&r2=732768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Multicaster.h Thu Jan 8 09:33:08 2009
@@ -46,7 +46,7 @@
{
public:
/** Starts in holding mode: connection data events are held, other events are mcast */
- Multicaster(Cpg& cpg_, const boost::shared_ptr<sys::Poller>& );
+ Multicaster(Cpg& cpg_, const boost::shared_ptr<sys::Poller>&, boost::function<void()> onError );
void mcastControl(const framing::AMQBody& controlBody, const ConnectionId&);
void mcastBuffer(const char*, size_t, const ConnectionId&);
void mcast(const Event& e);
@@ -60,6 +60,7 @@
void sendMcast(PollableEventQueue::Queue& );
sys::Mutex lock;
+ boost::function<void()> onError;
Cpg& cpg;
PollableEventQueue queue;
bool holding;