cluster: handle multicast errors.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@732768 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 11 insertions, 7 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index dd9de68bf5..ef6285481c 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -98,7 +98,7 @@ Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b, b
         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),
diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp
index 0f4944d392..783e4e5009 100644
--- a/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -69,9 +69,9 @@ struct ClusterOptions : public Options {
             ("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")
             ;
     }
 };
diff --git a/cpp/src/qpid/cluster/Multicaster.cpp b/cpp/src/qpid/cluster/Multicaster.cpp
index 37d2f81b39..a106ec128b 100644
--- a/cpp/src/qpid/cluster/Multicaster.cpp
+++ b/cpp/src/qpid/cluster/Multicaster.cpp
@@ -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 @@ void Multicaster::sendMcast(PollableEventQueue::Queue& values) {
         values.erase(values.begin(), i);
     }
     catch (const std::exception& e) {
-        throw ClusterLeaveException(e.what());
+        QPID_LOG(critical, "Multicast error: " << e.what());
+        queue.stop();
+        onError();
     }
 }
 
diff --git a/cpp/src/qpid/cluster/Multicaster.h b/cpp/src/qpid/cluster/Multicaster.h
index 8b306ce10e..a63bbc2180 100644
--- a/cpp/src/qpid/cluster/Multicaster.h
+++ b/cpp/src/qpid/cluster/Multicaster.h
@@ -46,7 +46,7 @@ class Multicaster
 {
   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 @@ class Multicaster
     void sendMcast(PollableEventQueue::Queue& );
 
     sys::Mutex lock;
+    boost::function<void()> onError;
     Cpg& cpg;
     PollableEventQueue queue;
     bool holding;