synchronized with trunk except for ruby dir

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

62 files changed, 6433 insertions, 963 deletions

diff --git a/cpp/src/qpid/cluster/ClassifierHandler.cpp b/cpp/src/qpid/cluster/ClassifierHandler.cpp
deleted file mode 100644
index b78f795d20..0000000000
--- a/cpp/src/qpid/cluster/ClassifierHandler.cpp
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#include "ClassifierHandler.h"
-
-#include "qpid/framing/FrameDefaultVisitor.h"
-#include "qpid/framing/AMQFrame.h"
-
-namespace qpid {
-namespace cluster {
-
-using namespace framing;
-
-struct ClassifierHandler::Visitor : public FrameDefaultVisitor {
-    Visitor(AMQFrame& f, ClassifierHandler& c)
-        : chosen(0), frame(f), classifier(c) { f.getBody()->accept(*this); }
-
-    void visit(const ExchangeDeclareBody&) { chosen=&classifier.wiring; }
-    void visit(const ExchangeDeleteBody&) { chosen=&classifier.wiring; }
-    void visit(const ExchangeBindBody&) { chosen=&classifier.wiring; }
-    void visit(const ExchangeUnbindBody&) { chosen=&classifier.wiring; }
-    void visit(const QueueDeclareBody&) { chosen=&classifier.wiring; }
-    void visit(const QueueDeleteBody&) { chosen=&classifier.wiring; }
-    void defaultVisit(const AMQBody&) { chosen=&classifier.other; }
-
-    using framing::FrameDefaultVisitor::visit;
-    using framing::FrameDefaultVisitor::defaultVisit;
-
-    FrameHandler* chosen;
-    AMQFrame& frame;
-    ClassifierHandler& classifier;
-};
-
-void ClassifierHandler::handle(AMQFrame& f) { Visitor(f, *this).chosen->handle(f); }
-
-}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ClassifierHandler.h b/cpp/src/qpid/cluster/ClassifierHandler.h
deleted file mode 100644
index 696e457c04..0000000000
--- a/cpp/src/qpid/cluster/ClassifierHandler.h
+++ /dev/null
@@ -1,50 +0,0 @@
-#ifndef QPID_CLUSTER_CLASSIFIERHANDLER_H
-#define QPID_CLUSTER_CLASSIFIERHANDLER_H
-
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#include "qpid/framing/FrameHandler.h"
-
-namespace qpid {
-namespace cluster {
-
-/**
- * Classify frames and forward to the appropriate handler.
- */
-class ClassifierHandler : public framing::FrameHandler
-{
-  public:
-    ClassifierHandler(framing::FrameHandler& wiring_,
-                      framing::FrameHandler& other_)
-        : wiring(wiring_), other(other_) {}
-
-    void handle(framing::AMQFrame&);
-    
-  private:
-    struct Visitor;
-  friend struct Visitor;
-    framing::FrameHandler& wiring;
-    framing::FrameHandler& other;
-};
-
-}} // namespace qpid::cluster
-
-
-
-#endif  /*!QPID_CLUSTER_CLASSIFIERHANDLER_H*/
diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 05ab9148b5..d049001eb0 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -16,305 +16,964 @@
  *
  */
 
-#include "Cluster.h"
-#include "ConnectionInterceptor.h"
-
+/** CLUSTER IMPLEMENTATION OVERVIEW
+ *
+ * The cluster works on the principle that if all members of the
+ * cluster receive identical input, they will all produce identical
+ * results. cluster::Connections intercept data received from clients
+ * and multicast it via CPG. The data is processed (passed to the
+ * broker::Connection) only when it is received from CPG in cluster
+ * order. Each cluster member has Connection objects for directly
+ * connected clients and "shadow" Connection objects for connections
+ * to other members.
+ *
+ * This assumes that all broker actions occur deterministically in
+ * response to data arriving on client connections. There are two
+ * situations where this assumption fails:
+ *  - sending data in response to polling local connections for writabiliy.
+ *  - taking actions based on a timer or timestamp comparison.
+ *
+ * IMPORTANT NOTE: any time code is added to the broker that uses timers,
+ * the cluster may need to be updated to take account of this.
+ * 
+ *
+ * USE OF TIMESTAMPS IN THE BROKER
+ *  
+ * The following are the current areas where broker uses timers or timestamps:
+ * 
+ * - Producer flow control: broker::SemanticState uses connection::getClusterOrderOutput.
+ *   a FrameHandler that sends frames to the client via the cluster. Used by broker::SessionState
+ *   
+ * - QueueCleaner, Message TTL: uses ExpiryPolicy, which is implemented by cluster::ExpiryPolicy.
+ * 
+ * - Connection heartbeat: sends connection controls, not part of session command counting so OK to ignore.
+ * 
+ * - LinkRegistry: only cluster elder is ever active for links.
+ * 
+ * - management::ManagementBroker: uses MessageHandler supplied by  cluster
+ *   to send messages to the broker via the cluster.
+ *   
+ * - Dtx: not yet supported with cluster.  
+ *
+ * cluster::ExpiryPolicy implements the strategy for message expiry.
+ *
+ * CLUSTER PROTOCOL OVERVIEW
+ * 
+ * Messages sent to/from CPG are called Events.
+ *
+ * An Event carries a ConnectionId, which includes a MemberId and a
+ * connection number.
+ * 
+ * Events are either
+ *  - Connection events: non-0 connection number and are associated with a connection.
+ *  - Cluster Events: 0 connection number, are not associated with a connection.
+ * 
+ * Events are further categorized as:
+ *  - Control: carries method frame(s) that affect cluster behavior.
+ *  - Data: carries raw data received from a client connection.
+ *
+ * The cluster defines extensions to the AMQP command set in ../../../xml/cluster.xml
+ * which defines two classes:
+ *  - cluster: cluster control information.
+ *  - cluster.connection: control information for a specific connection.
+ *
+ * The following combinations are legal:
+ *  - Data frames carrying connection data.
+ *  - Cluster control events carrying cluster commands.
+ *  - Connection control events carrying cluster.connection commands.
+ *  - Connection control events carrying non-cluster frames: frames sent to the client.
+ *    e.g. flow-control frames generated on a timer.
+ *
+ * CLUSTER INITIALIZATION OVERVIEW
+ *
+ * When a new member joins the CPG group, all members (including the
+ * new one) multicast their "initial status." The new member is in
+ * INIT mode until it gets a complete set of initial status messages
+ * from all cluster members.
+ *
+ * The newcomer uses initial status to determine
+ *  - The cluster UUID
+ *  - Am I speaking the correct version of the cluster protocol?
+ *  - Do I need to get an update from an existing active member?
+ *  - Can I recover from my own store?
+ *
+ * Initialization happens in the Cluster constructor (plugin
+ * early-init phase) because it needs to be done before the store
+ * initializes. In INIT mode sending & receiving from the cluster are
+ * done single-threaded, bypassing the normal PollableQueues because
+ * the Poller is not active at this point to service them.
+ */
+#include "qpid/Exception.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/cluster/ClusterSettings.h"
+#include "qpid/cluster/Connection.h"
+#include "qpid/cluster/UpdateClient.h"
+#include "qpid/cluster/RetractClient.h"
+#include "qpid/cluster/FailoverExchange.h"
+#include "qpid/cluster/UpdateExchange.h"
+
+#include "qpid/assert.h"
+#include "qmf/org/apache/qpid/cluster/ArgsClusterStopClusterNode.h"
+#include "qmf/org/apache/qpid/cluster/Package.h"
 #include "qpid/broker/Broker.h"
-#include "qpid/broker/SessionState.h"
 #include "qpid/broker/Connection.h"
+#include "qpid/broker/NullMessageStore.h"
+#include "qpid/broker/QueueRegistry.h"
+#include "qpid/broker/SessionState.h"
+#include "qpid/broker/SignalHandler.h"
 #include "qpid/framing/AMQFrame.h"
-#include "qpid/framing/ClusterNotifyBody.h"
-#include "qpid/framing/ClusterConnectionCloseBody.h"
+#include "qpid/framing/AMQP_AllOperations.h"
+#include "qpid/framing/AllInvoker.h"
+#include "qpid/framing/ClusterConfigChangeBody.h"
+#include "qpid/framing/ClusterConnectionDeliverCloseBody.h"
+#include "qpid/framing/ClusterConnectionAbortBody.h"
+#include "qpid/framing/ClusterRetractOfferBody.h"
+#include "qpid/framing/ClusterConnectionDeliverDoOutputBody.h"
+#include "qpid/framing/ClusterReadyBody.h"
+#include "qpid/framing/ClusterShutdownBody.h"
+#include "qpid/framing/ClusterUpdateOfferBody.h"
+#include "qpid/framing/ClusterUpdateRequestBody.h"
+#include "qpid/framing/ClusterConnectionAnnounceBody.h"
+#include "qpid/framing/ClusterErrorCheckBody.h"
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/log/Helpers.h"
 #include "qpid/log/Statement.h"
+#include "qpid/management/IdAllocator.h"
+#include "qpid/management/ManagementAgent.h"
 #include "qpid/memory.h"
-#include "qpid/shared_ptr.h"
+#include "qpid/sys/Thread.h"
 
+#include <boost/shared_ptr.hpp>
 #include <boost/bind.hpp>
 #include <boost/cast.hpp>
+#include <boost/current_function.hpp>
 #include <algorithm>
 #include <iterator>
 #include <map>
+#include <ostream>
 
 namespace qpid {
 namespace cluster {
+using namespace qpid;
 using namespace qpid::framing;
 using namespace qpid::sys;
+using namespace qpid::cluster;
+using namespace framing::cluster;
 using namespace std;
-using broker::Connection;
+using management::ManagementAgent;
+using management::ManagementObject;
+using management::Manageable;
+using management::Args;
+namespace _qmf = ::qmf::org::apache::qpid::cluster;
+
+/**
+ * NOTE: must increment this number whenever any incompatible changes in
+ * cluster protocol/behavior are made. It allows early detection and
+ * sensible reporting of an attempt to mix different versions in a
+ * cluster.
+ *
+ * Currently use SVN revision to avoid clashes with versions from
+ * different branches.
+ */
+const uint32_t Cluster::CLUSTER_VERSION = 884125;
 
-ostream& operator <<(ostream& out, const Cluster& cluster) {
-    return out << cluster.name.str() << "-" << cluster.self;
-}
+struct ClusterDispatcher : public framing::AMQP_AllOperations::ClusterHandler {
+    qpid::cluster::Cluster& cluster;
+    MemberId member;
+    Cluster::Lock& l;
+    ClusterDispatcher(Cluster& c, const MemberId& id, Cluster::Lock& l_) : cluster(c), member(id), l(l_) {}
 
-ostream& operator<<(ostream& out, const Cluster::MemberMap::value_type& m) {
-    return out << m.first << "=" << m.second.url;
-}
+    void updateRequest(const std::string& url) { cluster.updateRequest(member, url, l); }
 
-ostream& operator <<(ostream& out, const Cluster::MemberMap& members) {
-    ostream_iterator<Cluster::MemberMap::value_type> o(out, " ");
-    copy(members.begin(), members.end(), o);
-    return out;
-}
+    void initialStatus(uint32_t version, bool active, const Uuid& clusterId,
+                       uint8_t storeState, const Uuid& shutdownId)
+    {
+        cluster.initialStatus(member, version, active, clusterId, 
+                              framing::cluster::StoreState(storeState), shutdownId, l);
+    }
+    void ready(const std::string& url) { cluster.ready(member, url, l); }
+    void configChange(const std::string& current) { cluster.configChange(member, current, l); }
+    void updateOffer(uint64_t updatee) {
+        cluster.updateOffer(member, updatee, l);
+    }
+    void retractOffer(uint64_t updatee) { cluster.retractOffer(member, updatee, l); }
+    void messageExpired(uint64_t id) { cluster.messageExpired(member, id, l); }
+    void errorCheck(uint8_t type, const framing::SequenceNumber& frameSeq) {
+        cluster.errorCheck(member, type, frameSeq, l);
+    }
+
+    void shutdown(const Uuid& id) { cluster.shutdown(member, id, l); }
+
+    bool invoke(AMQBody& body) { return framing::invoke(*this, body).wasHandled(); }
+};
 
-Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b) :
-    broker(&b),
+Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
+    settings(set), 
+    broker(b),
+    mgmtObject(0),
     poller(b.getPoller()),
     cpg(*this),
-    name(name_),
-    url(url_),
+    name(settings.name),
+    myUrl(settings.url.empty() ? Url() : Url(settings.url)),
     self(cpg.self()),
-    cpgDispatchHandle(cpg,
-                      boost::bind(&Cluster::dispatch, this, _1), // read
-                      0,                                         // write
-                      boost::bind(&Cluster::disconnect, this, _1) // disconnect
-    ),
-    deliverQueue(boost::bind(&Cluster::deliverQueueCb, this, _1, _2)),
-    mcastQueue(boost::bind(&Cluster::mcastQueueCb, this, _1, _2))
+    clusterId(true),
+    expiryPolicy(new ExpiryPolicy(mcast, self, broker.getTimer())),
+    mcast(cpg, poller, boost::bind(&Cluster::leave, this)),
+    dispatcher(cpg, poller, boost::bind(&Cluster::leave, this)),
+    deliverEventQueue(boost::bind(&Cluster::deliveredEvent, this, _1),
+                      boost::bind(&Cluster::leave, this),
+                      "Error decoding events",
+                      poller),
+    deliverFrameQueue(boost::bind(&Cluster::deliveredFrame, this, _1),
+                      boost::bind(&Cluster::leave, this),
+                      "Error delivering frames",
+                      poller),
+    quorum(boost::bind(&Cluster::leave, this)),
+    decoder(boost::bind(&Cluster::deliverFrame, this, _1)),
+    discarding(true),
+    state(INIT),
+    initMap(self, settings.size),
+    store(broker.getDataDir().getPath()),
+    lastSize(0),
+    lastBroker(false),
+    updateRetracted(false),
+    error(*this)
 {
-    broker->addFinalizer(boost::bind(&Cluster::leave, this));
-    QPID_LOG(trace, "Joining cluster: " << name_);
-    cpg.join(name);
-    notify();
+    mAgent = broker.getManagementAgent();
+    if (mAgent != 0){
+        _qmf::Package  packageInit(mAgent);
+        mgmtObject = new _qmf::Cluster (mAgent, this, &broker,name,myUrl.str());
+        mAgent->addObject (mgmtObject);
+        mgmtObject->set_status("JOINING");
+    }
 
-    // FIXME aconway 2008-08-11: can we remove this loop?
-    // Dispatch till we show up in the cluster map.
-    while (empty()) 
-        cpg.dispatchOne();
+    // Failover exchange provides membership updates to clients.
+    failoverExchange.reset(new FailoverExchange(this));
+    broker.getExchanges().registerExchange(failoverExchange);
+
+    // Update exchange is used during updates to replicate messages
+    // without modifying delivery-properties.exchange.
+    broker.getExchanges().registerExchange(
+        boost::shared_ptr<broker::Exchange>(new UpdateExchange(this)));
+    // Load my store status before we go into initialization
+    if (! broker::NullMessageStore::isNullStore(&broker.getStore())) {
+        store.load();
+        if (store.getClusterId())
+            clusterId = store.getClusterId(); // Use stored ID if there is one.
+        QPID_LOG(notice, "Cluster store state: " << store)
+    }
 
-    // Start dispatching from the poller.
-    cpgDispatchHandle.startWatch(poller);
-    deliverQueue.start(poller);
-    mcastQueue.start(poller);
+    cpg.join(name);
+    // Pump the CPG dispatch manually till we get initialized. 
+    while (state == INIT)
+        cpg.dispatchOne();
 }
 
 Cluster::~Cluster() {
-    for (ShadowConnectionMap::iterator i = shadowConnectionMap.begin();
-         i != shadowConnectionMap.end();
-         ++i)
-    {
-        i->second->dirtyClose(); 
+    if (updateThread.id()) updateThread.join(); // Join the previous updatethread.
+}
+
+void Cluster::initialize() {
+    if (settings.quorum) quorum.start(poller);
+    if (myUrl.empty())
+        myUrl = Url::getIpAddressesUrl(broker.getPort(broker::Broker::TCP_TRANSPORT));
+    // Cluster constructor will leave us in either READY or JOINER state.
+    switch (state) {
+      case READY:
+        mcast.mcastControl(ClusterReadyBody(ProtocolVersion(), myUrl.str()), self);
+        break;
+      case JOINER:
+        mcast.mcastControl(ClusterUpdateRequestBody(ProtocolVersion(), myUrl.str()), self);
+        break;
+      default:
+        assert(0);
     }
-    std::for_each(localConnectionSet.begin(), localConnectionSet.end(), boost::bind(&ConnectionInterceptor::dirtyClose, _1));
+    QPID_LOG(notice, *this << (state == READY ? " joined" : " joining") << " cluster " << name);
+    broker.getKnownBrokers = boost::bind(&Cluster::getUrls, this);
+    broker.setExpiryPolicy(expiryPolicy);
+    dispatcher.start();
+    deliverEventQueue.start();
+    deliverFrameQueue.start();
+
+    // Add finalizer last for exception safety.
+    broker.addFinalizer(boost::bind(&Cluster::brokerShutdown, this)); 
 }
 
-// local connection initializes plugins
-void Cluster::initialize(broker::Connection& c) {
-    bool isLocal = &c.getOutput() != &shadowOut;
-    if (isLocal)
-        localConnectionSet.insert(new ConnectionInterceptor(c, *this));
+// Called in connection thread to insert a client connection.
+void Cluster::addLocalConnection(const boost::intrusive_ptr<Connection>& c) {
+    QPID_LOG(info, *this << " new local connection " << c->getId());
+    localConnections.insert(c);
+    assert(c->getId().getMember() == self);
+    // Announce the connection to the cluster.
+    if (c->isLocalClient())
+        mcast.mcastControl(ClusterConnectionAnnounceBody(ProtocolVersion(),
+                                                         c->getBrokerConnection().getSSF() ),
+                           c->getId());
 }
 
-void Cluster::leave() {
-    Mutex::ScopedLock l(lock);
-    if (!broker) return;                               // Already left.
-    // At this point the poller has already been shut down so
-    // no dispatches can occur thru the cpgDispatchHandle.
-    // 
-    // FIXME aconway 2008-08-11: assert this is the cae.
-    
-    QPID_LOG(debug, "Leaving cluster " << *this);
-    cpg.leave(name);
-    // broker= is set to 0 when the final config-change is delivered.
-    while(broker) {
-        Mutex::ScopedUnlock u(lock);
-        cpg.dispatchAll();
-    }
-    cpg.shutdown();
+// Called in connection thread to insert an updated shadow connection.
+void Cluster::addShadowConnection(const boost::intrusive_ptr<Connection>& c) {
+    QPID_LOG(info, *this << " new shadow connection " << c->getId());
+    // Safe to use connections here because we're pre-catchup, stalled
+    // and discarding, so deliveredFrame is not processing any
+    // connection events.
+    assert(discarding);         
+    pair<ConnectionMap::iterator, bool> ib
+        = connections.insert(ConnectionMap::value_type(c->getId(), c));
+    assert(ib.second);
+}
+
+void Cluster::erase(const ConnectionId& id) {
+    Lock l(lock);    
+    erase(id,l);
 }
 
-template <class T> void decodePtr(Buffer& buf, T*& ptr) {
-    uint64_t value = buf.getLongLong();
-    ptr = reinterpret_cast<T*>(value);
+// Called by Connection::deliverClose() in deliverFrameQueue thread.
+void Cluster::erase(const ConnectionId& id, Lock&) {
+    QPID_LOG(info, *this << " connection closed " << id);
+    connections.erase(id);
+    decoder.erase(id);
 }
 
-template <class T> void encodePtr(Buffer& buf, T* ptr) {
-    uint64_t value = reinterpret_cast<uint64_t>(ptr);
-    buf.putLongLong(value);
+std::vector<string> Cluster::getIds() const {
+    Lock l(lock);
+    return getIds(l);
 }
 
-void Cluster::send(const AMQFrame& frame, ConnectionInterceptor* connection) {
-    QPID_LOG(trace, "MCAST [" << connection << "] " << frame);
-    mcastQueue.push(Message(frame, self, connection));
+std::vector<string> Cluster::getIds(Lock&) const {
+    return map.memberIds();
 }
 
-void Cluster::mcastQueueCb(const MessageQueue::iterator& begin,
-                           const MessageQueue::iterator& end)
+std::vector<Url> Cluster::getUrls() const {
+    Lock l(lock);
+    return getUrls(l);
+}
+
+std::vector<Url> Cluster::getUrls(Lock&) const {
+    return map.memberUrls();
+} 
+
+void Cluster::leave() { 
+    Lock l(lock);
+    leave(l);
+}
+
+#define LEAVE_TRY(STMT) try { STMT; }                                   \
+    catch (const std::exception& e) {                                   \
+        QPID_LOG(warning, *this << " error leaving cluster: " << e.what()); \
+    } do {} while(0)
+
+void Cluster::leave(Lock&) { 
+    if (state != LEFT) {
+        state = LEFT;
+        QPID_LOG(notice, *this << " leaving cluster " << name);
+        // Finalize connections now now to avoid problems later in destructor.
+        LEAVE_TRY(localConnections.clear());
+        LEAVE_TRY(connections.clear());
+        LEAVE_TRY(broker::SignalHandler::shutdown());
+    }
+}
+
+// Deliver CPG message.
+void Cluster::deliver(
+    cpg_handle_t /*handle*/,
+    const cpg_name* /*group*/,
+    uint32_t nodeid,
+    uint32_t pid,
+    void* msg,
+    int msg_len) 
 {
-    // Static is OK because there is only one cluster allowed per
-    // process and only one thread in mcastQueueCb at a time.
-    static char buffer[64*1024]; // FIXME aconway 2008-07-04: buffer management.
-    MessageQueue::iterator i = begin;
-    while (i != end) {
-        Buffer buf(buffer, sizeof(buffer));
-        while (i != end && buf.available() > i->frame.size() + sizeof(uint64_t)) {
-            i->frame.encode(buf);
-            encodePtr(buf, i->connection);
-            ++i;
+    MemberId from(nodeid, pid);
+    framing::Buffer buf(static_cast<char*>(msg), msg_len);
+    Event e(Event::decodeCopy(from, buf));
+    deliverEvent(e);
+}
+
+void Cluster::deliverEvent(const Event& e) {
+    // During initialization, execute events directly in the same thread.
+    // Once initialized, push to pollable queue to be processed in another thread.
+    if (state == INIT)
+        deliveredEvent(e);
+    else
+        deliverEventQueue.push(e);
+}
+
+void Cluster::deliverFrame(const EventFrame& e) {
+    // During initialization, execute events directly in the same thread.
+    // Once initialized, push to pollable queue to be processed in another thread.
+    if (state == INIT)
+        deliveredFrame(e);
+    else
+        deliverFrameQueue.push(e);
+}
+
+const ClusterUpdateOfferBody* castUpdateOffer(const framing::AMQBody* body) {
+    return  (body && body->getMethod() &&
+             body->getMethod()->isA<ClusterUpdateOfferBody>()) ?
+        static_cast<const ClusterUpdateOfferBody*>(body) : 0;
+}
+
+const ClusterConnectionAnnounceBody* castAnnounce( const framing::AMQBody *body) {
+    return  (body && body->getMethod() &&
+             body->getMethod()->isA<ClusterConnectionAnnounceBody>()) ?
+      static_cast<const ClusterConnectionAnnounceBody*>(body) : 0;
+}
+
+// Handler for deliverEventQueue.
+// This thread decodes frames from events.
+void Cluster::deliveredEvent(const Event& e) {
+    if (e.isCluster()) {
+        EventFrame ef(e, e.getFrame());
+        // Stop the deliverEventQueue on update offers.
+        // This preserves the connection decoder fragments for an update.
+        const ClusterUpdateOfferBody* offer = castUpdateOffer(ef.frame.getBody());
+        if (offer) {
+            QPID_LOG(info, *this << " stall for update offer from " << e.getMemberId()
+                     << " to " << MemberId(offer->getUpdatee()));
+            deliverEventQueue.stop();
+        }
+        deliverFrame(ef);
+    }
+    else if(!discarding) { 
+        if (e.isControl())
+            deliverFrame(EventFrame(e, e.getFrame()));
+        else {
+            try { decoder.decode(e, e.getData()); }
+            catch (const Exception& ex) {
+                // Close a connection that is sending us invalid data.
+                QPID_LOG(error, *this << " aborting connection "
+                         << e.getConnectionId() << ": " << ex.what());
+                framing::AMQFrame abort((ClusterConnectionAbortBody()));
+                deliverFrame(EventFrame(EventHeader(CONTROL, e.getConnectionId()), abort));
+            }
         }
-        iovec iov = { buffer, buf.getPosition() };
-        cpg.mcast(name, &iov, 1);
     }
 }
 
-void Cluster::notify() {
-    send(AMQFrame(in_place<ClusterNotifyBody>(ProtocolVersion(), url.str())), 0);
+void Cluster::flagError(
+    Connection& connection, ErrorCheck::ErrorType type, const std::string& msg)
+{
+    Mutex::ScopedLock l(lock);
+    if (connection.isCatchUp()) {
+        QPID_LOG(critical, *this << " error on update connection " << connection
+                 << ": " << msg);
+        leave(l);
+    }
+    error.error(connection, type, map.getFrameSeq(), map.getMembers(), msg);
 }
 
-size_t Cluster::size() const {
+// Handler for deliverFrameQueue.
+// This thread executes the main logic.
+void Cluster::deliveredFrame(const EventFrame& efConst) {
     Mutex::ScopedLock l(lock);
-    return members.size();
+    if (state == LEFT) return;
+    EventFrame e(efConst);
+    const ClusterUpdateOfferBody* offer = castUpdateOffer(e.frame.getBody());
+    if (offer && error.isUnresolved()) {
+        // We can't honour an update offer that is delivered while an
+        // error is in progress so replace it with a retractOffer and re-start
+        // the event queue.
+        e.frame = AMQFrame(
+            ClusterRetractOfferBody(ProtocolVersion(), offer->getUpdatee()));
+        deliverEventQueue.start(); 
+    }
+    // Process each frame through the error checker.
+    if (error.isUnresolved()) {
+        error.delivered(e);
+        while (error.canProcess())  // There is a frame ready to process.
+            processFrame(error.getNext(), l);
+    }
+    else 
+        processFrame(e, l);
 }
 
-Cluster::MemberList Cluster::getMembers() const {
-    Mutex::ScopedLock l(lock);
-    MemberList result(members.size());
-    std::transform(members.begin(), members.end(), result.begin(),
-                   boost::bind(&MemberMap::value_type::second, _1));
-    return result;        
+
+void Cluster::processFrame(const EventFrame& e, Lock& l) {
+    if (e.isCluster()) {
+        QPID_LOG(trace, *this << " DLVR: " << e);
+        ClusterDispatcher dispatch(*this, e.connectionId.getMember(), l);
+        if (!framing::invoke(dispatch, *e.frame.getBody()).wasHandled())
+            throw Exception(QPID_MSG("Invalid cluster control"));
+    }
+    else if (state >= CATCHUP) {
+        map.incrementFrameSeq();
+        ConnectionPtr connection = getConnection(e, l);
+        if (connection) {
+            QPID_LOG(trace, *this << " DLVR " << map.getFrameSeq() << ":  " << e);
+            connection->deliveredFrame(e);
+        }
+        else
+            QPID_LOG(trace, *this << " DROP (no connection): " << e);
+    }
+    else // Drop connection frames while state < CATCHUP
+        QPID_LOG(trace, *this << " DROP (joining): " << e);
 }
 
-// ################ HERE - leaking shadow connections.
-// FIXME aconway 2008-08-11: revisit memory management for shadow
-// connections, what if the Connection is closed other than via
-// disconnect? Dangling pointer in shadow map. Use ptr_map for shadow
-// map, add deleted state to ConnectionInterceptor? Interceptors need
-// to know about map? Check how Connections can be deleted.
+// Called in deliverFrameQueue thread
+ConnectionPtr Cluster::getConnection(const EventFrame& e, Lock&) {
+    ConnectionId id = e.connectionId;
+    ConnectionMap::iterator i = connections.find(id);
+    if (i != connections.end()) return i->second;
+    ConnectionPtr cp;
+    // If the frame is an announcement for a new connection, add it.
+    if (e.frame.getBody() && e.frame.getMethod() &&
+        e.frame.getMethod()->isA<ClusterConnectionAnnounceBody>())
+    {
+        if (id.getMember() == self)  { // Announces one of my own
+            cp = localConnections.getErase(id);
+            assert(cp); 
+        }
+        else {              // New remote connection, create a shadow.
+            std::ostringstream mgmtId;
+            unsigned int ssf;
+            const ClusterConnectionAnnounceBody *announce = castAnnounce(e.frame.getBody());
+
+            mgmtId << id;
+            ssf = (announce && announce->hasSsf()) ? announce->getSsf() : 0;
+            QPID_LOG(debug, *this << "new connection's ssf =" << ssf );
+            cp = new Connection(*this, shadowOut, mgmtId.str(), id, ssf );
+        }
+        connections.insert(ConnectionMap::value_type(id, cp));
+    }
+    return cp;
+}
 
-ConnectionInterceptor* Cluster::getShadowConnection(const Cpg::Id& member, void* remotePtr) {
-    ShadowConnectionId id(member, remotePtr);
-    ShadowConnectionMap::iterator i = shadowConnectionMap.find(id);
-    if (i == shadowConnectionMap.end()) { // A new shadow connection.
-        std::ostringstream os;
-        os << name << ":"  << member << ":" << remotePtr;
-        assert(broker);
-        broker::Connection* c = new broker::Connection(&shadowOut, *broker, os.str());
-        ShadowConnectionMap::value_type value(id, new ConnectionInterceptor(*c, *this, id));
-        i = shadowConnectionMap.insert(value).first;
+Cluster::ConnectionVector Cluster::getConnections(Lock&) {
+    ConnectionVector result(connections.size());
+    std::transform(connections.begin(), connections.end(), result.begin(),
+                   boost::bind(&ConnectionMap::value_type::second, _1));
+    return result;
+}
+  
+struct AddrList {
+    const cpg_address* addrs;
+    int count;
+    const char *prefix, *suffix;
+    AddrList(const cpg_address* a, int n, const char* p="", const char* s="")
+        : addrs(a), count(n), prefix(p), suffix(s) {}
+};
+
+ostream& operator<<(ostream& o, const AddrList& a) {
+    if (!a.count) return o;
+    o << a.prefix;
+    for (const cpg_address* p = a.addrs; p < a.addrs+a.count; ++p) {
+        const char* reasonString;
+        switch (p->reason) {
+          case CPG_REASON_JOIN: reasonString =  "(joined) "; break;
+          case CPG_REASON_LEAVE: reasonString =  "(left) "; break;
+          case CPG_REASON_NODEDOWN: reasonString =  "(node-down) "; break;
+          case CPG_REASON_NODEUP: reasonString =  "(node-up) "; break;
+          case CPG_REASON_PROCDOWN: reasonString =  "(process-down) "; break;
+          default: reasonString = " ";
+        }
+        qpid::cluster::MemberId member(*p);
+        o << member << reasonString;
     }
-    return i->second;
+    return o << a.suffix;
 }
 
-void Cluster::deliver(
+void Cluster::configChange ( 
     cpg_handle_t /*handle*/,
-    cpg_name* /*group*/,
-    uint32_t nodeid,
-    uint32_t pid,
-    void* msg,
-    int msg_len)
+    const cpg_name */*group*/,
+    const cpg_address *current, int nCurrent,
+    const cpg_address *left, int nLeft,
+    const cpg_address *joined, int nJoined)
 {
-    Id from(nodeid, pid);
-    try {
-        Buffer buf(static_cast<char*>(msg), msg_len);
-        while (buf.available() > 0) {
-            AMQFrame frame;
-            if (!frame.decode(buf))  // Not enough data.
-                throw Exception("Received incomplete cluster event.");
-            void* connection;
-            decodePtr(buf, connection);
-            deliverQueue.push(Message(frame, from, connection));
+    Mutex::ScopedLock l(lock);
+    QPID_LOG(notice, *this << " membership change: "
+             << AddrList(current, nCurrent) << "("
+             << AddrList(joined, nJoined, "joined: ")
+             << AddrList(left, nLeft, "left: ")
+             << ")");
+    std::string addresses;
+    for (const cpg_address* p = current; p < current+nCurrent; ++p) 
+        addresses.append(MemberId(*p).str());
+    deliverEvent(Event::control(ClusterConfigChangeBody(ProtocolVersion(), addresses), self));
+}
+
+void Cluster::setReady(Lock&) {
+    state = READY;
+    if (mgmtObject!=0) mgmtObject->set_status("ACTIVE");
+    mcast.setReady();
+    broker.getQueueEvents().enable();
+}
+
+void Cluster::initMapCompleted(Lock& l) {
+    // Called on completion of the initial status map. 
+    QPID_LOG(debug, *this << " initial status map complete. ");
+    if (state == INIT) {
+        // We have status for all members so we can make join descisions.
+        initMap.checkConsistent();
+        elders = initMap.getElders();
+        QPID_LOG(debug, *this << " elders: " << elders);
+        if (!elders.empty()) { // I'm not the elder, I don't handle links & replication.
+            broker.getLinks().setPassive(true);
+            broker.getQueueEvents().disable();
+            QPID_LOG(info, *this << " not active for links.");
+        }
+        else {
+            QPID_LOG(info, this << " active for links.");
+        }
+        setClusterId(initMap.getClusterId(), l);
+        if (store.hasStore()) store.dirty(clusterId);
+
+        if (initMap.isUpdateNeeded())  { // Joining established cluster.
+            broker.setRecovery(false); // Ditch my current store.
+            broker.setClusterUpdatee(true);
+            state = JOINER;
+        }
+        else {                  // I can go ready.
+            discarding = false;
+            setReady(l);
+            memberUpdate(l);
         }
+        QPID_LOG(debug, *this << "Initialization complete");
+    }
+}
+
+void Cluster::configChange(const MemberId&, const std::string& configStr, Lock& l) {
+    if (state == LEFT) return;
+
+    MemberSet config = decodeMemberSet(configStr);
+    elders = intersection(elders, config);
+    if (elders.empty() && INIT < state && state < CATCHUP) {
+        QPID_LOG(critical, "Cannot update, all potential updaters left the cluster.");
+        leave(l);
+        return;
     }
+    bool memberChange = map.configChange(config);
+
+    // Update initital status for new members joining.
+    initMap.configChange(config);
+    if (initMap.isResendNeeded()) {
+        mcast.mcastControl(
+            ClusterInitialStatusBody(
+                ProtocolVersion(), CLUSTER_VERSION, state > INIT, clusterId, 
+                store.getState(), store.getShutdownId()
+            ),
+            self);
+    }
+    if (initMap.transitionToComplete()) initMapCompleted(l);
+
+    if (state >= CATCHUP && memberChange) {
+        memberUpdate(l);
+        if (elders.empty()) {
+            // We are the oldest, reactive links if necessary
+            QPID_LOG(info, this << " becoming active for links.");
+            broker.getLinks().setPassive(false);
+        }
+    }
+}
+
+void Cluster::makeOffer(const MemberId& id, Lock& ) {
+    if (state == READY && map.isJoiner(id)) {
+        state = OFFER;
+        QPID_LOG(info, *this << " send update-offer to " << id);
+        mcast.mcastControl(ClusterUpdateOfferBody(ProtocolVersion(), id), self);
+    }
+}
+
+// Called from Broker::~Broker when broker is shut down.  At this
+// point we know the poller has stopped so no poller callbacks will be
+// invoked. We must ensure that CPG has also shut down so no CPG
+// callbacks will be invoked.
+// 
+void Cluster::brokerShutdown()  {
+    try { cpg.shutdown(); }
     catch (const std::exception& e) {
-        // FIXME aconway 2008-01-30: exception handling.
-        QPID_LOG(critical, "Error in cluster deliver: " << e.what());
-        assert(0);
-        throw;
+        QPID_LOG(error, *this << " shutting down CPG: " << e.what());
     }
+    delete this;
+}
+
+void Cluster::updateRequest(const MemberId& id, const std::string& url, Lock& l) {
+    map.updateRequest(id, url);
+    makeOffer(id, l);
 }
 
-void Cluster::deliverQueueCb(const MessageQueue::iterator& begin,
-                             const MessageQueue::iterator& end)
+void Cluster::initialStatus(const MemberId& member, uint32_t version, bool active,
+                            const framing::Uuid& id, 
+                            framing::cluster::StoreState store,
+                            const framing::Uuid& shutdownId,
+                            Lock& l)
 {
-    for (MessageQueue::iterator i = begin; i != end; ++i) {
-        AMQFrame& frame(i->frame);
-        Id from(i->from);
-        ConnectionInterceptor* connection = reinterpret_cast<ConnectionInterceptor*>(i->connection);
-        try {
-            QPID_LOG(trace, "DLVR [" << from << " " << connection << "] " << frame);
-
-            if (!broker) {
-                QPID_LOG(warning, "Unexpected DLVR, already left the cluster.");
-                return;
-            }
-            if (connection && from != self) // Look up shadow for remote connections
-                connection = getShadowConnection(from, connection);
+    if (version != CLUSTER_VERSION) {
+        QPID_LOG(critical, *this << " incompatible cluster versions " <<
+                 version << " != " << CLUSTER_VERSION);
+        leave(l);
+        return;
+    }
+    initMap.received(
+        member,
+        ClusterInitialStatusBody(ProtocolVersion(), version, active, id, store, shutdownId)
+    );
+    if (initMap.transitionToComplete()) initMapCompleted(l);
+}
 
-            if (frame.getMethod() && frame.getMethod()->amqpClassId() == CLUSTER_CLASS_ID) 
-                handleMethod(from, connection, *frame.getMethod());
-            else 
-                connection->deliver(frame);
+void Cluster::ready(const MemberId& id, const std::string& url, Lock& l) {
+    if (map.ready(id, Url(url))) 
+        memberUpdate(l);
+    if (state == CATCHUP && id == self) {
+        setReady(l);
+        QPID_LOG(notice, *this << " caught up.");
+    }
+}
+
+void Cluster::updateOffer(const MemberId& updater, uint64_t updateeInt, Lock& l) {
+    // NOTE: deliverEventQueue has been stopped at the update offer by
+    // deliveredEvent in case an update is required.
+    if (state == LEFT) return;
+    MemberId updatee(updateeInt);
+    boost::optional<Url> url = map.updateOffer(updater, updatee);
+    if (updater == self) {
+        assert(state == OFFER);
+        if (url)               // My offer was first.
+            updateStart(updatee, *url, l);
+        else {                  // Another offer was first.
+            QPID_LOG(info, *this << " cancelled offer to " << updatee << " unstall");
+            setReady(l);
+            makeOffer(map.firstJoiner(), l); // Maybe make another offer.
+            deliverEventQueue.start(); // Go back to normal processing
         }
-        catch (const std::exception& e) {
-            // FIXME aconway 2008-01-30: exception handling.
-            QPID_LOG(critical, "Error in cluster deliverQueueCb: " << e.what());
-            assert(0);
-            throw;
+    }
+    else if (updatee == self && url) {
+        assert(state == JOINER);
+        state = UPDATEE;
+        QPID_LOG(notice, *this << " receiving update from " << updater);
+        checkUpdateIn(l);
+    }
+    else {
+        QPID_LOG(debug,*this << " unstall, ignore update " << updater
+                 << " to " << updatee);
+        deliverEventQueue.start(); // Not involved in update.
+    }
+}
+
+static client::ConnectionSettings connectionSettings(const ClusterSettings& settings) {
+    client::ConnectionSettings cs;
+    cs.username = settings.username;
+    cs.password = settings.password;
+    cs.mechanism = settings.mechanism;
+    return cs;
+}
+
+void Cluster::retractOffer(const MemberId& updater, uint64_t updateeInt, Lock& l) {
+    // An offer was received while handling an error, and converted to a retract.
+    // Behavior is very similar to updateOffer.
+    if (state == LEFT) return;
+    MemberId updatee(updateeInt);
+    boost::optional<Url> url = map.updateOffer(updater, updatee);
+    if (updater == self) {
+        assert(state == OFFER);
+        if (url)  {             // My offer was first.
+            if (updateThread.id())
+                updateThread.join(); // Join the previous updateThread to avoid leaks.
+            updateThread = Thread(new RetractClient(*url, connectionSettings(settings)));
         }
+        setReady(l);
+        makeOffer(map.firstJoiner(), l); // Maybe make another offer.
+        // Don't unstall the event queue, that was already done in deliveredFrame
     }
+    QPID_LOG(debug,*this << " retracted offer " << updater << " to " << updatee);
+}
+
+void Cluster::updateStart(const MemberId& updatee, const Url& url, Lock& l) {
+    // NOTE: deliverEventQueue is already stopped at the stall point by deliveredEvent.
+    if (state == LEFT) return;
+    assert(state == OFFER);
+    state = UPDATER;
+    QPID_LOG(notice, *this << " sending update to " << updatee << " at " << url);
+    if (updateThread.id())
+        updateThread.join(); // Join the previous updateThread to avoid leaks.
+    updateThread = Thread(
+        new UpdateClient(self, updatee, url, broker, map, *expiryPolicy,
+                         getConnections(l), decoder,
+                         boost::bind(&Cluster::updateOutDone, this),
+                         boost::bind(&Cluster::updateOutError, this, _1),
+                         connectionSettings(settings)));
 }
 
-// Handle cluster methods
-// FIXME aconway 2008-07-11: Generate/template a better dispatch mechanism.
-void Cluster::handleMethod(Id from, ConnectionInterceptor* connection, AMQMethodBody& method) {
-    assert(method.amqpClassId() == CLUSTER_CLASS_ID);
-    switch (method.amqpMethodId()) {
-      case CLUSTER_NOTIFY_METHOD_ID: {
-          ClusterNotifyBody& notify=static_cast<ClusterNotifyBody&>(method);
-          Mutex::ScopedLock l(lock);
-          members[from].url=notify.getUrl();
-          lock.notifyAll();
-          break;
-      }
-      case CLUSTER_CONNECTION_CLOSE_METHOD_ID: {
-          if (!connection->isLocal())
-              shadowConnectionMap.erase(connection->getShadowId());
-          else
-              localConnectionSet.erase(connection);
-          connection->deliverClosed();
-          break;
-      }
-      case CLUSTER_CONNECTION_DO_OUTPUT_METHOD_ID: {
-          connection->deliverDoOutput();
-          break;
-      }
+// Called in update thread.
+void Cluster::updateInDone(const ClusterMap& m) {
+    Lock l(lock);
+    updatedMap = m;
+    checkUpdateIn(l);
+}
+
+void Cluster::updateInRetracted() {
+    Lock l(lock);
+    updateRetracted = true;
+    map.clearStatus();
+    checkUpdateIn(l);
+}
+
+void Cluster::checkUpdateIn(Lock& l) {
+    if (state != UPDATEE) return; // Wait till we reach the stall point.
+    if (updatedMap) { // We're up to date
+        map = *updatedMap;
+        memberUpdate(l);
+        mcast.mcastControl(ClusterReadyBody(ProtocolVersion(), myUrl.str()), self);
+        state = CATCHUP;
+        broker.setClusterUpdatee(false);
+        discarding = false;     // ok to set, we're stalled for update.
+        QPID_LOG(notice, *this << " update complete, starting catch-up.");
+        deliverEventQueue.start();
+    }
+    else if (updateRetracted) { // Update was retracted, request another update
+        updateRetracted = false;
+        state = JOINER;
+        QPID_LOG(notice, *this << " update retracted, sending new update request.");
+        mcast.mcastControl(ClusterUpdateRequestBody(ProtocolVersion(), myUrl.str()), self);
+        deliverEventQueue.start();
+    }
+}
+
+void Cluster::updateOutDone() {
+    Monitor::ScopedLock l(lock);
+    updateOutDone(l);
+}
+
+void Cluster::updateOutDone(Lock& l) {
+    QPID_LOG(notice, *this << " update sent");
+    assert(state == UPDATER);
+    state = READY;
+     deliverEventQueue.start();       // Start processing events again.
+    makeOffer(map.firstJoiner(), l); // Try another offer
+}
+
+void Cluster::updateOutError(const std::exception& e)  {
+    Monitor::ScopedLock l(lock);
+    QPID_LOG(error, *this << " error sending update: " << e.what());    
+    updateOutDone(l);
+}
+
+void Cluster ::shutdown(const MemberId& , const Uuid& id, Lock& l) {
+    QPID_LOG(notice, *this << " cluster shut down by administrator.");
+    if (store.hasStore()) store.clean(Uuid(id));
+    leave(l);
+}
+
+ManagementObject* Cluster::GetManagementObject() const { return mgmtObject; }
+
+Manageable::status_t Cluster::ManagementMethod (uint32_t methodId, Args& args, string&) {
+    Lock l(lock);
+    QPID_LOG(debug, *this << " managementMethod [id=" << methodId << "]");
+    switch (methodId) {
+      case _qmf::Cluster::METHOD_STOPCLUSTERNODE :
+        {
+            _qmf::ArgsClusterStopClusterNode& iargs = (_qmf::ArgsClusterStopClusterNode&) args;
+            stringstream stream;
+            stream << self;
+            if (iargs.i_brokerId == stream.str())
+                stopClusterNode(l);
+        }
+        break;
+      case _qmf::Cluster::METHOD_STOPFULLCLUSTER :
+        stopFullCluster(l);
+        break;
       default:
-        assert(0);
+        return Manageable::STATUS_UNKNOWN_METHOD;
     }
+    return Manageable::STATUS_OK;
 }
 
-void Cluster::configChange(
-    cpg_handle_t /*handle*/,
-    cpg_name */*group*/,
-    cpg_address *current, int nCurrent,
-    cpg_address *left, int nLeft,
-    cpg_address */*joined*/, int nJoined)
-{
-    Mutex::ScopedLock l(lock);
-    for (int i = 0; i < nLeft; ++i)  
-        members.erase(left[i]);
-    for(int j = 0; j < nCurrent; ++j) 
-        members[current[j]].id = current[j];
-    QPID_LOG(debug, "Cluster members: " << nCurrent << " ("<< nLeft << " left, " << nJoined << " joined):"
-             << members);
-    assert(members.size() == size_t(nCurrent));
-    if (members.find(self) == members.end()) 
-        broker = 0;       // We have left the group, this is the final config change.
-    lock.notifyAll();     // Threads waiting for membership changes.  
+void Cluster::stopClusterNode(Lock& l) {
+    QPID_LOG(notice, *this << " cluster member stopped by administrator.");
+    leave(l);
 }
 
-void Cluster::dispatch(sys::DispatchHandle& h) {
-    cpg.dispatchAll();
-    h.rewatch();
+void Cluster::stopFullCluster(Lock& ) {
+    QPID_LOG(notice, *this << " shutting down cluster " << name);
+    mcast.mcastControl(ClusterShutdownBody(ProtocolVersion(), Uuid(true)), self);
 }
 
-void Cluster::disconnect(sys::DispatchHandle& h) {
-    h.stopWatch();
-    // FIXME aconway 2008-08-11: error handling if we are disconnected. 
-    // Kill the broker?
-    assert(0);
+void Cluster::memberUpdate(Lock& l) {
+    QPID_LOG(info, *this << " member update: " << map);
+    std::vector<Url> urls = getUrls(l);
+    std::vector<string> ids = getIds(l);
+    size_t size = urls.size();
+    failoverExchange->setUrls(urls);
+
+    if (size == 1 && lastSize > 1 && state >= CATCHUP) { 
+        QPID_LOG(notice, *this << " last broker standing, update queue policies");
+        lastBroker = true;
+        broker.getQueues().updateQueueClusterState(true);
+    }
+    else if (size > 1 && lastBroker) {
+        QPID_LOG(notice, *this << " last broker standing joined by " << size-1 << " replicas, updating queue policies" << size);
+        lastBroker = false;
+        broker.getQueues().updateQueueClusterState(false);
+    }
+    lastSize = size;
+
+    if (mgmtObject) {
+        mgmtObject->set_clusterSize(size); 
+        string urlstr;
+        for(std::vector<Url>::iterator iter = urls.begin(); iter != urls.end(); iter++ ) {
+            if (iter != urls.begin()) urlstr += ";";
+            urlstr += iter->str();
+        }
+        string idstr;
+        for(std::vector<string>::iterator iter = ids.begin(); iter != ids.end(); iter++ ) {
+            if (iter != ids.begin()) idstr += ";";
+            idstr += (*iter);
+        }
+        mgmtObject->set_members(urlstr);
+        mgmtObject->set_memberIDs(idstr);
+    }
+
+    // Close connections belonging to members that have left the cluster.
+    ConnectionMap::iterator i = connections.begin();
+    while (i != connections.end()) {
+        ConnectionMap::iterator j = i++;
+        MemberId m = j->second->getId().getMember();
+        if (m != self && !map.isMember(m)) {
+            j->second->getBrokerConnection().closed();
+            erase(j->second->getId(), l);
+        }
+    }
 }
 
-}} // namespace qpid::cluster
+std::ostream& operator<<(std::ostream& o, const Cluster& cluster) {
+    static const char* STATE[] = {
+        "INIT", "JOINER", "UPDATEE", "CATCHUP", "READY", "OFFER", "UPDATER", "LEFT"
+    };
+    assert(sizeof(STATE)/sizeof(*STATE) == Cluster::LEFT+1);
+    o << "cluster(" << cluster.self << " " << STATE[cluster.state];
+    if (cluster.error.isUnresolved()) o << "/error";
+    return o << ")";;
+}
 
+MemberId Cluster::getId() const {
+    return self;            // Immutable, no need to lock.
+}
 
+broker::Broker& Cluster::getBroker() const {
+    return broker; // Immutable,  no need to lock.
+}
+
+void Cluster::setClusterId(const Uuid& uuid, Lock&) {
+    clusterId = uuid;
+    if (mgmtObject) {
+        stringstream stream;
+        stream << self;
+        mgmtObject->set_clusterID(clusterId.str());
+        mgmtObject->set_memberID(stream.str());
+    }
+    QPID_LOG(debug, *this << " cluster-uuid = " << clusterId);
+}
+
+void Cluster::messageExpired(const MemberId&, uint64_t id, Lock&) {
+    expiryPolicy->deliverExpire(id);
+}
+
+void Cluster::errorCheck(const MemberId& from, uint8_t type, framing::SequenceNumber frameSeq, Lock&) {
+    // If we see an errorCheck here (rather than in the ErrorCheck
+    // class) then we have processed succesfully past the point of the
+    // error.
+    if (state >= CATCHUP) // Don't respond pre catchup, we don't know what happened
+        error.respondNone(from, type, frameSeq);
+}
 
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index 2b40193dd3..7872588307 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -19,151 +19,261 @@
  *
  */
 
-#include "qpid/cluster/Cpg.h"
-#include "qpid/cluster/ShadowConnectionOutputHandler.h"
-#include "qpid/cluster/PollableQueue.h"
-
+#include "ClusterMap.h"
+#include "ClusterSettings.h"
+#include "Cpg.h"
+#include "Decoder.h"
+#include "ErrorCheck.h"
+#include "Event.h"
+#include "EventFrame.h"
+#include "ExpiryPolicy.h"
+#include "FailoverExchange.h"
+#include "InitialStatusMap.h"
+#include "LockedConnectionMap.h"
+#include "Multicaster.h"
+#include "NoOpConnectionOutputHandler.h"
+#include "PollableQueue.h"
+#include "PollerDispatch.h"
+#include "Quorum.h"
+#include "StoreStatus.h"
+#include "UpdateReceiver.h"
+
+#include "qmf/org/apache/qpid/cluster/Cluster.h"
+#include "qpid/Url.h"
 #include "qpid/broker/Broker.h"
-#include "qpid/broker/Connection.h"
-#include "qpid/sys/Dispatcher.h"
+#include "qpid/management/Manageable.h"
 #include "qpid/sys/Monitor.h"
-#include "qpid/sys/Runnable.h"
-#include "qpid/sys/Thread.h"
-#include "qpid/log/Logger.h"
-#include "qpid/Url.h"
-#include "qpid/RefCounted.h"
 
-#include <boost/optional.hpp>
-#include <boost/function.hpp>
+#include <boost/bind.hpp>
 #include <boost/intrusive_ptr.hpp>
+#include <boost/optional.hpp>
 
+#include <algorithm>
 #include <map>
 #include <vector>
 
 namespace qpid {
+
+namespace framing {
+class AMQBody;
+class Uuid;
+}
+
 namespace cluster {
 
-class ConnectionInterceptor;
+class Connection;
+class EventFrame;
 
 /**
- * Connection to the cluster.
- * Keeps cluster membership data.
+ * Connection to the cluster
  */
-class Cluster : private Cpg::Handler, public RefCounted
-{
+class Cluster : private Cpg::Handler, public management::Manageable {
   public:
-    typedef boost::tuple<Cpg::Id, void*> ShadowConnectionId;
+    typedef boost::intrusive_ptr<Connection> ConnectionPtr;
+    typedef std::vector<ConnectionPtr> ConnectionVector;
 
-    /** Details of a cluster member */
-    struct Member {
-        Cpg::Id  id;
-        Url url;
-    };
-    
-    typedef std::vector<Member> MemberList;
-
-    /**
-     * Join a cluster.
-     * @param name of the cluster.
-     * @param url of this broker, sent to the cluster.
-     */
-    Cluster(const std::string& name, const Url& url, broker::Broker&);
+    // Public functions are thread safe unless otherwise mentioned in a comment.
 
+    // Construct the cluster in plugin earlyInitialize.
+    Cluster(const ClusterSettings&, broker::Broker&);
     virtual ~Cluster();
 
-    /** Initialize interceptors for a new connection */
-    void initialize(broker::Connection&);
-    
-    /** Get the current cluster membership. */
-    MemberList getMembers() const;
-
-    /** Number of members in the cluster. */
-    size_t size() const;
+    // Called by plugin initialize: cluster start-up requires transport plugins .
+    // Thread safety: only called by plugin initialize.
+    void initialize();
 
-    bool empty() const { return size() == 0; }
+    // Connection map.
+    void addLocalConnection(const ConnectionPtr&); 
+    void addShadowConnection(const ConnectionPtr&); 
+    void erase(const ConnectionId&);       
     
-    /** Send frame to the cluster */
-    void send(const framing::AMQFrame&, ConnectionInterceptor*);
+    // URLs of current cluster members.
+    std::vector<std::string> getIds() const;
+    std::vector<Url> getUrls() const;
+    boost::shared_ptr<FailoverExchange> getFailoverExchange() const { return failoverExchange; }
 
-    /** Leave the cluster */
+    // Leave the cluster - called when fatal errors occur.
     void leave();
-    
-    // Cluster frame handing functions
-    void notify(const std::string& url);
-    void connectionClose();
+
+    // Update completed - called in update thread
+    void updateInDone(const ClusterMap&);
+    void updateInRetracted();
+
+    MemberId getId() const;
+    broker::Broker& getBroker() const;
+    Multicaster& getMulticast() { return mcast; }
+
+    const ClusterSettings& getSettings() const { return settings; }
+
+    void deliverFrame(const EventFrame&);
+
+    // Called in deliverFrame thread to indicate an error from the broker.
+    void flagError(Connection&, ErrorCheck::ErrorType, const std::string& msg);
+
+    // Called only during update by Connection::shadowReady
+    Decoder& getDecoder() { return decoder; }
+
+    ExpiryPolicy& getExpiryPolicy() { return *expiryPolicy; }
+
+    UpdateReceiver& getUpdateReceiver() { return updateReceiver; }
 
   private:
-    typedef Cpg::Id Id;
-    typedef std::map<Id, Member>  MemberMap;
-    typedef std::map<ShadowConnectionId, ConnectionInterceptor*> ShadowConnectionMap;
-    typedef std::set<ConnectionInterceptor*> LocalConnectionSet;
-
-    /** Message sent over the cluster. */
-    struct Message {
-        framing::AMQFrame frame; Id from; void* connection;
-        Message(const framing::AMQFrame& f, const Id i, void* c)
-            : frame(f), from(i), connection(c) {}
-    };
-    typedef PollableQueue<Message> MessageQueue;
-
-    boost::function<void()> shutdownNext;
-    
-    void notify();              ///< Notify cluster of my details.
+    typedef sys::Monitor::ScopedLock Lock;
 
-    /** CPG deliver callback. */
-    void deliver(
+    typedef PollableQueue<Event> PollableEventQueue;
+    typedef PollableQueue<EventFrame> PollableFrameQueue;
+    typedef std::map<ConnectionId, ConnectionPtr> ConnectionMap;
+
+    /** Version number of the cluster protocol, to avoid mixed versions. */
+    static const uint32_t CLUSTER_VERSION;
+    
+    // NB: A dummy Lock& parameter marks functions that must only be
+    // called with Cluster::lock  locked.
+ 
+    void leave(Lock&);
+    std::vector<std::string> getIds(Lock&) const;
+    std::vector<Url> getUrls(Lock&) const;
+
+    // == Called in main thread from Broker destructor.
+    void brokerShutdown();
+
+    // == Called in deliverEventQueue thread
+    void deliveredEvent(const Event&); 
+
+    // == Called in deliverFrameQueue thread
+    void deliveredFrame(const EventFrame&); 
+    void processFrame(const EventFrame&, Lock&); 
+
+    // Cluster controls implement XML methods from cluster.xml.
+    void updateRequest(const MemberId&, const std::string&, Lock&);
+    void updateOffer(const MemberId& updater, uint64_t updatee, Lock&);
+    void retractOffer(const MemberId& updater, uint64_t updatee, Lock&);
+    void initialStatus(const MemberId&,
+                       uint32_t version,
+                       bool active,
+                       const framing::Uuid& clusterId,
+                       framing::cluster::StoreState,
+                       const framing::Uuid& shutdownId,
+                       Lock&);
+    void ready(const MemberId&, const std::string&, Lock&);
+    void configChange(const MemberId&, const std::string& current, Lock& l);
+    void messageExpired(const MemberId&, uint64_t, Lock& l);
+    void errorCheck(const MemberId&, uint8_t type, SequenceNumber frameSeq, Lock&);
+
+    void shutdown(const MemberId&, const framing::Uuid& shutdownId, Lock&);
+
+    // Helper functions
+    ConnectionPtr getConnection(const EventFrame&, Lock&);
+    ConnectionVector getConnections(Lock&);
+    void updateStart(const MemberId& updatee, const Url& url, Lock&);
+    void makeOffer(const MemberId&, Lock&);
+    void setReady(Lock&);
+    void memberUpdate(Lock&);
+    void setClusterId(const framing::Uuid&, Lock&);
+    void erase(const ConnectionId&, Lock&);       
+
+    void initMapCompleted(Lock&);
+
+
+
+    // == Called in CPG dispatch thread
+    void deliver( // CPG deliver callback. 
         cpg_handle_t /*handle*/,
-        struct cpg_name *group,
+        const struct cpg_name *group,
         uint32_t /*nodeid*/,
         uint32_t /*pid*/,
         void* /*msg*/,
         int /*msg_len*/);
 
-    /** CPG config change callback */
-    void configChange(
+    void deliverEvent(const Event&);
+    
+    void configChange( // CPG config change callback.
         cpg_handle_t /*handle*/,
-        struct cpg_name */*group*/,
-        struct cpg_address */*members*/, int /*nMembers*/,
-        struct cpg_address */*left*/, int /*nLeft*/,
-        struct cpg_address */*joined*/, int /*nJoined*/
+        const struct cpg_name */*group*/,
+        const struct cpg_address */*members*/, int /*nMembers*/,
+        const struct cpg_address */*left*/, int /*nLeft*/,
+        const struct cpg_address */*joined*/, int /*nJoined*/
     );
 
-    /** Callback to handle delivered frames from the deliverQueue. */
-    void deliverQueueCb(const MessageQueue::iterator& begin,
-                      const MessageQueue::iterator& end);
-
-    /** Callback to multi-cast frames from mcastQueue */
-    void mcastQueueCb(const MessageQueue::iterator& begin,
-                    const MessageQueue::iterator& end);
+    // == Called in management threads.
+    virtual qpid::management::ManagementObject* GetManagementObject() const;
+    virtual management::Manageable::status_t ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
 
+    void stopClusterNode(Lock&);
+    void stopFullCluster(Lock&);
 
-    /** Callback to dispatch CPG events. */
-    void dispatch(sys::DispatchHandle&);
-    /** Callback if CPG fd is disconnected. */
-    void disconnect(sys::DispatchHandle&);
+    // == Called in connection IO threads .
+    void checkUpdateIn(Lock&);
 
-    void handleMethod(Id from, ConnectionInterceptor* connection, framing::AMQMethodBody& method);
+    // == Called in UpdateClient thread.
+    void updateOutDone();
+    void updateOutError(const std::exception&);
+    void updateOutDone(Lock&);
 
-    ConnectionInterceptor* getShadowConnection(const Cpg::Id&, void*);
-
-    mutable sys::Monitor lock;  // Protect access to members.
-    broker::Broker* broker;
+    // Immutable members set on construction, never changed.
+    const ClusterSettings settings;
+    broker::Broker& broker;
+    qmf::org::apache::qpid::cluster::Cluster* mgmtObject; // mgnt owns lifecycle
     boost::shared_ptr<sys::Poller> poller;
     Cpg cpg;
-    Cpg::Name name;
-    Url url;
-    MemberMap members;
-    Id self;
-    ShadowConnectionMap shadowConnectionMap;
-    LocalConnectionSet localConnectionSet;
-    ShadowConnectionOutputHandler shadowOut;
-    sys::DispatchHandle cpgDispatchHandle;
-    MessageQueue deliverQueue;
-    MessageQueue mcastQueue;
-
-  friend std::ostream& operator <<(std::ostream&, const Cluster&);
-  friend std::ostream& operator <<(std::ostream&, const MemberMap::value_type&);
-  friend std::ostream& operator <<(std::ostream&, const MemberMap&);
+    const std::string name;
+    Url myUrl;
+    const MemberId self;
+    framing::Uuid clusterId;
+    NoOpConnectionOutputHandler shadowOut;
+    qpid::management::ManagementAgent* mAgent;
+    boost::intrusive_ptr<ExpiryPolicy> expiryPolicy;
+
+    // Thread safe members
+    Multicaster mcast;
+    PollerDispatch dispatcher;
+    PollableEventQueue deliverEventQueue;
+    PollableFrameQueue deliverFrameQueue;
+    boost::shared_ptr<FailoverExchange> failoverExchange;
+    Quorum quorum;
+    LockedConnectionMap localConnections;
+
+    // Used only in deliverEventQueue thread or when stalled for update.
+    Decoder decoder;
+    bool discarding;
+    
+
+    // Remaining members are protected by lock.
+
+    // TODO aconway 2009-03-06: Most of these members are also only used in
+    // deliverFrameQueue thread or during stall. Review and separate members
+    // that require a lock, drop lock when not needed.
+
+    mutable sys::Monitor lock;
+
+
+    //    Local cluster state, cluster map
+    enum {
+        INIT,    ///< Establishing inital cluster stattus.
+        JOINER,  ///< Sent update request, waiting for update offer.
+        UPDATEE, ///< Stalled receive queue at update offer, waiting for update to complete.
+        CATCHUP, ///< Update complete, unstalled but has not yet seen own "ready" event.
+        READY,   ///< Fully operational 
+        OFFER,   ///< Sent an offer, waiting for accept/reject.
+        UPDATER, ///< Offer accepted, sending a state update.
+        LEFT     ///< Final state, left the cluster.
+    } state;
+
+    ConnectionMap connections;
+    InitialStatusMap initMap;
+    StoreStatus store;
+    ClusterMap map;
+    MemberSet elders;
+    size_t lastSize;
+    bool lastBroker;
+    sys::Thread updateThread;
+    boost::optional<ClusterMap> updatedMap;
+    bool updateRetracted;
+    ErrorCheck error;
+    UpdateReceiver updateReceiver;
+
+  friend std::ostream& operator<<(std::ostream&, const Cluster&);
+  friend class ClusterDispatcher;
 };
 
 }} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ClusterMap.cpp b/cpp/src/qpid/cluster/ClusterMap.cpp
new file mode 100644
index 0000000000..8cac470ef3
--- /dev/null
+++ b/cpp/src/qpid/cluster/ClusterMap.cpp
@@ -0,0 +1,175 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/ClusterMap.h"
+#include "qpid/Url.h"
+#include "qpid/framing/FieldTable.h"
+#include "qpid/log/Statement.h"
+#include <boost/bind.hpp>
+#include <algorithm>
+#include <functional>
+#include <iterator>
+#include <ostream>
+
+using namespace std;
+using namespace boost;
+
+namespace qpid {
+using namespace framing;
+
+namespace cluster {
+
+namespace {
+
+void addFieldTableValue(FieldTable::ValueMap::value_type vt, ClusterMap::Map& map, ClusterMap::Set& set) {
+    MemberId id(vt.first);
+    set.insert(id);
+    string url = vt.second->get<string>();
+    if (!url.empty())
+        map.insert(ClusterMap::Map::value_type(id, Url(url)));
+}
+
+void insertFieldTableFromMapValue(FieldTable& ft, const ClusterMap::Map::value_type& vt) {
+    ft.setString(vt.first.str(), vt.second.str());
+}
+
+void assignFieldTable(FieldTable& ft, const ClusterMap::Map& map) {
+    ft.clear();
+    for_each(map.begin(), map.end(), bind(&insertFieldTableFromMapValue, ref(ft), _1));
+}
+
+}
+
+ClusterMap::ClusterMap() : frameSeq(0) {}
+
+ClusterMap::ClusterMap(const Map& map) : frameSeq(0) {
+    transform(map.begin(), map.end(), inserter(alive, alive.begin()), bind(&Map::value_type::first, _1));
+    members = map;
+}
+
+ClusterMap::ClusterMap(const FieldTable& joinersFt, const FieldTable& membersFt, framing::SequenceNumber frameSeq_)
+  : frameSeq(frameSeq_)
+{
+    for_each(joinersFt.begin(), joinersFt.end(), bind(&addFieldTableValue, _1, ref(joiners), ref(alive)));
+    for_each(membersFt.begin(), membersFt.end(), bind(&addFieldTableValue, _1, ref(members), ref(alive)));
+}
+
+void ClusterMap::toMethodBody(framing::ClusterConnectionMembershipBody& b) const {
+    b.getJoiners().clear();
+    for_each(joiners.begin(), joiners.end(), bind(&insertFieldTableFromMapValue, ref(b.getJoiners()), _1));
+    for(Set::const_iterator i = alive.begin(); i != alive.end(); ++i) {
+        if (!isMember(*i) && !isJoiner(*i))
+            b.getJoiners().setString(i->str(), string());
+    }
+    b.getMembers().clear();
+    for_each(members.begin(), members.end(), bind(&insertFieldTableFromMapValue, ref(b.getMembers()), _1));
+    b.setFrameSeq(frameSeq);
+}
+
+Url ClusterMap::getUrl(const Map& map, const  MemberId& id) {
+    Map::const_iterator i = map.find(id);
+    return i == map.end() ? Url() : i->second;
+}
+     
+MemberId ClusterMap::firstJoiner() const {
+    return joiners.empty() ? MemberId() : joiners.begin()->first;
+}
+
+vector<string> ClusterMap::memberIds() const {
+    vector<string> ids;
+    for (Map::const_iterator iter = members.begin();
+         iter != members.end(); iter++) {
+        stringstream stream;
+        stream << iter->first;
+        ids.push_back(stream.str());
+    }
+    return ids;
+}
+
+vector<Url> ClusterMap::memberUrls() const {
+    vector<Url> urls(members.size());
+    transform(members.begin(), members.end(), urls.begin(),
+                   bind(&Map::value_type::second, _1));
+    return urls;
+}
+
+ClusterMap::Set ClusterMap::getAlive() const { return alive; }
+
+ClusterMap::Set ClusterMap::getMembers() const {
+    Set s;
+    transform(members.begin(), members.end(), inserter(s, s.begin()),
+                   bind(&Map::value_type::first, _1));
+    return s;
+}
+
+ostream& operator<<(ostream& o, const ClusterMap::Map& m) {
+    ostream_iterator<MemberId> oi(o);
+    transform(m.begin(), m.end(), oi, bind(&ClusterMap::Map::value_type::first, _1));
+    return o;
+}
+
+ostream& operator<<(ostream& o, const ClusterMap& m) {
+    for (ClusterMap::Set::const_iterator i = m.alive.begin(); i != m.alive.end(); ++i) {
+        o << *i;
+        if (m.isMember(*i)) o << "(member)";
+        else if (m.isJoiner(*i)) o << "(joiner)";
+        else o << "(unknown)";
+        o << " ";
+    }
+    return o;
+}
+
+bool ClusterMap::updateRequest(const MemberId& id, const string& url) {
+    if (isAlive(id)) {
+        joiners[id] = Url(url);
+        return true;
+    }
+    return false;
+}
+
+bool ClusterMap::ready(const MemberId& id, const Url& url) {
+    return isAlive(id) &&  members.insert(Map::value_type(id,url)).second;
+}
+
+bool ClusterMap::configChange(const Set& update) {
+    bool memberChange = false;
+    Set removed;
+    set_difference(alive.begin(), alive.end(),
+                        update.begin(), update.end(),
+                        inserter(removed, removed.begin()));
+    alive = update;
+    for (Set::const_iterator i = removed.begin(); i != removed.end(); ++i) {
+        memberChange = memberChange || members.erase(*i);
+        joiners.erase(*i);
+    }
+    return memberChange;
+}
+
+optional<Url> ClusterMap::updateOffer(const MemberId& from, const MemberId& to) {
+    Map::iterator i = joiners.find(to);
+    if (isAlive(from) && i != joiners.end()) {
+        Url url= i->second;
+        joiners.erase(i);       // No longer a potential updatee.
+        return url;
+    }
+    return optional<Url>();
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ClusterMap.h b/cpp/src/qpid/cluster/ClusterMap.h
new file mode 100644
index 0000000000..98572813a8
--- /dev/null
+++ b/cpp/src/qpid/cluster/ClusterMap.h
@@ -0,0 +1,105 @@
+#ifndef QPID_CLUSTER_CLUSTERMAP_H
+#define QPID_CLUSTER_CLUSTERMAP_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "MemberSet.h"
+#include "qpid/Url.h"
+#include "qpid/framing/ClusterConnectionMembershipBody.h"
+#include "qpid/framing/SequenceNumber.h"
+
+#include <boost/function.hpp>
+#include <boost/optional.hpp>
+
+#include <vector>
+#include <deque>
+#include <map>
+#include <iosfwd>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * Map of established cluster members and joiners waiting for an update,
+ * along with other cluster state that must be updated.
+ */
+class ClusterMap {
+  public:
+    typedef std::map<MemberId, Url> Map;
+    typedef std::set<MemberId> Set;
+
+    ClusterMap();
+    ClusterMap(const Map& map);
+    ClusterMap(const framing::FieldTable& joiners, const framing::FieldTable& members, framing::SequenceNumber frameSeq);
+
+    /** Update from config change.
+     *@return true if member set changed.
+     */
+    bool configChange(const Set& members);
+
+    bool isJoiner(const MemberId& id) const { return joiners.find(id) != joiners.end(); }
+    bool isMember(const MemberId& id) const { return members.find(id) != members.end(); }
+    bool isAlive(const MemberId& id) const { return alive.find(id) != alive.end(); }
+    
+    Url getJoinerUrl(const MemberId& id) { return getUrl(joiners, id); }
+    Url getMemberUrl(const MemberId& id) { return getUrl(members, id); }
+
+    /** First joiner in the cluster in ID order, target for offers */
+    MemberId firstJoiner() const;
+
+    /** Convert map contents to a cluster control body. */
+    void toMethodBody(framing::ClusterConnectionMembershipBody&) const;
+
+    size_t aliveCount() const { return alive.size(); }
+    size_t memberCount() const { return members.size(); }
+    std::vector<std::string> memberIds() const;
+    std::vector<Url> memberUrls() const;
+    Set getAlive() const;
+    Set getMembers() const;
+
+    bool updateRequest(const MemberId& id, const std::string& url);       
+    /** Return non-empty Url if accepted */
+    boost::optional<Url> updateOffer(const MemberId& from, const MemberId& to);
+
+    /**@return true If this is a new member */ 
+    bool ready(const MemberId& id, const Url&);
+
+    framing::SequenceNumber getFrameSeq() { return frameSeq; }
+    framing::SequenceNumber incrementFrameSeq() { return ++frameSeq; }
+    
+    /** Clear out all knowledge of joiners & members, just keep alive set */
+    void clearStatus() { joiners.clear(); members.clear(); }
+    
+  private:
+    Url getUrl(const Map& map, const  MemberId& id);
+    
+    Map joiners, members;
+    Set alive;
+    framing::SequenceNumber frameSeq;
+
+  friend std::ostream& operator<<(std::ostream&, const Map&);
+  friend std::ostream& operator<<(std::ostream&, const ClusterMap&);
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_CLUSTERMAP_H*/
diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp
index 1d07660455..e4aee6730b 100644
--- a/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -16,89 +16,157 @@
  *
  */
 
-#include "ConnectionInterceptor.h"
+#include "config.h"
+#include "qpid/cluster/Connection.h"
+#include "qpid/cluster/ConnectionCodec.h"
+#include "qpid/cluster/ClusterSettings.h"
 
-#include "qpid/broker/Broker.h"
 #include "qpid/cluster/Cluster.h"
+#include "qpid/cluster/ConnectionCodec.h"
+#include "qpid/cluster/UpdateClient.h"
+
+#include "qpid/broker/Broker.h"
 #include "qpid/Plugin.h"
 #include "qpid/Options.h"
-#include "qpid/shared_ptr.h"
-
+#include "qpid/sys/AtomicValue.h"
+#include "qpid/log/Statement.h"
+
+#include "qpid/management/ManagementAgent.h"
+#include "qpid/management/IdAllocator.h"
+#include "qpid/broker/Exchange.h"
+#include "qpid/broker/Message.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/SessionState.h"
+#include "qpid/client/ConnectionSettings.h"
+
+#include <boost/shared_ptr.hpp>
 #include <boost/utility/in_place_factory.hpp>
+#include <boost/scoped_ptr.hpp>
 
 namespace qpid {
 namespace cluster {
 
 using namespace std;
+using broker::Broker;
+using management::IdAllocator;
+using management::ManagementAgent;
 
-struct ClusterValues {
-    string name;
-    string url;
-
-    Url getUrl(uint16_t port) const {
-        if (url.empty()) return Url::getIpAddressesUrl(port);
-        return Url(url);
-    }
-};
 
-/** Note separating options from values to work around boost version differences.
+/** Note separating options from settings to work around boost version differences.
  *  Old boost takes a reference to options objects, but new boost makes a copy.
  *  New boost allows a shared_ptr but that's not compatible with old boost.
  */
 struct ClusterOptions : public Options {
-    ClusterValues& values; 
+    ClusterSettings& settings; 
 
-    ClusterOptions(ClusterValues& v) : Options("Cluster Options"), values(v) {
+    ClusterOptions(ClusterSettings& v) : Options("Cluster Options"), settings(v) {
         addOptions()
-            ("cluster-name", optValue(values.name, "NAME"), "Name of cluster to join")
-            ("cluster-url", optValue(values.url,"URL"),
+            ("cluster-name", optValue(settings.name, "NAME"), "Name of cluster to join")
+            ("cluster-url", optValue(settings.url,"URL"),
              "URL of this broker, advertized to the cluster.\n"
              "Defaults to a URL listing all the local IP addresses\n")
+            ("cluster-username", optValue(settings.username, "USER"), "Username for connections between brokers")
+            ("cluster-password", optValue(settings.password, "PASS"), "Password for connections between brokers")
+            ("cluster-mechanism", optValue(settings.mechanism, "MECH"), "Authentication mechanism for connections between brokers")
+#if HAVE_LIBCMAN_H
+            ("cluster-cman", optValue(settings.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
+#endif
+            ("cluster-size", optValue(settings.size, "N"), "Wait for N cluster members before allowing clients to connect.")
+            ("cluster-read-max", optValue(settings.readMax,"N"), "Experimental: flow-control limit  reads per connection. 0=no limit.")
             ;
     }
 };
 
+struct UpdateClientIdAllocator : management::IdAllocator
+{
+    qpid::sys::AtomicValue<uint64_t> sequence;
+
+    UpdateClientIdAllocator() : sequence(0x4000000000000000LL) {}
+
+    uint64_t getIdFor(management::Manageable* m)
+    {
+        if (isUpdateQueue(m) || isUpdateExchange(m) || isUpdateSession(m) || isUpdateBinding(m)) {
+            return ++sequence;
+        } else {
+            return 0;
+        }
+    }
+
+    bool isUpdateQueue(management::Manageable* manageable)
+    {
+        qpid::broker::Queue* queue = dynamic_cast<qpid::broker::Queue*>(manageable);
+        return queue && queue->getName() == UpdateClient::UPDATE;
+    }
+
+    bool isUpdateExchange(management::Manageable* manageable)
+    {
+        qpid::broker::Exchange* exchange = dynamic_cast<qpid::broker::Exchange*>(manageable);
+        return exchange && exchange->getName() == UpdateClient::UPDATE;
+    }
+
+    bool isUpdateSession(management::Manageable* manageable)
+    {
+        broker::SessionState* session = dynamic_cast<broker::SessionState*>(manageable);
+        return session && session->getId().getName() == UpdateClient::UPDATE;
+    }
+
+    bool isUpdateBinding(management::Manageable* manageable)
+    {
+        broker::Exchange::Binding* binding = dynamic_cast<broker::Exchange::Binding*>(manageable);
+        return binding && binding->queue->getName() == UpdateClient::UPDATE;
+    }
+};
+
 struct ClusterPlugin : public Plugin {
 
-    ClusterValues values;
+    ClusterSettings settings;
     ClusterOptions options;
-    boost::intrusive_ptr<Cluster> cluster;
+    Cluster* cluster;
+    boost::scoped_ptr<ConnectionCodec::Factory> factory;
 
-    ClusterPlugin() : options(values) {}
+    ClusterPlugin() : options(settings), cluster(0) {}
 
+    // Cluster needs to be initialized after the store 
+    int initOrder() const { return Plugin::DEFAULT_INIT_ORDER+500; }
+    
     Options* getOptions() { return &options; }
 
-    void init(broker::Broker& b) {
-        if (values.name.empty()) return;  // Only if --cluster-name option was specified.
-        if (cluster) throw Exception("Cluster plugin cannot be initialized twice in one process.");
-        cluster = new Cluster(values.name, values.getUrl(b.getPort()), b);
-        b.addFinalizer(boost::bind(&ClusterPlugin::shutdown, this));
+    void earlyInitialize(Plugin::Target& target) {
+        if (settings.name.empty()) return; // Only if --cluster-name option was specified.
+        Broker* broker = dynamic_cast<Broker*>(&target);
+        if (!broker) return;
+        cluster = new Cluster(settings, *broker);
+        broker->setConnectionFactory(
+            boost::shared_ptr<sys::ConnectionCodec::Factory>(
+                new ConnectionCodec::Factory(broker->getConnectionFactory(), *cluster)));
+        ManagementAgent* mgmt = broker->getManagementAgent();
+        if (mgmt) {
+            std::auto_ptr<IdAllocator> allocator(new UpdateClientIdAllocator());
+            mgmt->setAllocator(allocator);
+        }
     }
 
-    template <class T> void init(T& t) {
-        if (cluster) cluster->initialize(t);
+    void disallow(ManagementAgent* agent, const string& className, const string& methodName) {
+        string message = "Management method " + className + ":" + methodName + " is not allowed on a clustered broker.";
+        agent->disallow(className, methodName, message);
     }
-    
-    template <class T> bool init(Plugin::Target& target) {
-        T* t = dynamic_cast<T*>(&target);
-        if (t) init(*t);
-        return t;
+    void disallowManagementMethods(ManagementAgent* agent) {
+        if (!agent) return;
+        disallow(agent, "queue", "purge");
+        disallow(agent, "session", "detach");
+        disallow(agent, "session", "close");
+        disallow(agent, "connection", "close");
     }
 
-    void earlyInitialize(Plugin::Target&) {}
-
     void initialize(Plugin::Target& target) {
-        if (init<broker::Broker>(target)) return;
-        if (!cluster) return;   // Remaining plugins only valid if cluster initialized.
-        if (init<broker::Connection>(target)) return;
+        Broker* broker = dynamic_cast<Broker*>(&target);
+        if (broker && cluster) {
+            disallowManagementMethods(broker->getManagementAgent());
+            cluster->initialize();
+        }
     }
-
-    void shutdown() { cluster = 0; }
 };
 
 static ClusterPlugin instance; // Static initialization.
 
-// For test purposes.
-boost::intrusive_ptr<Cluster> getGlobalCluster() { return instance.cluster; }
-    
 }} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ClusterSettings.h b/cpp/src/qpid/cluster/ClusterSettings.h
new file mode 100644
index 0000000000..d37c7792bf
--- /dev/null
+++ b/cpp/src/qpid/cluster/ClusterSettings.h
@@ -0,0 +1,50 @@
+#ifndef QPID_CLUSTER_CLUSTERSETTINGS_H
+#define QPID_CLUSTER_CLUSTERSETTINGS_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <qpid/Url.h>
+#include <string>
+
+namespace qpid {
+namespace cluster {
+
+struct ClusterSettings {
+    std::string name;
+    std::string url;
+    bool quorum;
+    size_t readMax;
+    std::string username, password, mechanism;
+    size_t size;
+
+    ClusterSettings() : quorum(false), readMax(10), mechanism("ANONYMOUS"), size(1)
+    {}
+  
+    Url getUrl(uint16_t port) const {
+        if (url.empty()) return Url::getIpAddressesUrl(port);
+        return Url(url);
+    }
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_CLUSTERSETTINGS_H*/
diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
new file mode 100644
index 0000000000..d223244f15
--- /dev/null
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -0,0 +1,479 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "Connection.h"
+#include "UpdateClient.h"
+#include "Cluster.h"
+#include "UpdateReceiver.h"
+
+#include "qpid/broker/SessionState.h"
+#include "qpid/broker/SemanticState.h"
+#include "qpid/broker/TxBuffer.h"
+#include "qpid/broker/TxPublish.h"
+#include "qpid/broker/TxAccept.h"
+#include "qpid/broker/RecoveredEnqueue.h"
+#include "qpid/broker/RecoveredDequeue.h"
+#include "qpid/broker/Exchange.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/framing/enum.h"
+#include "qpid/framing/AMQFrame.h"
+#include "qpid/framing/AllInvoker.h"
+#include "qpid/framing/DeliveryProperties.h"
+#include "qpid/framing/ClusterConnectionDeliverCloseBody.h"
+#include "qpid/framing/ConnectionCloseBody.h"
+#include "qpid/framing/ConnectionCloseOkBody.h"
+#include "qpid/log/Statement.h"
+
+#include <boost/current_function.hpp>
+
+// TODO aconway 2008-11-03:
+// 
+// Refactor code for receiving an update into a separate UpdateConnection
+// class.
+//
+
+
+namespace qpid {
+namespace cluster {
+
+using namespace framing;
+using namespace framing::cluster;
+
+qpid::sys::AtomicValue<uint64_t> Connection::catchUpId(0x5000000000000000LL);
+
+Connection::NullFrameHandler Connection::nullFrameHandler;
+
+struct NullFrameHandler : public framing::FrameHandler {
+    void handle(framing::AMQFrame&) {}
+};
+
+
+namespace {
+sys::AtomicValue<uint64_t> idCounter;
+const std::string shadowPrefix("[shadow]");
+}
+
+
+// Shadow connection
+    Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out, const std::string& logId,
+                           const ConnectionId& id, unsigned int ssf)
+    : cluster(c), self(id), catchUp(false), output(*this, out),
+      connection(&output, cluster.getBroker(), shadowPrefix+logId, ssf), expectProtocolHeader(false),
+      mcastFrameHandler(cluster.getMulticast(), self),
+      consumerNumbering(c.getUpdateReceiver().consumerNumbering)
+{ init(); }
+
+// Local connection
+Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
+                       const std::string& logId, MemberId member,
+                       bool isCatchUp, bool isLink, unsigned int ssf
+) : cluster(c), self(member, ++idCounter), catchUp(isCatchUp), output(*this, out),
+    connection(&output, cluster.getBroker(),
+               isCatchUp ? shadowPrefix+logId : logId,
+               ssf,
+               isLink,
+               isCatchUp ? ++catchUpId : 0),
+    expectProtocolHeader(isLink), mcastFrameHandler(cluster.getMulticast(), self),
+    consumerNumbering(c.getUpdateReceiver().consumerNumbering)
+{ init(); }
+
+void Connection::init() {
+    QPID_LOG(debug, cluster << " new connection: " << *this);
+    if (isLocalClient()) {  
+        connection.setClusterOrderOutput(mcastFrameHandler); // Actively send cluster-order frames from local node
+        cluster.addLocalConnection(this);
+        giveReadCredit(cluster.getSettings().readMax);
+    }
+    else {                                                  // Shadow or catch-up connection
+        connection.setClusterOrderOutput(nullFrameHandler); // Passive, discard cluster-order frames
+        connection.setClientThrottling(false);              // Disable client throttling, done by active node.
+        connection.setShadow(); // Mark the broker connection as a shadow.
+    }
+    if (!isCatchUp())
+        connection.setErrorListener(this);
+}
+
+void Connection::giveReadCredit(int credit) {
+    if (cluster.getSettings().readMax && credit) 
+        output.giveReadCredit(credit);
+}
+
+Connection::~Connection() {
+    connection.setErrorListener(0);
+    QPID_LOG(debug, cluster << " deleted connection: " << *this);
+}
+
+bool Connection::doOutput() {
+    return output.doOutput();
+}
+
+// Received from a directly connected client.
+void Connection::received(framing::AMQFrame& f) {
+    QPID_LOG(trace, cluster << " RECV " << *this << ": " << f);
+    if (isLocal()) {            // Local catch-up connection.
+        currentChannel = f.getChannel();
+        if (!framing::invoke(*this, *f.getBody()).wasHandled())
+            connection.received(f);
+    }
+    else {             // Shadow or updated catch-up connection.
+        if (f.getMethod() && f.getMethod()->isA<ConnectionCloseBody>()) {
+            if (isShadow()) 
+                cluster.addShadowConnection(this);
+            AMQFrame ok((ConnectionCloseOkBody()));
+            connection.getOutput().send(ok);
+            output.closeOutput();
+            catchUp = false;
+        }
+        else
+            QPID_LOG(warning, cluster << " ignoring unexpected frame " << *this << ": " << f);
+    }
+}
+
+bool Connection::checkUnsupported(const AMQBody& body) {
+    std::string message;
+    if (body.getMethod()) {
+        switch (body.getMethod()->amqpClassId()) {
+          case DTX_CLASS_ID: message = "DTX transactions are not currently supported by cluster."; break;
+        }
+    }
+    if (!message.empty())
+        connection.close(connection::CLOSE_CODE_FRAMING_ERROR, message);
+    return !message.empty();
+}
+
+struct GiveReadCreditOnExit {
+    Connection& connection;
+    int credit;
+    GiveReadCreditOnExit(Connection& connection_, int credit_) :
+        connection(connection_), credit(credit_) {}
+    ~GiveReadCreditOnExit() { connection.giveReadCredit(credit); }
+};
+
+// Called in delivery thread, in cluster order.
+void Connection::deliveredFrame(const EventFrame& f) {
+    GiveReadCreditOnExit gc(*this, f.readCredit);
+    assert(!catchUp);
+    currentChannel = f.frame.getChannel(); 
+    if (f.frame.getBody()       // frame can be emtpy with just readCredit
+        && !framing::invoke(*this, *f.frame.getBody()).wasHandled() // Connection contol.
+        && !checkUnsupported(*f.frame.getBody())) // Unsupported operation.
+    {
+        if (f.type == DATA) // incoming data frames to broker::Connection
+            connection.received(const_cast<AMQFrame&>(f.frame)); 
+        else {           // frame control, send frame via SessionState
+            broker::SessionState* ss = connection.getChannel(currentChannel).getSession();
+            if (ss) ss->out(const_cast<AMQFrame&>(f.frame));
+        }
+    }
+}
+
+// A local connection is closed by the network layer.
+void Connection::closed() {
+    try {
+        if (catchUp) {
+            QPID_LOG(critical, cluster << " catch-up connection closed prematurely " << *this);
+            cluster.leave();
+        }
+        else if (isUpdated()) {
+            QPID_LOG(debug, cluster << " closed update connection " << *this);
+            connection.closed();
+        }
+        else if (isLocal()) {
+            QPID_LOG(debug, cluster << " local close of replicated connection " << *this);
+            // This was a local replicated connection. Multicast a deliver
+            // closed and process any outstanding frames from the cluster
+            // until self-delivery of deliver-close.
+            output.closeOutput();
+            cluster.getMulticast().mcastControl(ClusterConnectionDeliverCloseBody(), self);
+        }
+    }
+    catch (const std::exception& e) {
+        QPID_LOG(error, cluster << " error closing connection " << *this << ": " << e.what());
+    }
+}
+
+// Self-delivery of close message, close the connection.
+void Connection::deliverClose () {
+    assert(!catchUp);
+    connection.closed();
+    cluster.erase(self);
+}
+
+// The connection has been killed for misbehaving
+void Connection::abort() {
+    connection.abort();
+    cluster.erase(self);
+}
+
+// Member of a shadow connection left the cluster.
+void Connection::left() {
+    assert(isShadow());
+    connection.closed();
+}
+
+// ConnectoinCodec::decode receives read buffers from  directly-connected clients.
+size_t Connection::decode(const char* buffer, size_t size) {
+    if (catchUp) {  // Handle catch-up locally.
+        Buffer buf(const_cast<char*>(buffer), size);
+        while (localDecoder.decode(buf))
+            received(localDecoder.getFrame());
+    }
+    else {                      // Multicast local connections.
+        assert(isLocal());
+        const char* remainingData = buffer;
+        size_t remainingSize = size;
+        if (expectProtocolHeader) {
+            //If this is an outgoing link, we will receive a protocol
+            //header which needs to be decoded first
+            framing::ProtocolInitiation pi;
+            Buffer buf(const_cast<char*>(buffer), size);
+            if (pi.decode(buf)) {
+                //TODO: check the version is correct
+                QPID_LOG(debug, "Outgoing clustered link connection received INIT(" << pi << ")");
+                expectProtocolHeader = false;
+                remainingData = buffer + pi.encodedSize();
+                remainingSize = size - pi.encodedSize();
+            } else {
+                QPID_LOG(debug, "Not enough data for protocol header on outgoing clustered link");
+                giveReadCredit(1); // We're not going to mcast so give read credit now.
+                return 0;
+            }
+        }
+        cluster.getMulticast().mcastBuffer(remainingData, remainingSize, self);
+    }
+    return size;
+}
+
+broker::SessionState& Connection::sessionState() {
+    return *connection.getChannel(currentChannel).getSession();
+}
+
+broker::SemanticState& Connection::semanticState() {
+    return sessionState().getSemanticState();
+}
+
+void Connection::consumerState(const string& name, bool blocked, bool notifyEnabled, const SequenceNumber& position)
+{
+    broker::SemanticState::ConsumerImpl& c = semanticState().find(name);
+    c.position = position;
+    c.setBlocked(blocked);
+    if (notifyEnabled) c.enableNotify(); else c.disableNotify();
+    consumerNumbering.add(c.shared_from_this());
+}
+
+
+void Connection::sessionState(
+    const SequenceNumber& replayStart,
+    const SequenceNumber& sendCommandPoint,
+    const SequenceSet& sentIncomplete,
+    const SequenceNumber& expected,
+    const SequenceNumber& received,
+    const SequenceSet& unknownCompleted,
+    const SequenceSet& receivedIncomplete)
+{
+    
+    sessionState().setState(
+        replayStart,
+        sendCommandPoint,
+        sentIncomplete,
+        expected,
+        received,
+        unknownCompleted,
+        receivedIncomplete);
+    QPID_LOG(debug, cluster << " received session state update for " << sessionState().getId());
+    // The output tasks will be added later in the update process. 
+    connection.getOutputTasks().removeAll();
+}
+
+void Connection::outputTask(uint16_t channel, const std::string& name) {
+    broker::SessionState* session = connection.getChannel(channel).getSession();
+    if (!session)
+        throw Exception(QPID_MSG(cluster << " channel not attached " << *this
+                                 << "[" <<  channel << "] "));
+    OutputTask* task = &session->getSemanticState().find(name);
+    connection.getOutputTasks().addOutputTask(task);
+}
+    
+void Connection::shadowReady(uint64_t memberId, uint64_t connectionId, const string& username, const string& fragment, uint32_t sendMax) {
+    ConnectionId shadowId = ConnectionId(memberId, connectionId);
+    QPID_LOG(debug, cluster << " catch-up connection " << *this << " becomes shadow " << shadowId);
+    self = shadowId;
+    connection.setUserId(username);
+    // OK to use decoder here because cluster is stalled for update.
+    cluster.getDecoder().get(self).setFragment(fragment.data(), fragment.size());
+    connection.setErrorListener(this);
+    output.setSendMax(sendMax);
+}
+
+void Connection::membership(const FieldTable& joiners, const FieldTable& members, const framing::SequenceNumber& frameSeq) {
+    QPID_LOG(debug, cluster << " incoming update complete on connection " << *this);
+    cluster.updateInDone(ClusterMap(joiners, members, frameSeq));
+    consumerNumbering.clear();
+    self.second = 0;        // Mark this as completed update connection.
+}
+
+void Connection::retractOffer() {
+    QPID_LOG(debug, cluster << " incoming update retracted on connection " << *this);
+    cluster.updateInRetracted();
+    self.second = 0;        // Mark this as completed update connection.
+}
+
+bool Connection::isLocal() const {
+    return self.first == cluster.getId() && self.second;
+}
+
+bool Connection::isShadow() const {
+    return self.first != cluster.getId();
+}
+
+bool Connection::isUpdated() const {
+    return self.first == cluster.getId() && self.second == 0;
+}
+
+
+boost::shared_ptr<broker::Queue> Connection::findQueue(const std::string& qname) {
+    boost::shared_ptr<broker::Queue> queue = cluster.getBroker().getQueues().find(qname);
+    if (!queue) throw Exception(QPID_MSG(cluster << " can't find queue " << qname));
+    return queue;
+}
+
+broker::QueuedMessage Connection::getUpdateMessage() {
+    boost::shared_ptr<broker::Queue> updateq = findQueue(UpdateClient::UPDATE);
+    assert(!updateq->isDurable());
+    broker::QueuedMessage m = updateq->get();
+    if (!m.payload) throw Exception(QPID_MSG(cluster << " empty update queue"));
+    return m;
+}
+
+void Connection::deliveryRecord(const string& qname,
+                                const SequenceNumber& position,
+                                const string& tag,
+                                const SequenceNumber& id,
+                                bool acquired,
+                                bool accepted,
+                                bool cancelled,
+                                bool completed,
+                                bool ended,
+                                bool windowing,
+                                bool enqueued,
+                                uint32_t credit)
+{
+    broker::QueuedMessage m;
+    broker::Queue::shared_ptr queue = findQueue(qname);
+    if (!ended) {               // Has a message
+        if (acquired) {         // Message is on the update queue
+            m = getUpdateMessage();
+            m.queue = queue.get();
+            m.position = position;
+            if (enqueued) queue->enqueued(m); //inform queue of the message 
+        } else {                // Message at original position in original queue
+            m = queue->find(position);
+        }
+        if (!m.payload)
+            throw Exception(QPID_MSG("deliveryRecord no update message"));
+    }
+
+    broker::DeliveryRecord dr(m, queue, tag, acquired, accepted, windowing, credit);
+    dr.setId(id);
+    if (cancelled) dr.cancel(dr.getTag());
+    if (completed) dr.complete();
+    if (ended) dr.setEnded();   // Exsitance of message
+    semanticState().record(dr); // Part of the session's unacked list.
+}
+
+void Connection::queuePosition(const string& qname, const SequenceNumber& position) {
+    findQueue(qname)->setPosition(position);
+}
+
+void Connection::expiryId(uint64_t id) {
+    cluster.getExpiryPolicy().setId(id);
+}
+
+std::ostream& operator<<(std::ostream& o, const Connection& c) {
+    const char* type="unknown";
+    if (c.isLocal()) type = "local";
+    else if (c.isShadow()) type = "shadow";
+    else if (c.isUpdated()) type = "updated";
+    return o << c.getId() << "(" << type << (c.isCatchUp() ? ",catchup" : "") << ")";
+}
+
+void Connection::txStart() {
+    txBuffer.reset(new broker::TxBuffer());
+}
+void Connection::txAccept(const framing::SequenceSet& acked) {
+    txBuffer->enlist(boost::shared_ptr<broker::TxAccept>(
+                         new broker::TxAccept(acked, semanticState().getUnacked())));
+}
+
+void Connection::txDequeue(const std::string& queue) {
+    txBuffer->enlist(boost::shared_ptr<broker::RecoveredDequeue>(
+                         new broker::RecoveredDequeue(findQueue(queue), getUpdateMessage().payload)));
+}
+
+void Connection::txEnqueue(const std::string& queue) {
+    txBuffer->enlist(boost::shared_ptr<broker::RecoveredEnqueue>(
+                         new broker::RecoveredEnqueue(findQueue(queue), getUpdateMessage().payload)));
+}
+
+void Connection::txPublish(const framing::Array& queues, bool delivered) {
+    boost::shared_ptr<broker::TxPublish> txPub(new broker::TxPublish(getUpdateMessage().payload));
+    for (framing::Array::const_iterator i = queues.begin(); i != queues.end(); ++i) 
+        txPub->deliverTo(findQueue((*i)->get<std::string>()));
+    txPub->delivered = delivered;
+    txBuffer->enlist(txPub);
+}
+
+void Connection::txEnd() {
+    semanticState().setTxBuffer(txBuffer);
+}
+
+void Connection::accumulatedAck(const qpid::framing::SequenceSet& s) {
+    semanticState().setAccumulatedAck(s);
+}
+
+void Connection::exchange(const std::string& encoded) {
+    Buffer buf(const_cast<char*>(encoded.data()), encoded.size());
+    broker::Exchange::shared_ptr ex = broker::Exchange::decode(cluster.getBroker().getExchanges(), buf);
+    QPID_LOG(debug, cluster << " updated exchange " << ex->getName());
+}
+
+void Connection::queue(const std::string& encoded) {
+    Buffer buf(const_cast<char*>(encoded.data()), encoded.size());
+    broker::Queue::shared_ptr q = broker::Queue::decode(cluster.getBroker().getQueues(), buf);
+    QPID_LOG(debug, cluster << " updated queue " << q->getName());
+}
+
+void Connection::sessionError(uint16_t , const std::string& msg) {
+    cluster.flagError(*this, ERROR_TYPE_SESSION, msg);
+    
+}
+
+void Connection::connectionError(const std::string& msg) {
+    cluster.flagError(*this, ERROR_TYPE_CONNECTION, msg);
+}
+
+void Connection::addQueueListener(const std::string& q, uint32_t listener) {
+    if (listener >= consumerNumbering.size())
+        throw Exception(QPID_MSG("Invalid listener ID: " << listener));
+    findQueue(q)->getListeners().addListener(consumerNumbering[listener]);
+}
+
+}} // Namespace qpid::cluster
+
diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
new file mode 100644
index 0000000000..7f94338348
--- /dev/null
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -0,0 +1,210 @@
+#ifndef QPID_CLUSTER_CONNECTION_H
+#define QPID_CLUSTER_CONNECTION_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "types.h"
+#include "OutputInterceptor.h"
+#include "EventFrame.h"
+#include "McastFrameHandler.h"
+#include "UpdateReceiver.h"
+
+#include "qpid/broker/Connection.h"
+#include "qpid/broker/SemanticState.h"
+#include "qpid/amqp_0_10/Connection.h"
+#include "qpid/sys/AtomicValue.h"
+#include "qpid/sys/ConnectionInputHandler.h"
+#include "qpid/sys/ConnectionOutputHandler.h"
+#include "qpid/framing/SequenceNumber.h"
+#include "qpid/framing/FrameDecoder.h"
+
+#include <iosfwd>
+
+namespace qpid {
+
+namespace framing { class AMQFrame; }
+
+namespace broker {
+class SemanticState;
+class QueuedMessage;
+class TxBuffer;
+class TxAccept;
+}
+
+namespace cluster {
+class Cluster;
+class Event;
+
+/** Intercept broker::Connection calls for shadow and local cluster connections. */
+class Connection :
+        public RefCounted,
+        public sys::ConnectionInputHandler,
+        public framing::AMQP_AllOperations::ClusterConnectionHandler,
+        private broker::Connection::ErrorListener
+        
+{
+  public:
+    
+    /** Local connection. */
+    Connection(Cluster&, sys::ConnectionOutputHandler& out, const std::string& logId, MemberId, bool catchUp, bool isLink,
+               unsigned int ssf);
+    /** Shadow connection. */
+    Connection(Cluster&, sys::ConnectionOutputHandler& out, const std::string& logId, const ConnectionId& id,
+               unsigned int ssf);
+    ~Connection();
+    
+    ConnectionId getId() const { return self; }
+    broker::Connection& getBrokerConnection() { return connection; }
+
+    /** Local connections may be clients or catch-up connections */
+    bool isLocal() const;
+
+    bool isLocalClient() const { return isLocal() && !isCatchUp(); }
+
+    /** True for connections that are shadowing remote broker connections */
+    bool isShadow() const;
+
+    /** True if the connection is in "catch-up" mode: building initial broker state. */
+    bool isCatchUp() const { return catchUp; }
+
+    /** True if the connection is a completed shared update connection */
+    bool isUpdated() const;
+
+    Cluster& getCluster() { return cluster; }
+
+    // ConnectionInputHandler methods
+    void received(framing::AMQFrame&);
+    void closed();
+    bool doOutput();
+    bool hasOutput() { return connection.hasOutput(); }
+    void idleOut() { connection.idleOut(); }
+    void idleIn() { connection.idleIn(); }
+
+    /** Called if the connectors member has left the cluster */
+    void left();
+    
+    // ConnectionCodec methods - called by IO layer with a read buffer.
+    size_t decode(const char* buffer, size_t size);
+
+    // Called for data delivered from the cluster.
+    void deliveredFrame(const EventFrame&);
+
+    void consumerState(const std::string& name, bool blocked, bool notifyEnabled, const qpid::framing::SequenceNumber& position);
+    
+    // ==== Used in catch-up mode to build initial state.
+    // 
+    // State update methods.
+    void sessionState(const framing::SequenceNumber& replayStart,
+                      const framing::SequenceNumber& sendCommandPoint,
+                      const framing::SequenceSet& sentIncomplete,
+                      const framing::SequenceNumber& expected,
+                      const framing::SequenceNumber& received,
+                      const framing::SequenceSet& unknownCompleted,
+                      const SequenceSet& receivedIncomplete);
+    
+    void outputTask(uint16_t channel, const std::string& name);
+    
+    void shadowReady(uint64_t memberId, uint64_t connectionId, const std::string& username, const std::string& fragment, uint32_t sendMax);
+
+    void membership(const framing::FieldTable&, const framing::FieldTable&, const framing::SequenceNumber& frameSeq);
+
+    void retractOffer();
+
+    void deliveryRecord(const std::string& queue,
+                        const framing::SequenceNumber& position,
+                        const std::string& tag,
+                        const framing::SequenceNumber& id,
+                        bool acquired,
+                        bool accepted,
+                        bool cancelled,
+                        bool completed,
+                        bool ended,
+                        bool windowing,
+                        bool enqueued,
+                        uint32_t credit);
+
+    void queuePosition(const std::string&, const framing::SequenceNumber&);
+    void expiryId(uint64_t);
+
+    void txStart();
+    void txAccept(const framing::SequenceSet&);
+    void txDequeue(const std::string&);
+    void txEnqueue(const std::string&);
+    void txPublish(const framing::Array&, bool);
+    void txEnd();
+    void accumulatedAck(const framing::SequenceSet&);
+
+    // Encoded queue/exchange replication.
+    void queue(const std::string& encoded);
+    void exchange(const std::string& encoded);
+
+    void giveReadCredit(int credit);
+    void announce(uint32_t) {}  // handled by Cluster.
+    void abort();
+    void deliverClose();
+
+    OutputInterceptor& getOutput() { return output; }
+
+    void addQueueListener(const std::string& queue, uint32_t listener);
+
+  private:
+    struct NullFrameHandler : public framing::FrameHandler {
+        void handle(framing::AMQFrame&) {}
+    };
+    
+
+    static NullFrameHandler nullFrameHandler;
+
+    // Error listener functions
+    void connectionError(const std::string&);
+    void sessionError(uint16_t channel, const std::string&);
+    
+    void init();
+    bool checkUnsupported(const framing::AMQBody& body);
+    void deliverDoOutput(uint32_t limit) { output.deliverDoOutput(limit); }
+
+    boost::shared_ptr<broker::Queue> findQueue(const std::string& qname);
+    broker::SessionState& sessionState();
+    broker::SemanticState& semanticState();
+    broker::QueuedMessage getUpdateMessage();
+
+    Cluster& cluster;
+    ConnectionId self;
+    bool catchUp;
+    OutputInterceptor output;
+    framing::FrameDecoder localDecoder;
+    broker::Connection connection;
+    framing::SequenceNumber deliverSeq;
+    framing::ChannelId currentChannel;
+    boost::shared_ptr<broker::TxBuffer> txBuffer;
+    bool expectProtocolHeader;
+    McastFrameHandler mcastFrameHandler;
+    UpdateReceiver::ConsumerNumbering& consumerNumbering;
+
+    static qpid::sys::AtomicValue<uint64_t> catchUpId;
+    
+  friend std::ostream& operator<<(std::ostream&, const Connection&);
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_CONNECTION_H*/
diff --git a/cpp/src/qpid/cluster/ConnectionCodec.cpp b/cpp/src/qpid/cluster/ConnectionCodec.cpp
new file mode 100644
index 0000000000..8f6f1d9ad5
--- /dev/null
+++ b/cpp/src/qpid/cluster/ConnectionCodec.cpp
@@ -0,0 +1,82 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/ConnectionCodec.h"
+#include "qpid/cluster/Connection.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/cluster/ProxyInputHandler.h"
+#include "qpid/broker/Connection.h"
+#include "qpid/framing/ConnectionCloseBody.h"
+#include "qpid/framing/ConnectionCloseOkBody.h"
+#include "qpid/log/Statement.h"
+#include "qpid/memory.h"
+#include <stdexcept>
+#include <boost/utility/in_place_factory.hpp>
+
+namespace qpid {
+namespace cluster {
+
+using namespace framing;
+
+sys::ConnectionCodec*
+ConnectionCodec::Factory::create(ProtocolVersion v, sys::OutputControl& out, const std::string& id,
+                                 unsigned int ssf) {
+    if (v == ProtocolVersion(0, 10))
+        return new ConnectionCodec(v, out, id, cluster, false, false, ssf);
+    else if (v == ProtocolVersion(0x80 + 0, 0x80 + 10)) // Catch-up connection
+        return new ConnectionCodec(v, out, id, cluster, true, false, ssf); 
+    return 0;
+}
+
+// Used for outgoing Link connections
+sys::ConnectionCodec*
+ConnectionCodec::Factory::create(sys::OutputControl& out, const std::string& logId,
+                                 unsigned int ssf) {
+    return new ConnectionCodec(ProtocolVersion(0,10), out, logId, cluster, false, true, ssf);
+}
+
+ConnectionCodec::ConnectionCodec(
+    const ProtocolVersion& v, sys::OutputControl& out,
+    const std::string& logId, Cluster& cluster, bool catchUp, bool isLink, unsigned int ssf
+) : codec(out, logId, isLink),
+    interceptor(new Connection(cluster, codec, logId, cluster.getId(), catchUp, isLink, ssf))
+{
+    std::auto_ptr<sys::ConnectionInputHandler> ih(new ProxyInputHandler(interceptor));
+    codec.setInputHandler(ih);
+    codec.setVersion(v);
+}
+
+ConnectionCodec::~ConnectionCodec() {}
+
+size_t ConnectionCodec::decode(const char* buffer, size_t size) {
+    return interceptor->decode(buffer, size);
+}
+
+bool ConnectionCodec::isClosed() const { return codec.isClosed(); }
+
+size_t ConnectionCodec::encode(const char* buffer, size_t size) { return codec.encode(buffer, size); }
+
+bool ConnectionCodec::canEncode() { return codec.canEncode(); }
+
+void ConnectionCodec::closed() { codec.closed(); }
+
+ProtocolVersion ConnectionCodec::getVersion() const { return codec.getVersion(); }
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ConnectionCodec.h b/cpp/src/qpid/cluster/ConnectionCodec.h
new file mode 100644
index 0000000000..74cb3c507d
--- /dev/null
+++ b/cpp/src/qpid/cluster/ConnectionCodec.h
@@ -0,0 +1,82 @@
+#ifndef QPID_CLUSTER_CONNCTIONCODEC_H
+#define QPID_CLUSTER_CONNCTIONCODEC_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/amqp_0_10/Connection.h"
+#include "qpid/cluster/Connection.h"
+#include <boost/shared_ptr.hpp>
+#include <boost/intrusive_ptr.hpp>
+
+namespace qpid {
+
+namespace broker {
+class Connection;
+}
+
+namespace cluster {
+class Cluster;
+
+/**
+ * Encapsulates the standard amqp_0_10::ConnectionCodec and sets up
+ * a cluster::Connection for the connection.
+ *
+ * The ConnectionCodec is deleted by the network layer when the
+ * connection closes. The cluster::Connection needs to be kept
+ * around until all cluster business on the connection is complete.
+ *
+ */
+class ConnectionCodec : public sys::ConnectionCodec {
+  public:
+    struct Factory : public sys::ConnectionCodec::Factory {
+        boost::shared_ptr<sys::ConnectionCodec::Factory> next;
+        Cluster& cluster;
+        Factory(boost::shared_ptr<sys::ConnectionCodec::Factory> f, Cluster& c)
+            : next(f), cluster(c) {}
+        sys::ConnectionCodec* create(framing::ProtocolVersion, sys::OutputControl&, const std::string& id,
+                                     unsigned int conn_ssf);
+        sys::ConnectionCodec* create(sys::OutputControl&, const std::string& id,
+                                     unsigned int conn_ssf);
+    };
+
+    ConnectionCodec(const framing::ProtocolVersion&, sys::OutputControl& out,
+                    const std::string& logId, Cluster& c, bool catchUp, bool isLink,
+                    unsigned int ssf);
+    ~ConnectionCodec();
+
+    // ConnectionCodec functions.
+    size_t decode(const char* buffer, size_t size);
+    size_t encode(const char* buffer, size_t size);
+    bool canEncode();
+    void closed();
+    bool isClosed() const;
+    framing::ProtocolVersion getVersion() const;
+    
+
+  private:
+    amqp_0_10::Connection codec;
+    boost::intrusive_ptr<cluster::Connection> interceptor;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_CONNCTIONCODEC_H*/
diff --git a/cpp/src/qpid/cluster/ConnectionInterceptor.cpp b/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
deleted file mode 100644
index c13651eccb..0000000000
--- a/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-#include "ConnectionInterceptor.h"
-#include "qpid/framing/ClusterConnectionCloseBody.h"
-#include "qpid/framing/ClusterConnectionDoOutputBody.h"
-#include "qpid/framing/AMQFrame.h"
-
-namespace qpid {
-namespace cluster {
-
-using namespace framing;
-
-template <class T, class U, class V> void shift(T& a, U& b, const V& c) { a = b; b = c; }
-
-ConnectionInterceptor::ConnectionInterceptor(
-    broker::Connection& conn, Cluster& clust, Cluster::ShadowConnectionId shadowId_)
-    : connection(&conn), cluster(clust), isClosed(false), shadowId(shadowId_)
-{
-    connection->addFinalizer(boost::bind(operator delete, this));
-    // Attach  my functions to Connection extension points.
-    shift(receivedNext, connection->receivedFn, boost::bind(&ConnectionInterceptor::received, this, _1));
-    shift(closedNext, connection->closedFn, boost::bind(&ConnectionInterceptor::closed, this));
-    shift(doOutputNext, connection->doOutputFn, boost::bind(&ConnectionInterceptor::doOutput, this));
-}
-
-ConnectionInterceptor::~ConnectionInterceptor() {
-    assert(connection == 0);
-}
-
-void ConnectionInterceptor::received(framing::AMQFrame& f) {
-    if (isClosed) return;
-    cluster.send(f, this);
-}
-
-void ConnectionInterceptor::deliver(framing::AMQFrame& f) {
-    receivedNext(f);
-}
-
-void ConnectionInterceptor::closed() {
-    if (isClosed) return;
-    try {
-        // Called when the local network connection is closed. We still
-        // need to process any outstanding cluster frames for this
-        // connection to ensure our sessions are up-to-date. We defer
-        // closing the Connection object till deliverClosed(), but replace
-        // its output handler with a null handler since the network output
-        // handler will be deleted.
-        // 
-        connection->setOutputHandler(&discardHandler); 
-        cluster.send(AMQFrame(in_place<ClusterConnectionCloseBody>()), this);
-        isClosed = true;
-    }
-    catch (const std::exception& e) {
-        QPID_LOG(error, QPID_MSG("While closing connection: " << e.what()));
-    }
-}
-
-void ConnectionInterceptor::deliverClosed() {
-    closedNext();
-    // Drop reference so connection will be deleted, which in turn
-    // will delete this via finalizer added in ctor.
-    connection = 0;             
-}
-
-void ConnectionInterceptor::dirtyClose() {
-    // Not closed via cluster self-delivery but closed locally.
-    // Used for dirty cluster shutdown where active connections
-    // must be cleaned up.
-    connection = 0;
-}
-
-bool  ConnectionInterceptor::doOutput() {
-    // FIXME aconway 2008-08-15: this is not correct.
-    // Run in write threads so order of execution of doOutput is not determinate.
-    // Will only work reliably for in single-consumer tests.   
-
-    if (connection->hasOutput()) {
-        cluster.send(AMQFrame(in_place<ClusterConnectionDoOutputBody>()), this);
-        return doOutputNext();
-    }
-    return false;
-}
-
-void ConnectionInterceptor::deliverDoOutput() {
-    // FIXME aconway 2008-08-15: see comment in doOutput.
-    if (isShadow()) 
-        doOutputNext();
-}
-
-}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ConnectionInterceptor.h b/cpp/src/qpid/cluster/ConnectionInterceptor.h
deleted file mode 100644
index 370572bd9d..0000000000
--- a/cpp/src/qpid/cluster/ConnectionInterceptor.h
+++ /dev/null
@@ -1,82 +0,0 @@
-#ifndef QPID_CLUSTER_CONNECTIONPLUGIN_H
-#define QPID_CLUSTER_CONNECTIONPLUGIN_H
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "Cluster.h"
-#include "qpid/broker/Connection.h"
-#include "qpid/sys/ConnectionOutputHandler.h"
-
-namespace qpid {
-namespace framing { class AMQFrame; }
-namespace cluster {
-
-/**
- * Plug-in associated with broker::Connections, both local and shadow.
- */
-class ConnectionInterceptor {
-  public:
-    ConnectionInterceptor(broker::Connection&, Cluster&,
-                          Cluster::ShadowConnectionId shadowId=Cluster::ShadowConnectionId(0,0));
-    ~ConnectionInterceptor();
-    
-    Cluster::ShadowConnectionId getShadowId() const { return shadowId; }
-
-    bool isLocal() const { return shadowId == Cluster::ShadowConnectionId(0,0); }
-
-    // self-delivery of intercepted extension points.
-    void deliver(framing::AMQFrame& f);
-    void deliverClosed();
-    void deliverDoOutput();
-
-    void dirtyClose();
-
-  private:
-    struct NullConnectionHandler : public qpid::sys::ConnectionOutputHandler {
-        void close() {}
-        void send(framing::AMQFrame&) {}
-        void doOutput() {}
-        void activateOutput() {}
-    };
-
-    bool isShadow() { return shadowId != Cluster::ShadowConnectionId(0,0); }
-    
-    // Functions to intercept to Connection extension points.
-    void received(framing::AMQFrame&);
-    void closed();
-    bool doOutput();
-
-    boost::function<void (framing::AMQFrame&)> receivedNext;
-    boost::function<void ()> closedNext;
-    boost::function<bool ()> doOutputNext;
-
-    boost::intrusive_ptr<broker::Connection> connection;
-    Cluster& cluster;
-    NullConnectionHandler discardHandler;
-    bool isClosed;
-    Cluster::ShadowConnectionId shadowId;
-};
-
-}} // namespace qpid::cluster
-
-#endif  /*!QPID_CLUSTER_CONNECTIONPLUGIN_H*/
-
diff --git a/cpp/src/qpid/cluster/Cpg.cpp b/cpp/src/qpid/cluster/Cpg.cpp
index 2ffd3509bf..3ae0c970c7 100644
--- a/cpp/src/qpid/cluster/Cpg.cpp
+++ b/cpp/src/qpid/cluster/Cpg.cpp
@@ -16,33 +16,85 @@
  *
  */
 
-#include "Cpg.h"
+#include "qpid/cluster/Cpg.h"
 #include "qpid/sys/Mutex.h"
-// Note cpg is currently unix-specific. Refactor if availble on other platforms.
+#include "qpid/sys/Time.h"
 #include "qpid/sys/posix/PrivatePosix.h"
 #include "qpid/log/Statement.h"
 
 #include <vector>
 #include <limits>
 #include <iterator>
+#include <sstream>
 
 #include <unistd.h>
 
+// This is a macro instead of a function because we don't want to
+// evaluate the MSG argument unless there is an error.
+#define CPG_CHECK(RESULT, MSG) \
+    if ((RESULT) != CPG_OK) throw Exception(errorStr((RESULT), (MSG)))
+
 namespace qpid {
 namespace cluster {
 
 using namespace std;
 
+
+
 Cpg* Cpg::cpgFromHandle(cpg_handle_t handle) {
     void* cpg=0;
-    check(cpg_context_get(handle, &cpg), "Cannot get CPG instance.");
+    CPG_CHECK(cpg_context_get(handle, &cpg), "Cannot get CPG instance.");
     if (!cpg) throw Exception("Cannot get CPG instance.");
     return reinterpret_cast<Cpg*>(cpg);
 }
 
+// Applies the same retry-logic to all cpg calls that need it.
+void Cpg::callCpg ( CpgOp & c ) {
+    cpg_error_t result;
+    unsigned int snooze = 10;
+    for ( unsigned int nth_try = 0; nth_try < cpgRetries; ++ nth_try ) {
+        if ( CPG_OK == (result = c.op(handle, & group))) {
+            QPID_LOG(info, c.opName << " successful.");
+            break;
+        }
+        else if ( result == CPG_ERR_TRY_AGAIN ) {
+            QPID_LOG(info, "Retrying " << c.opName );
+            sys::usleep ( snooze );
+            snooze *= 10;
+            snooze = (snooze <= maxCpgRetrySleep) ? snooze : maxCpgRetrySleep;
+        }
+        else break;  // Don't retry unless CPG tells us to.
+    }
+
+    if ( result != CPG_OK )
+        CPG_CHECK(result, c.msg(group));
+}
+
 // Global callback functions.
 void Cpg::globalDeliver (
     cpg_handle_t handle,
+    const struct cpg_name *group,
+    uint32_t nodeid,
+    uint32_t pid,
+    void* msg,
+    size_t msg_len)
+{
+    cpgFromHandle(handle)->handler.deliver(handle, group, nodeid, pid, msg, msg_len);
+}
+
+void Cpg::globalConfigChange(
+    cpg_handle_t handle,
+    const struct cpg_name *group,
+    const struct cpg_address *members, size_t nMembers,
+    const struct cpg_address *left, size_t nLeft,
+    const struct cpg_address *joined, size_t nJoined
+)
+{
+    cpgFromHandle(handle)->handler.configChange(handle, group, members, nMembers, left, nLeft, joined, nJoined);
+}
+
+void Cpg::globalDeliver (
+    cpg_handle_t handle,
     struct cpg_name *group,
     uint32_t nodeid,
     uint32_t pid,
@@ -65,103 +117,119 @@ void Cpg::globalConfigChange(
 
 int Cpg::getFd() {
     int fd;
-    check(cpg_fd_get(handle, &fd), "Cannot get CPG file descriptor");
+    CPG_CHECK(cpg_fd_get(handle, &fd), "Cannot get CPG file descriptor");
     return fd;
 }
 
 Cpg::Cpg(Handler& h) : IOHandle(new sys::IOHandlePrivate), handler(h), isShutdown(false) {
-    cpg_callbacks_t callbacks = { &globalDeliver, &globalConfigChange };
-    check(cpg_initialize(&handle, &callbacks), "Cannot initialize CPG");
-    check(cpg_context_set(handle, this), "Cannot set CPG context");
+    cpg_callbacks_t callbacks;
+    ::memset(&callbacks, 0, sizeof(callbacks));
+    callbacks.cpg_deliver_fn = &globalDeliver;
+    callbacks.cpg_confchg_fn = &globalConfigChange;
+
+    QPID_LOG(notice, "Initializing CPG");
+    cpg_error_t err = cpg_initialize(&handle, &callbacks);
+    int retries = 6; // FIXME aconway 2009-08-06: make this configurable.
+    while (err == CPG_ERR_TRY_AGAIN && --retries) {
+        QPID_LOG(notice, "Re-trying CPG initialization.");
+        sys::sleep(5);
+        err = cpg_initialize(&handle, &callbacks);
+    }
+    CPG_CHECK(err, "Failed to initialize CPG.");
+    CPG_CHECK(cpg_context_set(handle, this), "Cannot set CPG context");
     // Note: CPG is currently unix-specific. If CPG is ported to
     // windows then this needs to be refactored into
     // qpid::sys::<platform>
     IOHandle::impl->fd = getFd();
-    QPID_LOG(debug, "Initialized CPG handle 0x" << std::hex << handle);
 }
 
 Cpg::~Cpg() {
     try {
         shutdown();
     } catch (const std::exception& e) {
-        QPID_LOG(error, "Exception in Cpg destructor: " << e.what());
+        QPID_LOG(error, "Error during CPG shutdown: " << e.what());
     }
 }
 
-void Cpg::join(const Name& group) {
-    check(cpg_join(handle, const_cast<Name*>(&group)),cantJoinMsg(group));
+void Cpg::join(const std::string& name) {
+    group = name;
+    callCpg ( cpgJoinOp );
 }
     
-void Cpg::leave(const Name& group) {
-    check(cpg_leave(handle,const_cast<Name*>(&group)),cantLeaveMsg(group));
+void Cpg::leave() {
+    callCpg ( cpgLeaveOp );
 }
 
-bool Cpg::isFlowControlEnabled() {
+
+
+
+bool Cpg::mcast(const iovec* iov, int iovLen) {
+    // Check for flow control
     cpg_flow_control_state_t flowState;
-    check(cpg_flow_control_state_get(handle, &flowState), "Cannot get CPG flow control status.");
-    return flowState == CPG_FLOW_CONTROL_ENABLED;
-}
-
-// TODO aconway 2008-08-07: better handling of flow control.
-// Wait for flow control to be disabled.
-// FIXME aconway 2008-08-08: does flow control check involve a round-trip? If so maybe remove...
-void Cpg::waitForFlowControl() {
-    int delayNs=1000;           // one millisecond
-    int tries=8;                // double the delay on each try.
-    while (isFlowControlEnabled() && tries > 0) {
-        QPID_LOG(warning, "CPG flow control enabled, retry in " << delayNs << "ns");
-        ::usleep(delayNs);
-        --tries;
-        delayNs *= 2;
-    };
-    if (tries == 0) {
-        // FIXME aconway 2008-08-07: this is a fatal leave-the-cluster condition.
-        throw Cpg::Exception("CPG flow control enabled, failed to send.");
-    }
-}
+    CPG_CHECK(cpg_flow_control_state_get(handle, &flowState), "Cannot get CPG flow control status.");
+    if (flowState == CPG_FLOW_CONTROL_ENABLED)
+        return false;
 
-void Cpg::mcast(const Name& group, const iovec* iov, int iovLen) {
-    waitForFlowControl();
     cpg_error_t result;
     do {
         result = cpg_mcast_joined(handle, CPG_TYPE_AGREED, const_cast<iovec*>(iov), iovLen);
-        if (result != CPG_ERR_TRY_AGAIN) check(result, cantMcastMsg(group));
+        if (result != CPG_ERR_TRY_AGAIN) CPG_CHECK(result, cantMcastMsg(group));
     } while(result == CPG_ERR_TRY_AGAIN);
+    return true;
 }
 
 void Cpg::shutdown() {
     if (!isShutdown) {
         QPID_LOG(debug,"Shutting down CPG");
         isShutdown=true;
-        check(cpg_finalize(handle), "Error in shutdown of CPG");
+
+        callCpg ( cpgFinalizeOp );
     }
 }
 
+void Cpg::dispatchOne() {
+    CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_ONE), "Error in CPG dispatch");
+}
+
+void Cpg::dispatchAll() {
+    CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_ALL), "Error in CPG dispatch");
+}
+
+void Cpg::dispatchBlocking() {
+    CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_BLOCKING), "Error in CPG dispatch");
+}
+
 string Cpg::errorStr(cpg_error_t err, const std::string& msg) {
+    std::ostringstream  os;
+    os << msg << ": ";
     switch (err) {
-      case CPG_OK: return msg+": ok";
-      case CPG_ERR_LIBRARY: return msg+": library";
-      case CPG_ERR_TIMEOUT: return msg+": timeout";
-      case CPG_ERR_TRY_AGAIN: return msg+": timeout. The aisexec daemon may not be running";
-      case CPG_ERR_INVALID_PARAM: return msg+": invalid param";
-      case CPG_ERR_NO_MEMORY: return msg+": no memory";
-      case CPG_ERR_BAD_HANDLE: return msg+": bad handle";
-      case CPG_ERR_ACCESS: return msg+": access denied. You may need to set your group ID to 'ais'";
-      case CPG_ERR_NOT_EXIST: return msg+": not exist";
-      case CPG_ERR_EXIST: return msg+": exist";
-      case CPG_ERR_NOT_SUPPORTED: return msg+": not supported";
-      case CPG_ERR_SECURITY: return msg+": security";
-      case CPG_ERR_TOO_MANY_GROUPS: return msg+": too many groups";
-      default:
-        assert(0);
-        return ": unknown";
+      case CPG_OK: os << "ok"; break;
+      case CPG_ERR_LIBRARY: os << "library"; break;
+      case CPG_ERR_TIMEOUT: os << "timeout"; break;
+      case CPG_ERR_TRY_AGAIN: os << "try again"; break;
+      case CPG_ERR_INVALID_PARAM: os << "invalid param"; break;
+      case CPG_ERR_NO_MEMORY: os << "no memory"; break;
+      case CPG_ERR_BAD_HANDLE: os << "bad handle"; break;
+      case CPG_ERR_ACCESS: os << "access denied. You may need to set your group ID to 'ais'"; break;
+      case CPG_ERR_NOT_EXIST: os << "not exist"; break;
+      case CPG_ERR_EXIST: os << "exist"; break;
+      case CPG_ERR_NOT_SUPPORTED: os << "not supported"; break;
+      case CPG_ERR_SECURITY: os << "security"; break;
+      case CPG_ERR_TOO_MANY_GROUPS: os << "too many groups"; break;
+      default: os << ": unknown cpg error " << err;
     };
+    os << " (" << err << ")";
+    return os.str();
 }
 
 std::string Cpg::cantJoinMsg(const Name& group) {
     return "Cannot join CPG group "+group.str();
 }
 
+std::string Cpg::cantFinalizeMsg(const Name& group) {
+    return "Cannot finalize CPG group "+group.str();
+}
+
 std::string Cpg::cantLeaveMsg(const Name& group) {
     return "Cannot leave CPG group "+group.str();
 }
@@ -170,27 +238,44 @@ std::string Cpg::cantMcastMsg(const Name& group) {
     return "Cannot mcast to CPG group "+group.str();
 }
 
-Cpg::Id Cpg::self() const {
+MemberId Cpg::self() const {
     unsigned int nodeid;
-    check(cpg_local_get(handle, &nodeid), "Cannot get local CPG identity");
-    return Id(nodeid, getpid());
+    CPG_CHECK(cpg_local_get(handle, &nodeid), "Cannot get local CPG identity");
+    return MemberId(nodeid, getpid());
 }
 
-ostream& operator<<(ostream& o, std::pair<cpg_address*,int> a) {
-    ostream_iterator<Cpg::Id> i(o, " ");
-    std::copy(a.first, a.first+a.second, i);
-    return o;
+namespace { int byte(uint32_t value, int i) { return (value >> (i*8)) & 0xff; } }
+
+ostream& operator<<(ostream& out, const MemberId& id) {
+    if (id.first) {
+        out << byte(id.first, 0) << "."
+            << byte(id.first, 1) << "."
+            << byte(id.first, 2) << "."
+            << byte(id.first, 3)
+            << ":";
+    }
+    return out << id.second;
 }
 
-ostream& operator <<(ostream& out, const Cpg::Id& id) {
-    return out << id.getNodeId() << "-" << id.getPid();
+ostream& operator<<(ostream& o, const ConnectionId& c) {
+    return o << c.first << "-" << c.second;
 }
 
-ostream& operator <<(ostream& out, const cpg_name& name) {
-    return out << string(name.value, name.length);
+std::string MemberId::str() const  {
+    char s[8];
+    uint32_t x;
+    x = htonl(first);
+    ::memcpy(s, &x, 4);
+    x = htonl(second);
+    ::memcpy(s+4, &x, 4);
+    return std::string(s,8);
 }
 
+MemberId::MemberId(const std::string& s) {
+    uint32_t x;
+    memcpy(&x, &s[0], 4);
+    first = ntohl(x);
+    memcpy(&x, &s[4], 4);
+    second = ntohl(x);
+}
 }} // namespace qpid::cluster
-
-
-
diff --git a/cpp/src/qpid/cluster/Cpg.h b/cpp/src/qpid/cluster/Cpg.h
index 96fd692a77..6b81c602bd 100644
--- a/cpp/src/qpid/cluster/Cpg.h
+++ b/cpp/src/qpid/cluster/Cpg.h
@@ -20,25 +20,18 @@
  */
 
 #include "qpid/Exception.h"
+#include "qpid/cluster/types.h"
 #include "qpid/sys/IOHandle.h"
-#include "qpid/cluster/Dispatchable.h"
+#include "qpid/sys/Mutex.h"
 
-#include <boost/tuple/tuple.hpp>
-#include <boost/tuple/tuple_comparison.hpp>
 #include <boost/scoped_ptr.hpp>
 
 #include <cassert>
-
 #include <string.h>
 
-extern "C" {
-#include <openais/cpg.h>
-}
-
 namespace qpid {
 namespace cluster {
 
-
 /**
  * Lightweight C++ interface to cpg.h operations.
  * 
@@ -46,6 +39,7 @@ namespace cluster {
  * On error all functions throw Cpg::Exception.
  *
  */
+
 class Cpg : public sys::IOHandle {
   public:
     struct Exception : public ::qpid::Exception {
@@ -53,6 +47,7 @@ class Cpg : public sys::IOHandle {
     };
 
     struct Name : public cpg_name {
+        Name() { length = 0; }
         Name(const char* s) { copy(s, strlen(s)); }
         Name(const char* s, size_t n) { copy(s,n); }
         Name(const std::string& s) { copy(s.data(), s.size()); }
@@ -65,14 +60,6 @@ class Cpg : public sys::IOHandle {
         std::string str() const { return std::string(value, length); }
     };
 
-    // boost::tuple gives us == and < for free.
-    struct Id : public boost::tuple<uint32_t, uint32_t>  {
-        Id(uint32_t n=0, uint32_t p=0) : boost::tuple<uint32_t, uint32_t>(n, p) {}
-        Id(const cpg_address& addr) : boost::tuple<uint32_t, uint32_t>(addr.nodeid, addr.pid) {}
-        uint32_t getNodeId() const { return boost::get<0>(*this); }
-        uint32_t getPid() const { return boost::get<1>(*this); }
-    };
-
     static std::string str(const cpg_name& n) {
         return std::string(n.value, n.length);
     }
@@ -81,7 +68,7 @@ class Cpg : public sys::IOHandle {
         virtual ~Handler() {};
         virtual void deliver(
             cpg_handle_t /*handle*/,
-            struct cpg_name *group,
+            const struct cpg_name *group,
             uint32_t /*nodeid*/,
             uint32_t /*pid*/,
             void* /*msg*/,
@@ -89,10 +76,10 @@ class Cpg : public sys::IOHandle {
 
         virtual void configChange(
             cpg_handle_t /*handle*/,
-            struct cpg_name */*group*/,
-            struct cpg_address */*members*/, int /*nMembers*/,
-            struct cpg_address */*left*/, int /*nLeft*/,
-            struct cpg_address */*joined*/, int /*nJoined*/
+            const struct cpg_name */*group*/,
+            const struct cpg_address */*members*/, int /*nMembers*/,
+            const struct cpg_address */*left*/, int /*nLeft*/,
+            const struct cpg_address */*joined*/, int /*nJoined*/
         ) = 0;
     };
 
@@ -107,41 +94,115 @@ class Cpg : public sys::IOHandle {
     /** Disconnect from CPG */
     void shutdown();
     
-    /** Dispatch CPG events.
-     *@param type one of
-     * - CPG_DISPATCH_ONE - dispatch exactly one event.
-     * - CPG_DISPATCH_ALL - dispatch all available events, don't wait.
-     * - CPG_DISPATCH_BLOCKING - blocking dispatch loop.
-     */
-    void dispatch(cpg_dispatch_t type) {
-        check(cpg_dispatch(handle,type), "Error in CPG dispatch");
-    }
+    void dispatchOne();
+    void dispatchAll();
+    void dispatchBlocking();
 
-    void dispatchOne() { dispatch(CPG_DISPATCH_ONE); }
-    void dispatchAll() { dispatch(CPG_DISPATCH_ALL); }
-    void dispatchBlocking() { dispatch(CPG_DISPATCH_BLOCKING); }
+    void join(const std::string& group);    
+    void leave();
 
-    void join(const Name& group);    
-    void leave(const Name& group);
-    void mcast(const Name& group, const iovec* iov, int iovLen);
+    /** Multicast to the group. NB: must not be called concurrently.
+     * 
+     *@return true if the message was multi-cast, false if
+     * it was not sent due to flow control.
+     */
+    bool mcast(const iovec* iov, int iovLen);
 
     cpg_handle_t getHandle() const { return handle; }
 
-    Id self() const;
+    MemberId self() const;
 
     int getFd();
     
   private:
+
+    // Maximum number of retries for cog functions that can tell
+    // us to "try again later".
+    static const unsigned int cpgRetries = 5;
+
+    // Don't let sleep-time between cpg retries to go above 0.1 second.
+    static const unsigned int maxCpgRetrySleep = 100000;
+
+
+    // Base class for the Cpg operations that need retry capability.
+    struct CpgOp { 
+        std::string opName;
+
+        CpgOp ( std::string opName ) 
+          : opName(opName) { }
+
+        virtual cpg_error_t op ( cpg_handle_t handle, struct cpg_name * ) = 0; 
+        virtual std::string msg(const Name&) = 0;
+        virtual ~CpgOp ( ) { }
+    };
+
+
+    struct CpgJoinOp : public CpgOp { 
+        CpgJoinOp ( )
+          : CpgOp ( std::string("cpg_join") ) { }
+
+        cpg_error_t op(cpg_handle_t handle, struct cpg_name * group) { 
+            return cpg_join ( handle, group ); 
+        }
+
+        std::string msg(const Name& name) { return cantJoinMsg(name); }
+    };
+
+    struct CpgLeaveOp : public CpgOp { 
+        CpgLeaveOp ( )
+          : CpgOp ( std::string("cpg_leave") ) { }
+
+        cpg_error_t op(cpg_handle_t handle, struct cpg_name * group) { 
+            return cpg_leave ( handle, group ); 
+        }
+
+        std::string msg(const Name& name) { return cantLeaveMsg(name); }
+    };
+
+    struct CpgFinalizeOp : public CpgOp { 
+        CpgFinalizeOp ( )
+          : CpgOp ( std::string("cpg_finalize") ) { }
+
+        cpg_error_t op(cpg_handle_t handle, struct cpg_name *) { 
+            return cpg_finalize ( handle ); 
+        }
+
+        std::string msg(const Name& name) { return cantFinalizeMsg(name); }
+    };
+
+    // This fn standardizes retry policy across all Cpg ops that need it.
+    void callCpg ( CpgOp & );
+
+    CpgJoinOp     cpgJoinOp;
+    CpgLeaveOp    cpgLeaveOp;
+    CpgFinalizeOp cpgFinalizeOp;
+
     static std::string errorStr(cpg_error_t err, const std::string& msg);
     static std::string cantJoinMsg(const Name&);
-    static std::string cantLeaveMsg(const Name&); std::string cantMcastMsg(const Name&);
-
-    static void check(cpg_error_t result, const std::string& msg) {
-        if (result != CPG_OK) throw Exception(errorStr(result, msg));
-    }
+    static std::string cantLeaveMsg(const Name&);
+    static std::string cantMcastMsg(const Name&);
+    static std::string cantFinalizeMsg(const Name&);
 
     static Cpg* cpgFromHandle(cpg_handle_t);
 
+    // New versions for corosync 1.0 and higher
+    static void globalDeliver(
+        cpg_handle_t handle,
+        const struct cpg_name *group,
+        uint32_t nodeid,
+        uint32_t pid,
+        void* msg,
+        size_t msg_len);
+
+    static void globalConfigChange(
+        cpg_handle_t handle,
+        const struct cpg_name *group,
+        const struct cpg_address *members, size_t nMembers,
+        const struct cpg_address *left, size_t nLeft,
+        const struct cpg_address *joined, size_t nJoined
+    );
+
+    // Old versions for openais
     static void globalDeliver(
         cpg_handle_t handle,
         struct cpg_name *group,
@@ -158,18 +219,13 @@ class Cpg : public sys::IOHandle {
         struct cpg_address *joined, int nJoined
     );
 
-    bool isFlowControlEnabled();
-    void waitForFlowControl();
-
     cpg_handle_t handle;
     Handler& handler;
     bool isShutdown;
+    Name group;
+    sys::Mutex dispatchLock;
 };
 
-std::ostream& operator <<(std::ostream& out, const cpg_name& name);
-std::ostream& operator <<(std::ostream& out, const Cpg::Id& id);
-std::ostream& operator <<(std::ostream& out, const std::pair<cpg_address*,int> addresses);
-
 inline bool operator==(const cpg_name& a, const cpg_name& b) {
     return a.length==b.length &&  strncmp(a.value, b.value, a.length) == 0;
 }
@@ -177,5 +233,4 @@ inline bool operator!=(const cpg_name& a, const cpg_name& b) { return !(a == b);
 
 }} // namespace qpid::cluster
 
-
 #endif  /*!CPG_H*/
diff --git a/cpp/src/qpid/cluster/Decoder.cpp b/cpp/src/qpid/cluster/Decoder.cpp
new file mode 100644
index 0000000000..23ba372d78
--- /dev/null
+++ b/cpp/src/qpid/cluster/Decoder.cpp
@@ -0,0 +1,65 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/Decoder.h"
+#include "qpid/cluster/EventFrame.h"
+#include "qpid/framing/ClusterConnectionDeliverCloseBody.h"
+#include "qpid/framing/Buffer.h"
+#include "qpid/framing/AMQFrame.h"
+
+
+namespace qpid {
+namespace cluster {
+
+void Decoder::decode(const EventHeader& eh, const char* data) {
+    sys::Mutex::ScopedLock l(lock);
+    assert(eh.getType() == DATA); // Only handle connection data events.
+    const char* cp = static_cast<const char*>(data);
+    framing::Buffer buf(const_cast<char*>(cp), eh.getSize());
+    framing::FrameDecoder& decoder = map[eh.getConnectionId()];
+    if (decoder.decode(buf)) {  // Decoded a frame
+        framing::AMQFrame frame(decoder.getFrame());
+        while (decoder.decode(buf)) {
+            callback(EventFrame(eh, frame));
+            frame = decoder.getFrame();
+        }
+        // Set read-credit on the last frame ending in this event.
+        // Credit will be given when this frame is processed.
+        callback(EventFrame(eh, frame, 1)); 
+    }
+    else {
+        // We must give 1 unit read credit per event.
+        // This event does not complete any frames so 
+        // send an empty frame with the read credit.
+        callback(EventFrame(eh, framing::AMQFrame(), 1));
+    }    
+}
+
+void Decoder::erase(const ConnectionId& c) {
+    sys::Mutex::ScopedLock l(lock);
+    map.erase(c);
+}
+
+framing::FrameDecoder& Decoder::get(const ConnectionId& c) {
+    sys::Mutex::ScopedLock l(lock);
+    return map[c];
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/Decoder.h b/cpp/src/qpid/cluster/Decoder.h
new file mode 100644
index 0000000000..2e2af2868f
--- /dev/null
+++ b/cpp/src/qpid/cluster/Decoder.h
@@ -0,0 +1,59 @@
+#ifndef QPID_CLUSTER_DECODER_H
+#define QPID_CLUSTER_DECODER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/framing/FrameDecoder.h"
+#include "qpid/sys/Mutex.h"
+#include <boost/function.hpp>
+#include <map>
+
+namespace qpid {
+namespace cluster {
+
+class EventFrame;
+class EventHeader;
+
+/**
+ * A map of decoders for connections.
+ */
+class Decoder
+{
+  public:
+    typedef boost::function<void(const EventFrame&)> FrameHandler;
+
+    Decoder(FrameHandler fh) : callback(fh) {}
+    void decode(const EventHeader& eh, const char* data);
+    void erase(const ConnectionId&);
+    framing::FrameDecoder& get(const ConnectionId& c);
+
+  private:
+    typedef std::map<ConnectionId, framing::FrameDecoder> Map;
+    sys::Mutex lock;
+    Map map;
+    void process(const EventFrame&);
+    FrameHandler callback;
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_DECODER_H*/
diff --git a/cpp/src/qpid/cluster/ErrorCheck.cpp b/cpp/src/qpid/cluster/ErrorCheck.cpp
new file mode 100644
index 0000000000..d66db8551d
--- /dev/null
+++ b/cpp/src/qpid/cluster/ErrorCheck.cpp
@@ -0,0 +1,155 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/ErrorCheck.h"
+#include "qpid/cluster/EventFrame.h"
+#include "qpid/cluster/ClusterMap.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/framing/ClusterErrorCheckBody.h"
+#include "qpid/framing/ClusterConfigChangeBody.h"
+#include "qpid/log/Statement.h"
+
+#include <algorithm>
+
+namespace qpid {
+namespace cluster {
+
+using namespace std;
+using namespace framing;
+using namespace framing::cluster;
+
+ErrorCheck::ErrorCheck(Cluster& c)
+    : cluster(c), mcast(c.getMulticast()), frameSeq(0), type(ERROR_TYPE_NONE), connection(0)
+{}
+
+void ErrorCheck::error(
+    Connection& c, ErrorType t, framing::SequenceNumber seq, const MemberSet& ms,
+    const std::string& msg)
+{
+    // Detected a local error, inform cluster and set error state.
+    assert(t != ERROR_TYPE_NONE); // Must be an error.
+    assert(type == ERROR_TYPE_NONE); // Can't be called when already in an error state.
+    type = t;
+    unresolved = ms;
+    frameSeq = seq;
+    connection = &c;
+    message = msg;
+    QPID_LOG(debug, cluster<< (type == ERROR_TYPE_SESSION ? " channel" : " connection")
+             << " error " << frameSeq << " on " << c
+             << " must be resolved with: " << unresolved
+             << ": " << message);
+    mcast.mcastControl(
+        ClusterErrorCheckBody(ProtocolVersion(), type, frameSeq), cluster.getId());
+    // If there are already frames queued up by a previous error, review
+    // them with respect to this new error.
+    for (FrameQueue::iterator i = frames.begin(); i != frames.end(); i = review(i))
+        ;
+}
+
+void ErrorCheck::delivered(const EventFrame& e) {
+    frames.push_back(e);
+    review(frames.end()-1);
+}
+
+// Review a frame in the queue with respect to the current error.
+ErrorCheck::FrameQueue::iterator ErrorCheck::review(const FrameQueue::iterator& i) {
+    FrameQueue::iterator next = i+1;
+    if(!isUnresolved() || !i->frame.getBody() || !i->frame.getMethod())
+        return next;            // Only interested in control frames while unresolved.
+    const AMQMethodBody* method = i->frame.getMethod();
+    if (method->isA<const ClusterErrorCheckBody>()) {
+        const ClusterErrorCheckBody* errorCheck =
+            static_cast<const ClusterErrorCheckBody*>(method);
+
+        if (errorCheck->getFrameSeq() == frameSeq) { // Addresses current error
+            next = frames.erase(i);    // Drop matching error check controls
+            if (errorCheck->getType() < type) { // my error is worse than his
+                QPID_LOG(critical, cluster
+                         << " local error " << frameSeq << " did not occur on member "
+                         << i->getMemberId()
+                         << ": " << message);
+                throw Exception(
+                    QPID_MSG("local error did not occur on all cluster members " << ": " << message));
+            }
+            else {              // his error is worse/same as mine.
+                QPID_LOG(debug, cluster << " error " << frameSeq
+                         << " resolved with " << i->getMemberId());
+                unresolved.erase(i->getMemberId());
+                checkResolved();
+            }
+        }
+        else if (errorCheck->getFrameSeq() < frameSeq && errorCheck->getType() != NONE
+                 && i->connectionId.getMember() != cluster.getId())
+        {
+            // This error occured before the current error so we
+            // have processed past it.
+            next = frames.erase(i); // Drop the error check control
+            respondNone(i->connectionId.getMember(), errorCheck->getType(),
+                        errorCheck->getFrameSeq());
+        }
+        // if errorCheck->getFrameSeq() > frameSeq then leave it in the queue.
+    }
+    else if (method->isA<const ClusterConfigChangeBody>()) {
+        const ClusterConfigChangeBody* configChange =
+            static_cast<const ClusterConfigChangeBody*>(method);
+        if (configChange) {
+            MemberSet members(decodeMemberSet(configChange->getCurrent()));
+            QPID_LOG(debug, cluster << " apply config change to error "
+                     << frameSeq << ": " << members);
+            MemberSet intersect;
+            set_intersection(members.begin(), members.end(),
+                             unresolved.begin(), unresolved.end(),
+                             inserter(intersect, intersect.begin()));
+            unresolved.swap(intersect);
+            checkResolved();
+        }
+    }
+    return next;
+}
+
+void ErrorCheck::checkResolved() {
+    if (unresolved.empty()) {   // No more potentially conflicted members, we're clear.
+        type = ERROR_TYPE_NONE;
+        QPID_LOG(debug, cluster << " error " << frameSeq << " resolved.");
+    }
+    else 
+        QPID_LOG(debug, cluster << " error " << frameSeq
+                 << " must be resolved with " << unresolved);
+}
+
+EventFrame ErrorCheck::getNext() {
+    assert(canProcess());
+    EventFrame e(frames.front());
+    frames.pop_front();
+    return e;
+}
+
+void ErrorCheck::respondNone(const MemberId& from, uint8_t type, framing::SequenceNumber frameSeq) {
+    // Don't respond to non-errors or to my own errors.
+    if (type == ERROR_TYPE_NONE || from == cluster.getId())
+        return;
+    QPID_LOG(debug, cluster << " error " << frameSeq << " did not occur locally.");
+    mcast.mcastControl(
+        ClusterErrorCheckBody(ProtocolVersion(), ERROR_TYPE_NONE, frameSeq),
+        cluster.getId()
+    );
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ErrorCheck.h b/cpp/src/qpid/cluster/ErrorCheck.h
new file mode 100644
index 0000000000..de8cedafb3
--- /dev/null
+++ b/cpp/src/qpid/cluster/ErrorCheck.h
@@ -0,0 +1,90 @@
+#ifndef QPID_CLUSTER_ERRORCHECK_H
+#define QPID_CLUSTER_ERRORCHECK_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/MemberSet.h"
+#include "qpid/cluster/Multicaster.h"
+#include "qpid/framing/enum.h"
+#include "qpid/framing/SequenceNumber.h"
+#include <boost/function.hpp>
+#include <deque>
+#include <set>
+
+namespace qpid {
+namespace cluster {
+
+class EventFrame;
+class Cluster;
+class Multicaster;
+class Connection;
+
+/**
+ * Error checking logic.
+ * 
+ * When an error occurs queue up frames until we can determine if all
+ * nodes experienced the error. If not, we shut down.
+ */
+class ErrorCheck
+{
+  public:
+    typedef framing::cluster::ErrorType ErrorType;
+    typedef framing::SequenceNumber SequenceNumber;
+    
+    ErrorCheck(Cluster&);
+
+    /** A local error has occured */
+    void error(Connection&, ErrorType, SequenceNumber frameSeq, const MemberSet&,
+               const std::string& msg);
+
+    /** Called when a frame is delivered */
+    void delivered(const EventFrame&);
+
+    /**@pre canProcess **/
+    EventFrame getNext();
+
+    bool canProcess() const { return type == NONE && !frames.empty(); }
+
+    bool isUnresolved() const { return type != NONE; }
+
+    /** Respond to an error check saying we had no error. */
+    void respondNone(const MemberId&, uint8_t type, SequenceNumber frameSeq);
+    
+  private:
+    static const ErrorType NONE = framing::cluster::ERROR_TYPE_NONE;
+    typedef std::deque<EventFrame>  FrameQueue;
+    FrameQueue::iterator review(const FrameQueue::iterator&);
+    void checkResolved();
+    
+    Cluster& cluster;
+    Multicaster& mcast;
+    FrameQueue frames;
+    MemberSet unresolved;
+    SequenceNumber frameSeq;
+    ErrorType type;
+    Connection* connection;
+    std::string message;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_ERRORCHECK_H*/
diff --git a/cpp/src/qpid/cluster/Event.cpp b/cpp/src/qpid/cluster/Event.cpp
new file mode 100644
index 0000000000..52564990f6
--- /dev/null
+++ b/cpp/src/qpid/cluster/Event.cpp
@@ -0,0 +1,136 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/types.h"
+#include "qpid/cluster/Event.h"
+#include "qpid/cluster/Cpg.h"
+#include "qpid/framing/Buffer.h"
+#include "qpid/framing/AMQFrame.h"
+#include "qpid/assert.h"
+#include <ostream>
+#include <iterator>
+#include <algorithm>
+
+namespace qpid {
+namespace cluster {
+
+using framing::Buffer;
+using framing::AMQFrame;
+
+const size_t EventHeader::HEADER_SIZE =
+    sizeof(uint8_t) +  // type
+    sizeof(uint64_t) + // connection pointer only, CPG provides member ID.
+    sizeof(uint32_t)  // payload size
+    ;
+
+EventHeader::EventHeader(EventType t, const ConnectionId& c,  size_t s)
+    : type(t), connectionId(c), size(s) {}
+
+
+Event::Event() {}
+
+Event::Event(EventType t, const ConnectionId& c,  size_t s)
+    : EventHeader(t,c,s), store(RefCountedBuffer::create(s+HEADER_SIZE))
+{}
+
+void EventHeader::decode(const MemberId& m, framing::Buffer& buf) {
+    if (buf.available() <= HEADER_SIZE)
+        throw Exception("Not enough for multicast header");
+    type = (EventType)buf.getOctet();
+    if(type != DATA && type != CONTROL)
+        throw Exception("Invalid multicast event type");
+    connectionId = ConnectionId(m, buf.getLongLong());
+    size = buf.getLong();
+}
+
+Event Event::decodeCopy(const MemberId& m, framing::Buffer& buf) {
+    Event e;
+    e.decode(m, buf);           // Header
+    if (buf.available() < e.size)
+        throw Exception("Not enough data for multicast event");
+    e.store = RefCountedBuffer::create(e.size + HEADER_SIZE);
+    memcpy(e.getData(), buf.getPointer() + buf.getPosition(), e.size);
+    return e;
+}
+
+Event Event::control(const framing::AMQFrame& f, const ConnectionId& cid) {
+    Event e(CONTROL, cid, f.encodedSize());
+    Buffer buf(e);
+    f.encode(buf);
+    return e;
+}
+
+Event Event::control(const framing::AMQBody& body, const ConnectionId& cid) {
+    return control(framing::AMQFrame(body), cid);
+}
+
+iovec Event::toIovec() const {
+    encodeHeader();
+    iovec iov = { const_cast<char*>(getStore()), getStoreSize() };
+    return iov;
+}
+
+void EventHeader::encode(Buffer& b) const {
+    b.putOctet(type);
+    b.putLongLong(connectionId.getNumber());
+    b.putLong(size);
+}
+
+// Encode my header in my buffer.
+void Event::encodeHeader () const {
+    Buffer b(const_cast<char*>(getStore()), HEADER_SIZE);
+    encode(b);
+    assert(b.getPosition() == HEADER_SIZE);
+}
+
+Event::operator Buffer() const  {
+    return Buffer(const_cast<char*>(getData()), getSize());
+}
+
+const AMQFrame& Event::getFrame() const {
+    assert(type == CONTROL);
+    if (!frame.getBody()) {
+    Buffer buf(*this);
+    QPID_ASSERT(frame.decode(buf));
+    }
+    return frame;
+}
+
+static const char* EVENT_TYPE_NAMES[] = { "data", "control" };
+
+std::ostream& operator<< (std::ostream& o, EventType t) {
+    return o << EVENT_TYPE_NAMES[t];
+}
+
+std::ostream& operator<< (std::ostream& o, const EventHeader& e) {
+    return o << "Event[" << e.getConnectionId() << " " << e.getType()
+             << " " << e.getSize() << " bytes]";
+}
+
+std::ostream& operator<< (std::ostream& o, const Event& e) {
+    o << "Event[" << e.getConnectionId() << " ";
+    if (e.getType() == CONTROL)
+        o << e.getFrame();
+    else
+        o << " data " << e.getSize() << " bytes";
+    return o << "]";
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/Event.h b/cpp/src/qpid/cluster/Event.h
new file mode 100644
index 0000000000..07f74d3ba5
--- /dev/null
+++ b/cpp/src/qpid/cluster/Event.h
@@ -0,0 +1,116 @@
+#ifndef QPID_CLUSTER_EVENT_H
+#define QPID_CLUSTER_EVENT_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/RefCountedBuffer.h"
+#include "qpid/framing/AMQFrame.h"
+#include <sys/uio.h>            // For iovec
+#include <iosfwd>
+
+#include "qpid/cluster/types.h"
+
+namespace qpid {
+
+namespace framing {
+class AMQBody;
+class AMQFrame;
+class Buffer;
+}
+
+namespace cluster {
+
+/** Header data for a multicast event */
+class EventHeader {
+  public:
+    EventHeader(EventType t=DATA, const ConnectionId& c=ConnectionId(), size_t size=0);
+    void decode(const MemberId& m, framing::Buffer&);
+    void encode(framing::Buffer&) const;
+
+    EventType getType() const { return type; }
+    ConnectionId getConnectionId() const { return connectionId; }
+    MemberId getMemberId() const { return connectionId.getMember(); }
+
+    /** Size of payload data, excluding header. */
+    size_t getSize() const { return size; }
+    /** Size of header + payload. */ 
+    size_t getStoreSize() const { return size + HEADER_SIZE; }
+
+    bool isCluster() const { return connectionId.getNumber() == 0; }
+    bool isConnection() const { return connectionId.getNumber() != 0; }
+    bool isControl() const { return type == CONTROL; }
+
+  protected:
+    static const size_t HEADER_SIZE;
+    
+    EventType type;
+    ConnectionId connectionId;
+    size_t size;
+};
+
+/**
+ * Events are sent to/received from the cluster.
+ * Refcounted so they can be stored on queues.
+ */
+class Event : public EventHeader {
+  public:
+    Event();
+    /** Create an event with a buffer that can hold size bytes plus an event header. */
+    Event(EventType t, const ConnectionId& c, size_t);
+
+    /** Create an event copied from delivered data. */
+    static Event decodeCopy(const MemberId& m, framing::Buffer&);
+
+    /** Create a control event. */
+    static Event control(const framing::AMQBody&, const ConnectionId&);
+
+    /** Create a control event. */
+    static Event control(const framing::AMQFrame&, const ConnectionId&);
+    
+    // Data excluding header.
+    char* getData() { return store + HEADER_SIZE; }
+    const char* getData() const { return store + HEADER_SIZE; }
+
+    // Store including header
+    char* getStore() { return store; }
+    const char* getStore() const { return store; }
+
+    const framing::AMQFrame& getFrame() const;        
+    
+    operator framing::Buffer() const;
+
+    iovec toIovec() const;
+    
+  private:
+    void encodeHeader() const;
+
+    RefCountedBuffer::pointer store;
+    mutable framing::AMQFrame frame;
+};
+
+std::ostream& operator << (std::ostream&, const Event&);
+std::ostream& operator << (std::ostream&, const EventHeader&);
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_EVENT_H*/
diff --git a/cpp/src/qpid/cluster/EventFrame.cpp b/cpp/src/qpid/cluster/EventFrame.cpp
new file mode 100644
index 0000000000..5fbe1fe57c
--- /dev/null
+++ b/cpp/src/qpid/cluster/EventFrame.cpp
@@ -0,0 +1,41 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/EventFrame.h"
+#include "qpid/cluster/Connection.h"
+
+namespace qpid {
+namespace cluster {
+
+EventFrame::EventFrame() {}
+
+EventFrame::EventFrame(const EventHeader& e, const framing::AMQFrame& f, int rc)
+    : connectionId(e.getConnectionId()), frame(f), readCredit(rc), type(e.getType())
+{}
+
+std::ostream& operator<<(std::ostream& o, const EventFrame& e) {
+    if (e.frame.getBody()) o << e.frame;
+    else o << "null-frame";
+    o << " " << e.type << " " << e.connectionId;
+    if (e.readCredit) o << " read-credit=" << e.readCredit;
+    return o;
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/EventFrame.h b/cpp/src/qpid/cluster/EventFrame.h
new file mode 100644
index 0000000000..61447c5525
--- /dev/null
+++ b/cpp/src/qpid/cluster/EventFrame.h
@@ -0,0 +1,60 @@
+#ifndef QPID_CLUSTER_EVENTFRAME_H
+#define QPID_CLUSTER_EVENTFRAME_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/cluster/Event.h"
+#include "qpid/framing/AMQFrame.h"
+#include <boost/intrusive_ptr.hpp>
+#include <iosfwd>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * A frame decoded from an Event.
+ */
+struct EventFrame
+{
+  public:
+    EventFrame();
+
+    EventFrame(const EventHeader& e, const framing::AMQFrame& f, int rc=0);
+
+    bool isCluster() const { return connectionId.getNumber() == 0; }
+    bool isConnection() const { return connectionId.getNumber() != 0; }
+    bool isLastInEvent() const { return readCredit; }
+    MemberId getMemberId() const { return connectionId.getMember(); }
+
+
+    ConnectionId connectionId;
+    framing::AMQFrame frame;   
+    int readCredit; ///< last frame in an event, give credit when processed.
+    EventType type;
+};
+
+std::ostream& operator<<(std::ostream& o, const EventFrame& e);
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_EVENTFRAME_H*/
diff --git a/cpp/src/qpid/cluster/ExpiryPolicy.cpp b/cpp/src/qpid/cluster/ExpiryPolicy.cpp
new file mode 100644
index 0000000000..190eeb7293
--- /dev/null
+++ b/cpp/src/qpid/cluster/ExpiryPolicy.cpp
@@ -0,0 +1,85 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/broker/Message.h"
+#include "qpid/cluster/ExpiryPolicy.h"
+#include "qpid/cluster/Multicaster.h"
+#include "qpid/framing/ClusterMessageExpiredBody.h"
+#include "qpid/sys/Time.h"
+#include "qpid/sys/Timer.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace cluster {
+
+ExpiryPolicy::ExpiryPolicy(Multicaster& m, const MemberId& id, sys::Timer& t)
+    : expiryId(0), expiredPolicy(new Expired), mcast(m), memberId(id), timer(t) {}
+
+struct ExpiryTask : public sys::TimerTask {
+    ExpiryTask(const boost::intrusive_ptr<ExpiryPolicy>& policy, uint64_t id, sys::AbsTime when)
+        : TimerTask(when), expiryPolicy(policy), expiryId(id) {}
+    void fire() { expiryPolicy->sendExpire(expiryId); }
+    boost::intrusive_ptr<ExpiryPolicy> expiryPolicy;
+    const uint64_t expiryId;
+};
+
+void ExpiryPolicy::willExpire(broker::Message& m) {
+    uint64_t id = expiryId++;
+    assert(unexpiredById.find(id) == unexpiredById.end());
+    assert(unexpiredByMessage.find(&m) == unexpiredByMessage.end());
+    unexpiredById[id] = &m;
+    unexpiredByMessage[&m] = id;
+    timer.add(new ExpiryTask(this, id, m.getExpiration()));
+}
+
+void ExpiryPolicy::forget(broker::Message& m) {
+    MessageIdMap::iterator i = unexpiredByMessage.find(&m);
+    assert(i != unexpiredByMessage.end());
+    unexpiredById.erase(i->second);
+    unexpiredByMessage.erase(i);
+}
+
+bool ExpiryPolicy::hasExpired(broker::Message& m) {
+    return unexpiredByMessage.find(&m) == unexpiredByMessage.end();
+}
+
+void ExpiryPolicy::sendExpire(uint64_t id) {
+    mcast.mcastControl(framing::ClusterMessageExpiredBody(framing::ProtocolVersion(), id), memberId);
+}
+
+void ExpiryPolicy::deliverExpire(uint64_t id) {
+    IdMessageMap::iterator i = unexpiredById.find(id);
+    if (i != unexpiredById.end()) {
+        i->second->setExpiryPolicy(expiredPolicy); // hasExpired() == true; 
+        unexpiredByMessage.erase(i->second);
+        unexpiredById.erase(i);
+    }
+}
+
+boost::optional<uint64_t> ExpiryPolicy::getId(broker::Message& m) {
+    MessageIdMap::iterator i = unexpiredByMessage.find(&m);
+    return i == unexpiredByMessage.end() ? boost::optional<uint64_t>() : i->second;
+}
+
+bool ExpiryPolicy::Expired::hasExpired(broker::Message&) { return true; }
+void ExpiryPolicy::Expired::willExpire(broker::Message&) { }
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/ExpiryPolicy.h b/cpp/src/qpid/cluster/ExpiryPolicy.h
new file mode 100644
index 0000000000..bdbe3a61dc
--- /dev/null
+++ b/cpp/src/qpid/cluster/ExpiryPolicy.h
@@ -0,0 +1,89 @@
+#ifndef QPID_CLUSTER_EXPIRYPOLICY_H
+#define QPID_CLUSTER_EXPIRYPOLICY_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/broker/ExpiryPolicy.h"
+#include "qpid/sys/Mutex.h"
+#include <boost/function.hpp>
+#include <boost/intrusive_ptr.hpp>
+#include <boost/optional.hpp>
+#include <map>
+
+namespace qpid {
+
+namespace broker {
+class Message;
+}
+
+namespace sys {
+class Timer;
+}
+
+namespace cluster {
+class Multicaster;
+
+/**
+ * Cluster expiry policy
+ */
+class ExpiryPolicy : public broker::ExpiryPolicy
+{
+  public:
+    ExpiryPolicy(Multicaster&, const MemberId&, sys::Timer&);
+
+    void willExpire(broker::Message&);
+    bool hasExpired(broker::Message&);
+    void forget(broker::Message&);
+
+    // Send expiration notice to cluster.
+    void sendExpire(uint64_t);
+
+    // Cluster delivers expiry notice.
+    void deliverExpire(uint64_t);
+
+    void setId(uint64_t id) { expiryId = id; }
+    uint64_t getId() const { return expiryId; }
+    
+    boost::optional<uint64_t> getId(broker::Message&);
+    
+  private:
+    typedef std::map<broker::Message*,  uint64_t> MessageIdMap;
+    typedef std::map<uint64_t, broker::Message*> IdMessageMap;
+
+    struct Expired : public broker::ExpiryPolicy {
+        bool hasExpired(broker::Message&);
+        void willExpire(broker::Message&);
+    };
+
+    MessageIdMap unexpiredByMessage;
+    IdMessageMap unexpiredById;
+    uint64_t expiryId;
+    boost::intrusive_ptr<Expired> expiredPolicy;
+    Multicaster& mcast;
+    MemberId memberId;
+    sys::Timer& timer;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_EXPIRYPOLICY_H*/
diff --git a/cpp/src/qpid/cluster/FailoverExchange.cpp b/cpp/src/qpid/cluster/FailoverExchange.cpp
new file mode 100644
index 0000000000..e01c41494b
--- /dev/null
+++ b/cpp/src/qpid/cluster/FailoverExchange.cpp
@@ -0,0 +1,99 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/FailoverExchange.h"
+#include "qpid/broker/Message.h"
+#include "qpid/broker/DeliverableMessage.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/framing/MessageProperties.h"
+#include "qpid/framing/AMQFrame.h"
+#include "qpid/framing/AMQHeaderBody.h"
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/log/Statement.h"
+#include "qpid/framing/Array.h"
+#include <boost/bind.hpp>
+#include <algorithm>
+
+namespace qpid {
+namespace cluster {
+using namespace std;
+
+using namespace broker;
+using namespace framing;
+
+const string FailoverExchange::TYPE_NAME("amq.failover");
+
+FailoverExchange::FailoverExchange(management::Manageable* parent) : Exchange(TYPE_NAME, parent) {
+    if (mgmtExchange != 0)
+        mgmtExchange->set_type(TYPE_NAME);
+}
+
+
+void FailoverExchange::setUrls(const vector<Url>& u) {
+    Lock l(lock);
+    urls=u;
+    if (urls.empty()) return;
+    std::for_each(queues.begin(), queues.end(),
+                  boost::bind(&FailoverExchange::sendUpdate, this, _1));
+}
+
+string FailoverExchange::getType() const { return TYPE_NAME; }
+
+bool FailoverExchange::bind(Queue::shared_ptr queue, const string&, const framing::FieldTable*) {
+    Lock l(lock);
+    sendUpdate(queue);
+    return queues.insert(queue).second;
+}
+
+bool FailoverExchange::unbind(Queue::shared_ptr queue, const string&, const framing::FieldTable*) {
+    Lock l(lock);
+    return queues.erase(queue);
+}
+
+bool FailoverExchange::isBound(Queue::shared_ptr queue, const string* const, const framing::FieldTable*) {
+    Lock l(lock);
+    return queues.find(queue) != queues.end();
+}
+
+void FailoverExchange::route(Deliverable&, const string& , const framing::FieldTable* ) {
+    QPID_LOG(warning, "Message received by exchange " << TYPE_NAME << " ignoring");
+}
+
+void FailoverExchange::sendUpdate(const Queue::shared_ptr& queue) {
+    // Called with lock held.
+    if (urls.empty()) return;
+    framing::Array array(0x95);
+    for (Urls::const_iterator i = urls.begin(); i != urls.end(); ++i) 
+        array.add(boost::shared_ptr<Str16Value>(new Str16Value(i->str())));
+    const ProtocolVersion v;
+    boost::intrusive_ptr<Message> msg(new Message);
+    AMQFrame command(MessageTransferBody(v, TYPE_NAME, 1, 0));
+    command.setLastSegment(false);
+    msg->getFrames().append(command);
+    AMQHeaderBody header;
+    header.get<MessageProperties>(true)->setContentLength(0);
+    header.get<MessageProperties>(true)->getApplicationHeaders().setArray(TYPE_NAME, array);
+    AMQFrame headerFrame(header);
+    headerFrame.setFirstSegment(false);
+    msg->getFrames().append(headerFrame);    
+    DeliverableMessage(msg).deliverTo(queue);
+}
+
+}} // namespace cluster
diff --git a/cpp/src/qpid/cluster/FailoverExchange.h b/cpp/src/qpid/cluster/FailoverExchange.h
new file mode 100644
index 0000000000..738cd2a602
--- /dev/null
+++ b/cpp/src/qpid/cluster/FailoverExchange.h
@@ -0,0 +1,68 @@
+#ifndef QPID_CLUSTER_FAILOVEREXCHANGE_H
+#define QPID_CLUSTER_FAILOVEREXCHANGE_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/broker/Exchange.h"
+#include "qpid/broker/DeliverableMessage.h"
+#include "qpid/Url.h"
+
+#include <vector>
+#include <set>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * Failover exchange provides failover host list, as specified in AMQP 0-10.
+ */
+class FailoverExchange : public broker::Exchange
+{
+  public:
+    static const std::string TYPE_NAME;
+
+    FailoverExchange(management::Manageable* parent);
+    
+    void setUrls(const std::vector<Url>&);
+
+    // Exchange overrides
+    std::string getType() const;
+    bool bind(broker::Queue::shared_ptr queue, const std::string& routingKey, const framing::FieldTable* args);
+    bool unbind(broker::Queue::shared_ptr queue, const std::string& routingKey, const framing::FieldTable* args);
+    bool isBound(broker::Queue::shared_ptr queue, const std::string* const routingKey, const framing::FieldTable* const args);
+    void route(broker::Deliverable& msg, const std::string& routingKey, const framing::FieldTable* args);
+
+  private:
+    void sendUpdate(const broker::Queue::shared_ptr&);
+    
+    typedef sys::Mutex::ScopedLock Lock;
+    typedef std::vector<Url> Urls;
+    typedef std::set<broker::Queue::shared_ptr> Queues;
+
+    sys::Mutex lock;
+    Urls urls;
+    Queues queues;
+    
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_FAILOVEREXCHANGE_H*/
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.cpp b/cpp/src/qpid/cluster/InitialStatusMap.cpp
new file mode 100644
index 0000000000..59338f89d4
--- /dev/null
+++ b/cpp/src/qpid/cluster/InitialStatusMap.cpp
@@ -0,0 +1,197 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "InitialStatusMap.h"
+#include "StoreStatus.h"
+#include <algorithm>
+#include <boost/bind.hpp>
+
+namespace qpid {
+namespace cluster {
+
+using namespace std;
+using namespace boost;
+using namespace framing::cluster;
+using namespace framing;
+
+InitialStatusMap::InitialStatusMap(const MemberId& self_, size_t size_)
+    : self(self_), completed(), resendNeeded(), size(size_)
+{}
+
+void InitialStatusMap::configChange(const MemberSet& members) {
+    resendNeeded = false;
+    bool wasComplete = isComplete();
+    if (firstConfig.empty()) firstConfig = members;
+    MemberSet::const_iterator i = members.begin();
+    Map::iterator j = map.begin();
+    while (i != members.end() || j != map.end()) {
+        if (i == members.end()) { // j not in members, member left
+            Map::iterator k = j++;
+            map.erase(k);
+        }
+        else if (j == map.end()) { // i not in map, member joined
+            resendNeeded = true;
+            map[*i] = optional<Status>();
+            ++i;
+        }
+        else if (*i < j->first) { // i not in map, member joined
+            resendNeeded = true;
+            map[*i] = optional<Status>();
+            ++i;
+        }
+        else if (*i > j->first) { // j not in members, member left
+            Map::iterator k = j++;
+            map.erase(k);
+        }
+        else {
+            i++; j++;
+        }
+    }
+    if (resendNeeded) {         // Clear all status
+        for (Map::iterator i = map.begin(); i != map.end(); ++i)
+            i->second = optional<Status>();
+    }
+    completed = isComplete() && !wasComplete; // Set completed on the transition.
+}
+
+void InitialStatusMap::received(const MemberId& m, const Status& s){
+    bool wasComplete = isComplete();
+    map[m] = s;
+    completed = isComplete() && !wasComplete; // Set completed on the transition.
+}
+
+bool InitialStatusMap::notInitialized(const Map::value_type& v) {
+    return !v.second;
+}
+
+bool InitialStatusMap::isComplete() {
+    return !map.empty() && find_if(map.begin(), map.end(), &notInitialized) == map.end()
+        && (map.size() >= size);
+}
+
+bool InitialStatusMap::transitionToComplete() {
+    return completed;
+}
+
+bool InitialStatusMap::isResendNeeded() {
+    bool ret = resendNeeded;
+    resendNeeded = false;
+    return ret;
+}
+
+bool InitialStatusMap::isActive(const Map::value_type& v) {
+    return v.second && v.second->getActive();
+}
+
+bool InitialStatusMap::hasStore(const Map::value_type& v) {
+    return v.second &&
+        (v.second->getStoreState() == STORE_STATE_CLEAN_STORE ||
+         v.second->getStoreState() == STORE_STATE_DIRTY_STORE);
+}
+
+bool InitialStatusMap::isUpdateNeeded() {
+    assert(isComplete());
+    // We need an update if there are any active members.
+    if (find_if(map.begin(), map.end(), &isActive) != map.end()) return true;
+
+    // Otherwise it depends on store status, get my own status:
+    Map::iterator me = map.find(self);
+    assert(me != map.end());
+    assert(me->second);
+    switch (me->second->getStoreState()) {
+      case STORE_STATE_NO_STORE:
+      case STORE_STATE_EMPTY_STORE:
+        // If anybody has a store then we need an update.
+        return find_if(map.begin(), map.end(), &hasStore) != map.end();
+      case STORE_STATE_DIRTY_STORE: return true; 
+      case STORE_STATE_CLEAN_STORE: return false; // Use our own store
+    }
+    return false;
+}
+
+MemberSet InitialStatusMap::getElders() {
+    assert(isComplete());
+    MemberSet elders;
+    // Elders are from first config change, active or higher node-id.
+    for (MemberSet::iterator i = firstConfig.begin(); i != firstConfig.end(); ++i) {
+        if (map.find(*i) != map.end() && (map[*i]->getActive() || *i > self))
+            elders.insert(*i);
+    }
+    return elders;
+}
+
+// Get cluster ID from an active member or the youngest newcomer.
+framing::Uuid InitialStatusMap::getClusterId() {
+    assert(isComplete());
+    assert(!map.empty());
+    Map::iterator i = find_if(map.begin(), map.end(), &isActive);
+    if (i != map.end())
+        return i->second->getClusterId(); // An active member
+    else
+        return map.begin()->second->getClusterId(); // Youngest newcomer in node-id order
+}
+
+void checkId(Uuid& expect, const Uuid& actual, const string& msg) {
+    if (!expect) expect = actual;
+    assert(expect);
+    if (expect != actual)
+        throw Exception(msg);
+}
+
+void InitialStatusMap::checkConsistent() {
+    assert(isComplete());
+    int clean = 0;
+    int dirty = 0;
+    int empty = 0;
+    int none = 0;
+    int active = 0;
+    Uuid clusterId;
+    Uuid shutdownId;
+
+    for (Map::iterator i = map.begin(); i != map.end(); ++i) {
+        if (i->second->getActive()) ++active;
+        switch (i->second->getStoreState()) {
+          case STORE_STATE_NO_STORE: ++none; break;
+          case STORE_STATE_EMPTY_STORE: ++empty; break;
+          case STORE_STATE_DIRTY_STORE:
+            ++dirty;
+            checkId(clusterId, i->second->getClusterId(),
+                    "Cluster-ID mismatch. Stores belong to different clusters.");
+            break;
+          case STORE_STATE_CLEAN_STORE:
+            ++clean;
+            checkId(clusterId, i->second->getClusterId(),
+                    "Cluster-ID mismatch. Stores belong to different clusters.");
+            checkId(shutdownId, i->second->getShutdownId(),
+                    "Shutdown-ID mismatch. Stores were not shut down together");
+            break;
+        }
+    }
+    // Can't mix transient and persistent members.
+    if (none && (clean+dirty+empty))
+        throw Exception("Mixing transient and persistent brokers in a cluster");
+    // If there are no active members and there are dirty stores there
+    // must be at least one clean store.
+    if (!active && dirty && !clean)
+        throw Exception("Cannot recover, no clean store");
+}
+
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.h b/cpp/src/qpid/cluster/InitialStatusMap.h
new file mode 100644
index 0000000000..40fd9ee49d
--- /dev/null
+++ b/cpp/src/qpid/cluster/InitialStatusMap.h
@@ -0,0 +1,75 @@
+#ifndef QPID_CLUSTER_INITIALSTATUSMAP_H
+#define QPID_CLUSTER_INITIALSTATUSMAP_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "MemberSet.h"
+#include <qpid/framing/ClusterInitialStatusBody.h>
+#include <boost/optional.hpp>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * Track status of cluster members during initialization.
+ */
+class InitialStatusMap
+{
+  public:
+    typedef framing::ClusterInitialStatusBody Status;
+
+    InitialStatusMap(const MemberId& self, size_t size);
+    /** Process a config change. @return true if we need to re-send our status */
+    void configChange(const MemberSet& newConfig);
+    /** @return true if we need to re-send status */
+    bool isResendNeeded();
+
+    /** Process received status */
+    void received(const MemberId&, const Status& is);
+
+    /**@return true if the map is complete. */
+    bool isComplete();
+    /**@return true if the map was completed by the last config change or received. */
+    bool transitionToComplete();
+    /**@pre isComplete(). @return this node's elders */
+    MemberSet getElders();
+    /**@pre isComplete(). @return True if we need an update. */
+    bool isUpdateNeeded();
+    /**@pre isComplete(). @return Cluster-wide cluster ID. */
+    framing::Uuid getClusterId();
+    /**@pre isComplete(). @throw Exception if there are any inconsistencies. */
+    void checkConsistent();
+
+  private:
+    typedef std::map<MemberId, boost::optional<Status> > Map;
+    static bool notInitialized(const Map::value_type&);
+    static bool isActive(const Map::value_type&);
+    static bool hasStore(const Map::value_type&);
+    Map map;
+    MemberSet firstConfig;
+    MemberId self;
+    bool completed, resendNeeded;
+    size_t size;
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_INITIALSTATUSMAP_H*/
diff --git a/cpp/src/qpid/cluster/LockedConnectionMap.h b/cpp/src/qpid/cluster/LockedConnectionMap.h
new file mode 100644
index 0000000000..ac744d4f94
--- /dev/null
+++ b/cpp/src/qpid/cluster/LockedConnectionMap.h
@@ -0,0 +1,65 @@
+#ifndef QPID_CLUSTER_LOCKEDCONNECTIONMAP_H
+#define QPID_CLUSTER_LOCKEDCONNECTIONMAP_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/sys/Mutex.h"
+#include "qpid/cluster/Connection.h"
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * Thread safe map of connections.
+ */
+class LockedConnectionMap
+{
+  public:
+    void insert(const ConnectionPtr& c) {
+        sys::Mutex::ScopedLock l(lock);
+        assert(map.find(c->getId()) == map.end());
+        map[c->getId()] = c;
+    }
+    
+    ConnectionPtr getErase(const ConnectionId& c) {
+        sys::Mutex::ScopedLock l(lock);
+        Map::iterator i = map.find(c);
+        if (i != map.end()) {
+            ConnectionPtr cp = i->second;
+            map.erase(i);
+            return cp;
+        }
+        else
+            return 0;
+    }
+
+    void clear() { sys::Mutex::ScopedLock l(lock); map.clear(); }
+
+  private:
+    typedef std::map<ConnectionId, ConnectionPtr> Map;
+    mutable sys::Mutex lock;
+    Map map;
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_LOCKEDCONNECTIONMAP_H*/
diff --git a/cpp/src/qpid/cluster/McastFrameHandler.h b/cpp/src/qpid/cluster/McastFrameHandler.h
new file mode 100644
index 0000000000..17e4c2e9f0
--- /dev/null
+++ b/cpp/src/qpid/cluster/McastFrameHandler.h
@@ -0,0 +1,46 @@
+#ifndef QPID_CLUSTER_MCASTFRAMEHANDLER_H
+#define QPID_CLUSTER_MCASTFRAMEHANDLER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/cluster/Multicaster.h"
+#include "qpid/framing/FrameHandler.h"
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * A frame handler that multicasts frames as CONTROL events.
+ */
+class McastFrameHandler : public framing::FrameHandler
+{
+  public:
+    McastFrameHandler(Multicaster& m, const ConnectionId& cid) : mcast(m), connection(cid) {}
+    void handle(framing::AMQFrame& frame) { mcast.mcastControl(frame, connection); }
+  private:
+    Multicaster& mcast;
+    ConnectionId connection;
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_MCASTFRAMEHANDLER_H*/
diff --git a/cpp/src/qpid/cluster/MemberSet.cpp b/cpp/src/qpid/cluster/MemberSet.cpp
new file mode 100644
index 0000000000..5dc148609f
--- /dev/null
+++ b/cpp/src/qpid/cluster/MemberSet.cpp
@@ -0,0 +1,52 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "MemberSet.h"
+#include <ostream>
+#include <algorithm>
+
+namespace qpid {
+namespace cluster {
+
+MemberSet decodeMemberSet(const std::string& s) {
+    MemberSet set;
+    for (std::string::const_iterator i = s.begin(); i < s.end(); i += 8) {
+        assert(size_t(i-s.begin())+8 <= s.size());
+        set.insert(MemberId(std::string(i, i+8)));
+    }
+    return set;
+}
+
+MemberSet intersection(const MemberSet& a, const MemberSet& b)
+{
+    MemberSet intersection;
+    std::set_intersection(a.begin(), a.end(),
+                          b.begin(), b.end(),
+                          std::inserter(intersection, intersection.begin()));
+    return intersection;
+
+}
+
+std::ostream& operator<<(std::ostream& o, const MemberSet& ms) {
+    copy(ms.begin(), ms.end(), std::ostream_iterator<MemberId>(o, " "));
+    return o;
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/MemberSet.h b/cpp/src/qpid/cluster/MemberSet.h
new file mode 100644
index 0000000000..df3df7c319
--- /dev/null
+++ b/cpp/src/qpid/cluster/MemberSet.h
@@ -0,0 +1,43 @@
+#ifndef QPID_CLUSTER_MEMBERSET_H
+#define QPID_CLUSTER_MEMBERSET_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "types.h"
+#include <set>
+#include <iosfwd>
+
+namespace qpid {
+namespace cluster {
+
+typedef std::set<MemberId> MemberSet;
+
+MemberSet decodeMemberSet(const std::string&);
+
+MemberSet intersection(const MemberSet& a, const MemberSet& b);
+
+std::ostream& operator<<(std::ostream& o, const MemberSet& ms);
+
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_MEMBERSET_H*/
diff --git a/cpp/src/qpid/cluster/Multicaster.cpp b/cpp/src/qpid/cluster/Multicaster.cpp
new file mode 100644
index 0000000000..229d7edb1e
--- /dev/null
+++ b/cpp/src/qpid/cluster/Multicaster.cpp
@@ -0,0 +1,103 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/Multicaster.h"
+#include "qpid/cluster/Cpg.h"
+#include "qpid/log/Statement.h"
+#include "qpid/framing/AMQBody.h"
+#include "qpid/framing/AMQFrame.h"
+
+namespace qpid {
+namespace cluster {
+
+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),
+    ready(false)
+{
+    queue.start();
+}
+
+void Multicaster::mcastControl(const framing::AMQBody& body, const ConnectionId& id) {
+    mcast(Event::control(body, id));
+}
+
+void Multicaster::mcastControl(const framing::AMQFrame& frame, const ConnectionId& id) {
+    mcast(Event::control(frame, id));
+}
+
+void Multicaster::mcastBuffer(const char* data, size_t size, const ConnectionId& id) {
+    Event e(DATA, id, size);
+    memcpy(e.getData(), data, size);
+    mcast(e);
+}
+
+void Multicaster::mcast(const Event& e) {
+    {
+        sys::Mutex::ScopedLock l(lock);
+        if (!ready) {
+            if (e.isConnection()) holdingQueue.push_back(e);
+            else {
+                iovec iov = e.toIovec();
+                // FIXME aconway 2009-11-23: configurable retry --cluster-retry
+                if (!cpg.mcast(&iov, 1))
+                    throw Exception("CPG flow control error during initialization");
+                QPID_LOG(trace, "MCAST (direct) " << e);
+            }
+            return;
+        }
+    }
+    QPID_LOG(trace, "MCAST " << e);
+    queue.push(e);
+}
+
+
+Multicaster::PollableEventQueue::Batch::const_iterator Multicaster::sendMcast(const PollableEventQueue::Batch& values) {
+    try {
+        PollableEventQueue::Batch::const_iterator i = values.begin();
+        while( i != values.end()) {
+            iovec iov = i->toIovec();
+            if (!cpg.mcast(&iov, 1)) {
+                // cpg didn't send because of CPG flow control.
+                break; 
+            }
+            ++i;
+        }
+        return i;
+    }
+    catch (const std::exception& e) {
+        QPID_LOG(critical, "Multicast error: " << e.what());
+        queue.stop();
+        onError();
+        return values.end();
+    }
+}
+
+void Multicaster::setReady() {
+    sys::Mutex::ScopedLock l(lock);
+    ready = true;
+    std::for_each(holdingQueue.begin(), holdingQueue.end(), boost::bind(&Multicaster::mcast, this, _1));
+    holdingQueue.clear();
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/Multicaster.h b/cpp/src/qpid/cluster/Multicaster.h
new file mode 100644
index 0000000000..2db84a9ce0
--- /dev/null
+++ b/cpp/src/qpid/cluster/Multicaster.h
@@ -0,0 +1,87 @@
+#ifndef QPID_CLUSTER_MULTICASTER_H
+#define QPID_CLUSTER_MULTICASTER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/types.h"
+#include "qpid/cluster/Event.h"
+#include "qpid/sys/PollableQueue.h"
+#include "qpid/sys/Mutex.h"
+#include <boost/shared_ptr.hpp>
+#include <deque>
+
+namespace qpid {
+
+namespace sys {
+class Poller;
+}
+
+namespace cluster {
+
+class Cpg;
+
+/**
+ * Multicast to the cluster. Shared, thread safe object.
+ *
+ * Runs in two modes;
+ * 
+ * initializing: Hold connection mcast events. Multicast cluster
+ * events directly in the calling thread. This mode is used before
+ * joining the cluster where the poller may not yet be active and we
+ * want to hold any connection traffic till we join.
+ *
+ * ready: normal operation. Queues all mcasts on a pollable queue,
+ * multicasts connection and cluster events.
+ */
+class Multicaster
+{
+  public:
+    /** Starts in initializing mode. */
+    Multicaster(Cpg& cpg_,
+                const boost::shared_ptr<sys::Poller>&,
+                boost::function<void()> onError
+    );
+    void mcastControl(const framing::AMQBody& controlBody, const ConnectionId&);
+    void mcastControl(const framing::AMQFrame& controlFrame, const ConnectionId&);
+    void mcastBuffer(const char*, size_t, const ConnectionId&);
+    void mcast(const Event& e);
+
+    /** Switch to ready mode. */
+    void setReady();
+
+  private:
+    typedef sys::PollableQueue<Event> PollableEventQueue;
+    typedef std::deque<Event> PlainEventQueue;
+
+    PollableEventQueue::Batch::const_iterator sendMcast(const PollableEventQueue::Batch& );
+
+    sys::Mutex lock;
+    boost::function<void()> onError;
+    Cpg& cpg;
+    PollableEventQueue queue;
+    bool ready;
+    PlainEventQueue holdingQueue;
+    std::vector<struct ::iovec> ioVector;
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_MULTICASTER_H*/
diff --git a/cpp/src/qpid/cluster/ShadowConnectionOutputHandler.h b/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
index 6d429535e6..566a82476e 100644
--- a/cpp/src/qpid/cluster/ShadowConnectionOutputHandler.h
+++ b/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
@@ -1,5 +1,5 @@
-#ifndef QPID_CLUSTER_SHADOWCONNECTIONOUTPUTHANDLER_H
-#define QPID_CLUSTER_SHADOWCONNECTIONOUTPUTHANDLER_H
+#ifndef QPID_CLUSTER_NOOPCONNECTIONOUTPUTHANDLER_H
+#define QPID_CLUSTER_NOOPCONNECTIONOUTPUTHANDLER_H
 
 /*
  *
@@ -30,17 +30,18 @@ namespace framing { class AMQFrame; }
 namespace cluster {
 
 /**
- * Output handler for frames sent to shadow connections.
- * Simply discards frames.
+ * Output handler shadow connections, simply discards frames.
  */
-class ShadowConnectionOutputHandler : public sys::ConnectionOutputHandler
+class NoOpConnectionOutputHandler : public sys::ConnectionOutputHandler
 {
   public:
     virtual void send(framing::AMQFrame&) {}
     virtual void close() {}
+    virtual void abort() {}
     virtual void activateOutput() {}
+    virtual void giveReadCredit(int32_t) {}
 };
 
 }} // namespace qpid::cluster
 
-#endif  /*!QPID_CLUSTER_SHADOWCONNECTIONOUTPUTHANDLER_H*/
+#endif  /*!QPID_CLUSTER_NOOPCONNECTIONOUTPUTHANDLER_H*/
diff --git a/cpp/src/qpid/cluster/Numbering.h b/cpp/src/qpid/cluster/Numbering.h
new file mode 100644
index 0000000000..2d2d931384
--- /dev/null
+++ b/cpp/src/qpid/cluster/Numbering.h
@@ -0,0 +1,70 @@
+#ifndef QPID_CLUSTER_NUMBERING_H
+#define QPID_CLUSTER_NUMBERING_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <map>
+#include <vector>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * A set of numbered T, with two way mapping number->T T->number
+ * Used to construct numberings of objects by code sending and receiving updates.
+ */
+template <class T> class Numbering
+{
+  public:
+    size_t size() const { return byNumber.size(); }
+
+    size_t add(const T& t) {
+        size_t n = (*this)[t];  // Already in the set?
+        if (n == size()) {
+            byObject[t] = n;
+            byNumber.push_back(t);
+        }
+        return n;
+    }
+
+    void clear() { byObject.clear(); byNumber.clear(); }
+
+    /**@return object at index n or T() if n > size() */
+    T operator[](size_t n) const { return(n < size()) ? byNumber[n] : T(); }
+
+    /**@return index of t or size() if t is not in the map */
+    size_t operator[](const T& t) const {
+        typename Map::const_iterator i = byObject.find(t);
+        return (i != byObject.end()) ? i->second : size();
+    }
+
+    bool contains(const T& t) const { return (*this)[t] == size(); }
+
+  private:
+    typedef std::map<T, size_t> Map;
+    Map byObject;
+    std::vector<T> byNumber;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_NUMBERING_H*/
diff --git a/cpp/src/qpid/cluster/OutputInterceptor.cpp b/cpp/src/qpid/cluster/OutputInterceptor.cpp
new file mode 100644
index 0000000000..cb75fe5561
--- /dev/null
+++ b/cpp/src/qpid/cluster/OutputInterceptor.cpp
@@ -0,0 +1,116 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/OutputInterceptor.h"
+#include "qpid/cluster/Connection.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/framing/ClusterConnectionDeliverDoOutputBody.h"
+#include "qpid/framing/AMQFrame.h"
+#include "qpid/log/Statement.h"
+#include <boost/current_function.hpp>
+
+
+namespace qpid {
+namespace cluster {
+
+using namespace framing;
+using namespace std;
+
+NoOpConnectionOutputHandler OutputInterceptor::discardHandler;
+
+OutputInterceptor::OutputInterceptor(Connection& p, sys::ConnectionOutputHandler& h)
+    : parent(p), closing(false), next(&h), sendMax(1), sent(0), sentDoOutput(false)
+{}
+
+void OutputInterceptor::send(framing::AMQFrame& f) {
+    sys::Mutex::ScopedLock l(lock);
+    next->send(f);
+}
+
+void OutputInterceptor::activateOutput() {
+    if (parent.isCatchUp()) {
+        sys::Mutex::ScopedLock l(lock);
+        next->activateOutput();
+    }
+    else
+        sendDoOutput(sendMax);
+}
+
+void OutputInterceptor::abort() {
+    sys::Mutex::ScopedLock l(lock);
+    if (parent.isLocal()) {
+        next->abort();
+    }
+}
+
+void OutputInterceptor::giveReadCredit(int32_t credit) {
+    sys::Mutex::ScopedLock l(lock);
+    next->giveReadCredit(credit);
+}
+
+// Called in write thread when the IO layer has no more data to write.
+// We do nothing in the write thread, we run doOutput only on delivery
+// of doOutput requests.
+bool OutputInterceptor::doOutput() { return false; }
+
+// Send output up to limit, calculate new limit. 
+void OutputInterceptor::deliverDoOutput(uint32_t limit) {
+    sentDoOutput = false;
+    sendMax = limit;
+    size_t newLimit = limit;
+    if (parent.isLocal()) {
+        size_t buffered = getBuffered();
+        if (buffered == 0 && sent == sendMax) // Could have sent more, increase the limit.
+            newLimit = sendMax*2; 
+        else if (buffered > 0 && sent > 1) // Data left unsent, reduce the limit.
+            newLimit = sent-1;
+    }
+    sent = 0;
+    while (sent < limit && parent.getBrokerConnection().doOutput())
+        ++sent;
+    if (sent == limit) sendDoOutput(newLimit);
+}
+
+void OutputInterceptor::sendDoOutput(size_t newLimit) {
+    if (parent.isLocal() && !sentDoOutput && !closing) {
+        sentDoOutput = true;
+        parent.getCluster().getMulticast().mcastControl(
+            ClusterConnectionDeliverDoOutputBody(ProtocolVersion(), newLimit),
+            parent.getId());
+    }
+}
+
+void OutputInterceptor::closeOutput() {
+    sys::Mutex::ScopedLock l(lock);
+    closing = true;
+    next = &discardHandler;
+}
+
+void OutputInterceptor::close() {
+    sys::Mutex::ScopedLock l(lock);
+    next->close();
+}
+
+size_t OutputInterceptor::getBuffered() const {
+    sys::Mutex::ScopedLock l(lock);
+    return next->getBuffered();
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/OutputInterceptor.h b/cpp/src/qpid/cluster/OutputInterceptor.h
new file mode 100644
index 0000000000..65bd82a4fc
--- /dev/null
+++ b/cpp/src/qpid/cluster/OutputInterceptor.h
@@ -0,0 +1,79 @@
+#ifndef QPID_CLUSTER_OUTPUTINTERCEPTOR_H
+#define QPID_CLUSTER_OUTPUTINTERCEPTOR_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/NoOpConnectionOutputHandler.h"
+#include "qpid/sys/ConnectionOutputHandler.h"
+#include "qpid/sys/Mutex.h"
+#include "qpid/broker/ConnectionFactory.h"
+#include <boost/function.hpp>
+
+namespace qpid {
+namespace framing { class AMQFrame; }
+namespace cluster {
+
+class Connection;
+
+/**
+ * Interceptor for connection OutputHandler, manages outgoing message replication.
+ */
+class OutputInterceptor : public sys::ConnectionOutputHandler {
+  public:
+    OutputInterceptor(cluster::Connection& p, sys::ConnectionOutputHandler& h);
+
+    // sys::ConnectionOutputHandler functions
+    void send(framing::AMQFrame& f);
+    void abort();
+    void activateOutput();
+    void giveReadCredit(int32_t);
+    void close();
+    size_t getBuffered() const;
+
+    // Delivery point for doOutput requests.
+    void deliverDoOutput(uint32_t limit);
+    // Intercept doOutput requests on Connection.
+    bool doOutput();
+
+    void closeOutput();
+
+    uint32_t getSendMax() const { return sendMax; }
+    void setSendMax(uint32_t sendMax_) { sendMax=sendMax_; }
+    
+    cluster::Connection& parent;
+    
+  private:
+    typedef sys::Mutex::ScopedLock Locker;
+
+    void sendDoOutput(size_t newLimit);
+
+    mutable sys::Mutex lock;
+    bool closing;
+    sys::ConnectionOutputHandler* next;
+    static NoOpConnectionOutputHandler discardHandler;
+    uint32_t sendMax, sent;
+    bool sentDoOutput;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_OUTPUTINTERCEPTOR_H*/
diff --git a/cpp/src/qpid/cluster/PollableCondition.cpp b/cpp/src/qpid/cluster/PollableCondition.cpp
deleted file mode 100644
index eecf95ff8d..0000000000
--- a/cpp/src/qpid/cluster/PollableCondition.cpp
+++ /dev/null
@@ -1,100 +0,0 @@
-#ifndef QPID_SYS_LINUX_POLLABLECONDITION_CPP
-#define QPID_SYS_LINUX_POLLABLECONDITION_CPP
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-// FIXME aconway 2008-08-11: this could be of more general interest,
-// move to common lib.
-// 
-
-#include "qpid/sys/posix/PrivatePosix.h"
-#include "qpid/cluster/PollableCondition.h"
-#include "qpid/Exception.h"
-
-#include <unistd.h>
-#include <fcntl.h>
-
-namespace qpid {
-namespace cluster {
-
-PollableCondition::PollableCondition() : IOHandle(new sys::IOHandlePrivate) {
-    int fds[2];
-    if (::pipe(fds) == -1)
-        throw ErrnoException(QPID_MSG("Can't create PollableCondition"));
-    impl->fd = fds[0];
-    writeFd = fds[1];
-    if (::fcntl(impl->fd, F_SETFL, O_NONBLOCK) == -1)
-        throw ErrnoException(QPID_MSG("Can't create PollableCondition"));
-    if (::fcntl(writeFd, F_SETFL, O_NONBLOCK) == -1)
-        throw ErrnoException(QPID_MSG("Can't create PollableCondition"));
-}
-
-bool PollableCondition::clear() {
-    char buf[256];
-    ssize_t n;
-    bool wasSet = false;
-    while ((n = ::read(impl->fd, buf, sizeof(buf))) > 0) 
-        wasSet = true;
-    if (n == -1 && errno != EAGAIN) throw ErrnoException(QPID_MSG("Error clearing PollableCondition"));
-    return wasSet;
-}
-
-void PollableCondition::set() {
-    static const char dummy=0;
-    ssize_t n = ::write(writeFd, &dummy, 1);
-    if (n == -1 && errno != EAGAIN) throw ErrnoException("Error setting PollableCondition");
-}
-
-
-#if 0
-// FIXME aconway 2008-08-12: More efficient Linux implementation using
-// eventfd system call.  Do a configure.ac test to enable this when
-// eventfd is available.
-
-#include <sys/eventfd.h>
-
-namespace qpid {
-namespace cluster {
-
-PollableCondition::PollableCondition() : IOHandle(new sys::IOHandlePrivate) {
-    impl->fd = ::eventfd(0, 0);
-    if (impl->fd < 0) throw ErrnoException("conditionfd() failed");
-}
-
-bool PollableCondition::clear() {
-    char buf[8];
-    ssize_t n = ::read(impl->fd, buf, 8);
-    if (n != 8) throw ErrnoException("read failed on conditionfd");
-    return *reinterpret_cast<uint64_t*>(buf);
-}
-
-void PollableCondition::set() {
-    static const uint64_t value=1;
-    ssize_t n = ::write(impl->fd, reinterpret_cast<const void*>(&value), 8);
-    if (n != 8) throw ErrnoException("write failed on conditionfd");
-}
-    
-#endif
-
-}} // namespace qpid::cluster
-
-#endif  /*!QPID_SYS_LINUX_POLLABLECONDITION_CPP*/
diff --git a/cpp/src/qpid/cluster/PollableQueue.h b/cpp/src/qpid/cluster/PollableQueue.h
index 0bba2ba790..65f615d8b6 100644
--- a/cpp/src/qpid/cluster/PollableQueue.h
+++ b/cpp/src/qpid/cluster/PollableQueue.h
@@ -22,78 +22,53 @@
  *
  */
 
-#include "qpid/cluster/PollableCondition.h"
-#include "qpid/sys/Dispatcher.h"
-#include "qpid/sys/Mutex.h"
-#include <boost/function.hpp>
-#include <boost/bind.hpp>
-#include <deque>
+#include "qpid/sys/PollableQueue.h"
+#include  <qpid/log/Statement.h>
 
 namespace qpid {
-
-namespace sys { class Poller; }
-
 namespace cluster {
 
-// FIXME aconway 2008-08-11: this could be of more general interest,
-// move to common lib.
-
 /**
- * A queue that can be polled by sys::Poller.  Any thread can push to
- * the queue, on wakeup the poller thread processes all items on the
- * queue by passing them to a callback in a batch.
+ * More convenient version of PollableQueue that handles iterating
+ * over the batch and error handling.
  */
-template <class T>
-class PollableQueue {
-    typedef std::deque<T> Queue;
-
+template <class T> class PollableQueue : public sys::PollableQueue<T> {
   public:
-    typedef typename Queue::iterator iterator;
-    
-    /** Callback to process a range of items. */
-    typedef boost::function<void (const iterator&, const iterator&)> Callback;
-
-    /** When the queue is selected by the poller, values are passed to callback cb. */
-    explicit PollableQueue(const Callback& cb);
-
-    /** Push a value onto the queue. Thread safe */
-    void push(const T& t) { ScopedLock l(lock); queue.push_back(t); condition.set(); }
+    typedef boost::function<void (const T&)> Callback;
+    typedef boost::function<void()> ErrorCallback;
+
+    PollableQueue(Callback f, ErrorCallback err, const std::string& msg,
+                  const boost::shared_ptr<sys::Poller>& poller)
+        : sys::PollableQueue<T>(boost::bind(&PollableQueue<T>::handleBatch, this, _1),
+                                poller),
+          callback(f), error(err), message(msg)
+    {}
+
+    typename sys::PollableQueue<T>::Batch::const_iterator
+    handleBatch(const typename sys::PollableQueue<T>::Batch& values) {
+        try {
+            typename sys::PollableQueue<T>::Batch::const_iterator i = values.begin();
+            while (i != values.end() && !this->isStopped()) {
+                callback(*i);
+                ++i;
+            }
+            return i;
+        }
+        catch (const std::exception& e) {
+            QPID_LOG(error, message << ": " << e.what());
+            this->stop();
+            error();
+            return values.end();
+        }
+    }
 
-    /** Start polling. */ 
-    void start(const boost::shared_ptr<sys::Poller>& poller) { handle.startWatch(poller); }
-
-    /** Stop polling. */
-    void stop() { handle.stopWatch(); }
-    
   private:
-    typedef sys::Mutex::ScopedLock ScopedLock;
-    typedef sys::Mutex::ScopedUnlock ScopedUnlock;
-
-    void dispatch(sys::DispatchHandle&);
-    
-    sys::Mutex lock;
     Callback callback;
-    PollableCondition condition;
-    sys::DispatchHandle handle;
-    Queue queue;
-    Queue batch;
+    ErrorCallback error;
+    std::string message;
 };
 
-template <class T> PollableQueue<T>::PollableQueue(const Callback& cb) // FIXME aconway 2008-08-12: 
-    : callback(cb),
-      handle(condition, boost::bind(&PollableQueue<T>::dispatch, this, _1), 0, 0)
-{}
-
-template <class T> void PollableQueue<T>::dispatch(sys::DispatchHandle& h) {
-    ScopedLock l(lock);         // Lock for concurrent push() 
-    batch.clear();
-    batch.swap(queue);
-    condition.clear();
-    ScopedUnlock u(lock);
-    callback(batch.begin(), batch.end()); // Process the batch outside the lock.
-    h.rewatch();
-}
-
+    
 }} // namespace qpid::cluster
 
 #endif  /*!QPID_CLUSTER_POLLABLEQUEUE_H*/
diff --git a/cpp/src/qpid/cluster/PollerDispatch.cpp b/cpp/src/qpid/cluster/PollerDispatch.cpp
new file mode 100644
index 0000000000..a839ef863b
--- /dev/null
+++ b/cpp/src/qpid/cluster/PollerDispatch.cpp
@@ -0,0 +1,67 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/PollerDispatch.h"
+
+#include "qpid/log/Statement.h"
+#include <boost/bind.hpp>
+
+namespace qpid {
+namespace cluster {
+
+PollerDispatch::PollerDispatch(Cpg& c, boost::shared_ptr<sys::Poller> p,
+                               boost::function<void()> e)
+    : cpg(c), poller(p), onError(e),
+      dispatchHandle(cpg,
+                     boost::bind(&PollerDispatch::dispatch, this, _1), // read
+                     0,         // write
+                     boost::bind(&PollerDispatch::disconnect, this, _1) // disconnect
+      ),
+      started(false)
+{}
+    
+PollerDispatch::~PollerDispatch() {
+    if (started)
+        dispatchHandle.stopWatch();
+}
+
+void PollerDispatch::start() {
+    dispatchHandle.startWatch(poller);
+    started = true;
+}
+
+// Entry point: called by IO to dispatch CPG events.
+void PollerDispatch::dispatch(sys::DispatchHandle& h) {
+    try {
+        cpg.dispatchAll();
+        h.rewatch();
+    } catch (const std::exception& e) {
+        QPID_LOG(critical, "Error in cluster dispatch: " << e.what());
+        onError();
+    }
+}
+
+// Entry point: called if disconnected from  CPG.
+void PollerDispatch::disconnect(sys::DispatchHandle& ) {
+    QPID_LOG(critical, "Disconnected from cluster");
+    onError();
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/PollableCondition.h b/cpp/src/qpid/cluster/PollerDispatch.h
index 6bfca6cabe..63801e0de9 100644
--- a/cpp/src/qpid/cluster/PollableCondition.h
+++ b/cpp/src/qpid/cluster/PollerDispatch.h
@@ -1,5 +1,5 @@
-#ifndef QPID_SYS_POLLABLECONDITION_H
-#define QPID_SYS_POLLABLECONDITION_H
+#ifndef QPID_CLUSTER_POLLERDISPATCH_H
+#define QPID_CLUSTER_POLLERDISPATCH_H
 
 /*
  *
@@ -22,39 +22,39 @@
  *
  */
 
-#include "qpid/sys/IOHandle.h"
-
-// FIXME aconway 2008-08-11: this could be of more general interest,
-// move to sys namespace in common lib. 
-// 
+#include "qpid/cluster/Cpg.h"
+#include "qpid/sys/Poller.h"
+#include "qpid/sys/DispatchHandle.h"
+#include <boost/function.hpp>
 
 namespace qpid {
 namespace cluster {
 
 /**
- * A pollable condition to integrate in-process conditions with IO
- * conditions in a polling loop.
- *
- * Setting the condition makes it readable for a poller.
- * 
- * Writable/disconnected conditions are undefined and should not be
- * polled for.
+ * Dispatch CPG events via the poller.
  */
-class PollableCondition : public sys::IOHandle {
+class PollerDispatch  {
   public:
-    PollableCondition();
+    PollerDispatch(Cpg&, boost::shared_ptr<sys::Poller> poller,
+                   boost::function<void()> onError) ;
 
-    /** Set the condition, triggers readable in a poller. */ 
-    void set();
+    ~PollerDispatch();
 
-    /** Get the current state of the condition, then clear it.
-     *@return The state of the condition before it was cleared.
-     */
-    bool clear();
+    void start();
 
   private:
-    int writeFd;
+    // Poller callbacks
+    void dispatch(sys::DispatchHandle&); // Dispatch CPG events.
+    void disconnect(sys::DispatchHandle&); // CPG was disconnected
+
+    Cpg& cpg;
+    boost::shared_ptr<sys::Poller> poller;
+    boost::function<void()> onError;
+    sys::DispatchHandleRef dispatchHandle;
+    bool started;
+
+
 };
 }} // namespace qpid::cluster
 
-#endif  /*!QPID_SYS_POLLABLECONDITION_H*/
+#endif  /*!QPID_CLUSTER_POLLERDISPATCH_H*/
diff --git a/cpp/src/qpid/cluster/ProxyInputHandler.h b/cpp/src/qpid/cluster/ProxyInputHandler.h
new file mode 100644
index 0000000000..228f8d092d
--- /dev/null
+++ b/cpp/src/qpid/cluster/ProxyInputHandler.h
@@ -0,0 +1,57 @@
+#ifndef QPID_CLUSTER_PROXYINPUTHANDLER_H
+#define QPID_CLUSTER_PROXYINPUTHANDLER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/sys/ConnectionInputHandler.h"
+#include <boost/intrusive_ptr.hpp>
+
+namespace qpid {
+
+namespace framing { class AMQFrame; }
+
+namespace cluster {
+
+/**
+ * Proxies ConnectionInputHandler functions and ensures target.closed() 
+ * is called, on deletion if not before.
+ */
+class ProxyInputHandler : public sys::ConnectionInputHandler
+{
+  public:
+    ProxyInputHandler(boost::intrusive_ptr<cluster::Connection> t) : target(t) {}
+    ~ProxyInputHandler() { closed(); }
+    
+    void received(framing::AMQFrame& f) { target->received(f); }
+    void closed() { if (target) target->closed(); target = 0; }
+    void idleOut() { target->idleOut(); }
+    void idleIn() { target->idleIn(); }
+    bool doOutput() { return target->doOutput(); }
+    bool hasOutput() { return target->hasOutput(); }
+    
+  private:
+    boost::intrusive_ptr<cluster::Connection> target;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_PROXYINPUTHANDLER_H*/
diff --git a/cpp/src/qpid/cluster/Quorum.h b/cpp/src/qpid/cluster/Quorum.h
new file mode 100644
index 0000000000..bbfa473f94
--- /dev/null
+++ b/cpp/src/qpid/cluster/Quorum.h
@@ -0,0 +1,32 @@
+#ifndef QPID_CLUSTER_QUORUM_H
+#define QPID_CLUSTER_QUORUM_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "config.h"
+
+#if HAVE_LIBCMAN_H
+#include "qpid/cluster/Quorum_cman.h"
+#else
+#include "qpid/cluster/Quorum_null.h"
+#endif
+
+#endif  /*!QPID_CLUSTER_QUORUM_H*/
diff --git a/cpp/src/qpid/cluster/Quorum_cman.cpp b/cpp/src/qpid/cluster/Quorum_cman.cpp
new file mode 100644
index 0000000000..507d9649b9
--- /dev/null
+++ b/cpp/src/qpid/cluster/Quorum_cman.cpp
@@ -0,0 +1,103 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/Quorum_cman.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/log/Statement.h"
+#include "qpid/Options.h"
+#include "qpid/sys/Time.h"
+#include "qpid/sys/posix/PrivatePosix.h"
+
+namespace qpid {
+namespace cluster {
+
+namespace {
+
+boost::function<void()> errorFn;
+
+void cmanCallbackFn(cman_handle_t handle, void */*privdata*/, int reason, int /*arg*/) {
+    if (reason == CMAN_REASON_STATECHANGE && !cman_is_quorate(handle)) {
+        QPID_LOG(critical, "Lost contact with cluster quorum.");
+        if (errorFn) errorFn();
+        cman_stop_notification(handle);
+    }
+}
+}
+
+Quorum::Quorum(boost::function<void()> err) : enable(false), cman(0), cmanFd(0) {
+    errorFn = err;
+}
+
+Quorum::~Quorum() {
+    dispatchHandle.reset();
+    if (cman) cman_finish(cman);
+}
+
+void Quorum::start(boost::shared_ptr<sys::Poller> p) {
+    poller = p;
+    enable = true;
+    QPID_LOG(debug, "Connecting to quorum service.");
+    cman = cman_init(0);
+    if (cman == 0) throw ErrnoException("Can't connect to cman service");
+    if (!cman_is_quorate(cman)) {
+        QPID_LOG(notice, "Waiting for cluster quorum.");
+        while(!cman_is_quorate(cman)) sys::sleep(5);
+    }
+    int err = cman_start_notification(cman, cmanCallbackFn);
+    if (err != 0) throw ErrnoException("Can't register for cman notifications");
+    watch(getFd());
+}
+
+void Quorum::watch(int fd) {
+    cmanFd = fd;
+    dispatchHandle.reset(
+        new sys::DispatchHandleRef(
+            sys::PosixIOHandle(cmanFd),
+            boost::bind(&Quorum::dispatch, this, _1), // read
+            0, // write
+            boost::bind(&Quorum::disconnect, this, _1)  // disconnect
+        ));
+    dispatchHandle->startWatch(poller);
+}
+
+int Quorum::getFd() {
+    int fd = cman_get_fd(cman);
+    if (fd == 0) throw ErrnoException("Can't get cman file descriptor");
+    return fd;
+}
+
+void Quorum::dispatch(sys::DispatchHandle&) {
+    try {
+        cman_dispatch(cman, CMAN_DISPATCH_ALL);
+        int fd = getFd();
+        if (fd != cmanFd) watch(fd);
+    } catch (const std::exception& e) {
+        QPID_LOG(critical, "Error in quorum dispatch: " << e.what());
+        errorFn();
+    }
+}
+
+void Quorum::disconnect(sys::DispatchHandle&) {
+    QPID_LOG(critical, "Disconnected from quorum service");
+    errorFn();
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/Quorum_cman.h b/cpp/src/qpid/cluster/Quorum_cman.h
new file mode 100644
index 0000000000..130f1baf64
--- /dev/null
+++ b/cpp/src/qpid/cluster/Quorum_cman.h
@@ -0,0 +1,64 @@
+#ifndef QPID_CLUSTER_QUORUM_CMAN_H
+#define QPID_CLUSTER_QUORUM_CMAN_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <qpid/sys/DispatchHandle.h>
+#include <boost/function.hpp>
+#include <boost/shared_ptr.hpp>
+#include <memory>
+
+extern "C" {
+#include <libcman.h>
+}
+
+namespace qpid {
+namespace sys {
+class Poller;
+}
+
+namespace cluster {
+class Cluster;
+
+class Quorum {
+  public:
+    Quorum(boost::function<void ()> onError);
+    ~Quorum();
+    void start(boost::shared_ptr<sys::Poller>);
+    
+  private:
+    void dispatch(sys::DispatchHandle&);
+    void disconnect(sys::DispatchHandle&);
+    int getFd();
+    void watch(int fd);
+    
+    bool enable;
+    cman_handle_t cman;
+    int cmanFd;
+    std::auto_ptr<sys::DispatchHandleRef> dispatchHandle;
+    boost::shared_ptr<sys::Poller> poller;
+};
+
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_QUORUM_CMAN_H*/
diff --git a/cpp/src/qpid/cluster/Quorum_null.h b/cpp/src/qpid/cluster/Quorum_null.h
new file mode 100644
index 0000000000..dc27f0a43b
--- /dev/null
+++ b/cpp/src/qpid/cluster/Quorum_null.h
@@ -0,0 +1,42 @@
+#ifndef QPID_CLUSTER_QUORUM_NULL_H
+#define QPID_CLUSTER_QUORUM_NULL_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <boost/shared_ptr.hpp>
+#include <boost/function.hpp>
+
+namespace qpid {
+namespace cluster {
+class Cluster;
+
+/** Null implementation of quorum. */
+
+class Quorum {
+  public:
+    Quorum(boost::function<void ()>) {}
+    void start(boost::shared_ptr<sys::Poller>) {}
+};
+
+#endif  /*!QPID_CLUSTER_QUORUM_NULL_H*/
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/RetractClient.cpp b/cpp/src/qpid/cluster/RetractClient.cpp
new file mode 100644
index 0000000000..7d9f52fc39
--- /dev/null
+++ b/cpp/src/qpid/cluster/RetractClient.cpp
@@ -0,0 +1,61 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/RetractClient.h"
+#include "qpid/cluster/UpdateClient.h"
+#include "qpid/framing/ClusterConnectionRetractOfferBody.h"
+#include "qpid/client/ConnectionAccess.h"
+#include "qpid/client/ConnectionImpl.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace cluster {
+
+using namespace framing;
+
+namespace {
+
+struct AutoClose {
+    client::Connection& connection;
+    AutoClose(client::Connection& c) : connection(c) {}
+    ~AutoClose() { connection.close(); }
+};
+}
+
+RetractClient::RetractClient(const Url& u, const client::ConnectionSettings& cs)
+    : url(u), connectionSettings(cs)
+{}
+
+RetractClient::~RetractClient() { delete this; }
+
+
+void RetractClient::run() {
+    try {
+        client::Connection c = UpdateClient::catchUpConnection();
+        c.open(url, connectionSettings);
+        AutoClose ac(c);
+        AMQFrame retract((ClusterConnectionRetractOfferBody()));
+        client::ConnectionAccess::getImpl(c)->handle(retract);
+    } catch (const std::exception& e) {
+        QPID_LOG(error, " while retracting retract to " << url << ": " << e.what()); 
+    }
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/RetractClient.h b/cpp/src/qpid/cluster/RetractClient.h
new file mode 100644
index 0000000000..fb896197cc
--- /dev/null
+++ b/cpp/src/qpid/cluster/RetractClient.h
@@ -0,0 +1,49 @@
+#ifndef QPID_CLUSTER_RETRACTCLIENT_H
+#define QPID_CLUSTER_RETRACTCLIENT_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/client/ConnectionSettings.h"
+#include "qpid/sys/Runnable.h"
+
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * A client that retracts an offer to a remote broker using AMQP. @see UpdateClient
+ */
+class RetractClient : public sys::Runnable {
+  public:
+
+    RetractClient(const Url&, const client::ConnectionSettings&);
+    ~RetractClient();
+    void run();                 // Will delete this when finished.
+
+  private:
+    Url url;
+    client::ConnectionSettings connectionSettings;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_RETRACTCLIENT_H*/
diff --git a/cpp/src/qpid/cluster/StoreStatus.cpp b/cpp/src/qpid/cluster/StoreStatus.cpp
new file mode 100644
index 0000000000..6e412c23f7
--- /dev/null
+++ b/cpp/src/qpid/cluster/StoreStatus.cpp
@@ -0,0 +1,110 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "StoreStatus.h"
+#include "qpid/Exception.h"
+#include <boost/filesystem/path.hpp>
+#include <boost/filesystem/fstream.hpp>
+#include <boost/filesystem/operations.hpp>
+#include <fstream>
+
+namespace qpid {
+namespace cluster {
+
+using framing::Uuid;
+using namespace framing::cluster;
+namespace fs=boost::filesystem;
+using std::ostream;
+
+StoreStatus::StoreStatus(const std::string& d)
+    : state(STORE_STATE_NO_STORE), dataDir(d)
+{}
+
+namespace {
+
+const char* SUBDIR="cluster";
+const char* CLUSTER_ID_FILE="cluster.uuid";
+const char* SHUTDOWN_ID_FILE="shutdown.uuid";
+
+Uuid loadUuid(const fs::path& path) {
+    Uuid ret;
+    if (exists(path)) {
+        fs::ifstream i(path);
+        i >> ret;
+    }
+    return ret;
+}
+
+void saveUuid(const fs::path& path, const Uuid& uuid) {
+    fs::ofstream o(path);
+    o << uuid;
+}
+
+} // namespace
+
+
+void StoreStatus::load() {
+    fs::path dir = fs::path(dataDir, fs::native)/SUBDIR;
+    create_directory(dir);
+    clusterId = loadUuid(dir/CLUSTER_ID_FILE);
+    shutdownId = loadUuid(dir/SHUTDOWN_ID_FILE);
+
+    if (clusterId && shutdownId) state = STORE_STATE_CLEAN_STORE;
+    else if (clusterId) state = STORE_STATE_DIRTY_STORE;
+    else state = STORE_STATE_EMPTY_STORE;
+}
+
+void StoreStatus::save() {
+    fs::path dir = fs::path(dataDir, fs::native)/SUBDIR;
+    create_directory(dir);
+    saveUuid(dir/CLUSTER_ID_FILE, clusterId);
+    saveUuid(dir/SHUTDOWN_ID_FILE, shutdownId);
+}
+
+void StoreStatus::dirty(const Uuid& clusterId_) {
+    clusterId = clusterId_;
+    shutdownId = Uuid();
+    state = STORE_STATE_DIRTY_STORE;
+    save();
+}
+
+void StoreStatus::clean(const Uuid& shutdownId_) {
+    state = STORE_STATE_CLEAN_STORE;
+    shutdownId = shutdownId_;
+    save();
+}
+
+ostream& operator<<(ostream& o, const StoreStatus& s) {
+    switch (s.getState()) {
+      case STORE_STATE_NO_STORE: o << "no store"; break;
+      case STORE_STATE_EMPTY_STORE: o << "empty store"; break;
+      case STORE_STATE_DIRTY_STORE:
+        o << "dirty store, cluster-id=" << s.getClusterId();
+        break;
+      case STORE_STATE_CLEAN_STORE:
+        o << "clean store, cluster-id=" << s.getClusterId()
+          << " shutdown-id=" << s.getShutdownId();
+        break;
+    }
+    return o;
+}
+
+}} // namespace qpid::cluster
+
diff --git a/cpp/src/qpid/cluster/StoreStatus.h b/cpp/src/qpid/cluster/StoreStatus.h
new file mode 100644
index 0000000000..522020ed69
--- /dev/null
+++ b/cpp/src/qpid/cluster/StoreStatus.h
@@ -0,0 +1,64 @@
+#ifndef QPID_CLUSTER_STORESTATE_H
+#define QPID_CLUSTER_STORESTATE_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/framing/Uuid.h"
+#include "qpid/framing/enum.h"
+#include <iosfwd>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * State of the store for cluster purposes.
+ */
+class StoreStatus
+{
+  public:
+    typedef framing::Uuid Uuid;
+    typedef framing::cluster::StoreState StoreState;
+
+    StoreStatus(const std::string& dir);
+
+    framing::cluster::StoreState getState() const { return state; }
+    const Uuid& getClusterId() const { return clusterId; }
+    const Uuid& getShutdownId() const { return shutdownId; }
+
+    void dirty(const Uuid& start); // Start using the store.
+    void clean(const Uuid& stop); // Stop using the store.
+
+    void load();
+    void save();
+
+    bool hasStore() { return state != framing::cluster::STORE_STATE_NO_STORE; }
+
+  private:
+    framing::cluster::StoreState state;
+    Uuid clusterId, shutdownId;
+    std::string dataDir;
+};
+
+std::ostream& operator<<(std::ostream&, const StoreStatus&);
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_STORESTATE_H*/
diff --git a/cpp/src/qpid/cluster/UpdateClient.cpp b/cpp/src/qpid/cluster/UpdateClient.cpp
new file mode 100644
index 0000000000..b20cc907a2
--- /dev/null
+++ b/cpp/src/qpid/cluster/UpdateClient.cpp
@@ -0,0 +1,484 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/cluster/UpdateClient.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/cluster/ClusterMap.h"
+#include "qpid/cluster/Connection.h"
+#include "qpid/cluster/Decoder.h"
+#include "qpid/cluster/ExpiryPolicy.h"
+#include "qpid/client/SessionBase_0_10Access.h" 
+#include "qpid/client/ConnectionAccess.h" 
+#include "qpid/client/SessionImpl.h" 
+#include "qpid/client/ConnectionImpl.h"
+#include "qpid/client/Future.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/QueueRegistry.h"
+#include "qpid/broker/Message.h"
+#include "qpid/broker/Exchange.h"
+#include "qpid/broker/ExchangeRegistry.h"
+#include "qpid/broker/SessionHandler.h"
+#include "qpid/broker/SessionState.h"
+#include "qpid/broker/TxOpVisitor.h"
+#include "qpid/broker/DtxAck.h"
+#include "qpid/broker/TxAccept.h"
+#include "qpid/broker/TxPublish.h"
+#include "qpid/broker/RecoveredDequeue.h"
+#include "qpid/broker/RecoveredEnqueue.h"
+#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/framing/TypeCode.h"
+#include "qpid/log/Statement.h"
+#include "qpid/Url.h"
+#include <boost/bind.hpp>
+#include <boost/cast.hpp>
+#include <algorithm>
+
+namespace qpid {
+namespace cluster {
+
+using broker::Broker;
+using broker::Exchange;
+using broker::Queue;
+using broker::QueueBinding;
+using broker::Message;
+using broker::SemanticState;
+
+using namespace framing;
+namespace arg=client::arg;
+using client::SessionBase_0_10Access;
+
+struct ClusterConnectionProxy : public AMQP_AllProxy::ClusterConnection {
+    ClusterConnectionProxy(client::Connection c) :
+        AMQP_AllProxy::ClusterConnection(*client::ConnectionAccess::getImpl(c)) {}
+    ClusterConnectionProxy(client::AsyncSession s) :
+        AMQP_AllProxy::ClusterConnection(SessionBase_0_10Access(s).get()->out) {}
+};
+
+// Create a connection with special version that marks it as a catch-up connection.
+client::Connection UpdateClient::catchUpConnection() {
+    client::Connection c;
+    client::ConnectionAccess::setVersion(c, ProtocolVersion(0x80 , 0x80 + 10));
+    return c;
+}
+
+// Send a control body directly to the session.
+void send(client::AsyncSession& s, const AMQBody& body) {
+    client::SessionBase_0_10Access sb(s);
+    sb.get()->send(body);
+}
+
+// TODO aconway 2008-09-24: optimization: update connections/sessions in parallel.
+
+UpdateClient::UpdateClient(const MemberId& updater, const MemberId& updatee, const Url& url,
+                           broker::Broker& broker, const ClusterMap& m, ExpiryPolicy& expiry_, 
+                           const Cluster::ConnectionVector& cons, Decoder& decoder_,
+                           const boost::function<void()>& ok,
+                           const boost::function<void(const std::exception&)>& fail,
+                           const client::ConnectionSettings& cs
+)
+    : updaterId(updater), updateeId(updatee), updateeUrl(url), updaterBroker(broker), map(m),
+      expiry(expiry_), connections(cons), decoder(decoder_),
+      connection(catchUpConnection()), shadowConnection(catchUpConnection()),
+      done(ok), failed(fail), connectionSettings(cs)
+{}
+
+UpdateClient::~UpdateClient() {}
+
+// Reserved exchange/queue name for catch-up, avoid clashes with user queues/exchanges.
+const std::string UpdateClient::UPDATE("qpid.cluster-update");
+
+void UpdateClient::run() {
+    try {
+        connection.open(updateeUrl, connectionSettings);
+        session = connection.newSession(UPDATE);
+        update();
+        done();
+    } catch (const std::exception& e) {
+        failed(e);
+    }
+    delete this;
+}
+
+void UpdateClient::update() {
+    QPID_LOG(debug, updaterId << " updating state to " << updateeId
+             << " at " << updateeUrl);
+    Broker& b = updaterBroker;
+    b.getExchanges().eachExchange(boost::bind(&UpdateClient::updateExchange, this, _1));
+    b.getQueues().eachQueue(boost::bind(&UpdateClient::updateNonExclusiveQueue, this, _1));
+
+    // Update queue is used to transfer acquired messages that are no
+    // longer on their original queue.
+    session.queueDeclare(arg::queue=UPDATE, arg::autoDelete=true);
+    session.sync();
+    std::for_each(connections.begin(), connections.end(), boost::bind(&UpdateClient::updateConnection, this, _1));
+    session.queueDelete(arg::queue=UPDATE);
+    session.close();
+
+    // Update queue listeners: must come after sessions so consumerNumbering is populated.
+    b.getQueues().eachQueue(boost::bind(&UpdateClient::updateQueueListeners, this, _1));
+
+    ClusterConnectionProxy(session).expiryId(expiry.getId());
+    ClusterConnectionMembershipBody membership;
+    map.toMethodBody(membership);
+    AMQFrame frame(membership);
+    client::ConnectionAccess::getImpl(connection)->handle(frame);
+
+    connection.close();
+    QPID_LOG(debug,  updaterId << " update completed to " << updateeId
+             << " at " << updateeUrl << ": " << membership);
+}
+
+namespace {
+template <class T> std::string encode(const T& t) {
+    std::string encoded;
+    encoded.resize(t.encodedSize());
+    framing::Buffer buf(const_cast<char*>(encoded.data()), encoded.size());
+    t.encode(buf);
+    return encoded;
+}
+} // namespace
+
+void UpdateClient::updateExchange(const boost::shared_ptr<Exchange>& ex) {
+    QPID_LOG(debug, updaterId << " updating exchange " << ex->getName());
+    ClusterConnectionProxy(session).exchange(encode(*ex));
+}
+
+/** Bind a queue to the update exchange and update messges to it
+ * setting the message possition as needed.
+ */
+class MessageUpdater {
+    std::string queue;
+    bool haveLastPos;
+    framing::SequenceNumber lastPos;
+    client::AsyncSession session;
+    ExpiryPolicy& expiry;
+    
+  public:
+
+    MessageUpdater(const string& q, const client::AsyncSession s, ExpiryPolicy& expiry_) : queue(q), haveLastPos(false), session(s), expiry(expiry_) {
+        session.exchangeBind(queue, UpdateClient::UPDATE);
+    }
+
+    ~MessageUpdater() {
+        try {
+            session.exchangeUnbind(queue, UpdateClient::UPDATE);
+        }
+        catch (const std::exception& e) {
+            // Don't throw in a destructor.
+            QPID_LOG(error, "Unbinding update queue " << queue << ": " << e.what());
+        }
+    }
+
+
+    void updateQueuedMessage(const broker::QueuedMessage& message) {
+        // Send the queue position if necessary.
+        if (!haveLastPos || message.position - lastPos != 1)  {
+            ClusterConnectionProxy(session).queuePosition(queue, message.position.getValue()-1);
+            haveLastPos = true;
+        }
+        lastPos = message.position;
+
+        // Send the expiry ID if necessary.
+        if (message.payload->getProperties<DeliveryProperties>()->getTtl()) {
+            boost::optional<uint64_t> expiryId = expiry.getId(*message.payload);
+            if (!expiryId) return; // Message already expired, don't replicate.
+            ClusterConnectionProxy(session).expiryId(*expiryId);
+        }
+
+        // We can't send a broker::Message via the normal client API,
+        // and it would be expensive to copy it into a client::Message
+        // so we go a bit under the client API covers here.
+        //
+        SessionBase_0_10Access sb(session);
+        // Disable client code that clears the delivery-properties.exchange
+        sb.get()->setDoClearDeliveryPropertiesExchange(false);
+        framing::MessageTransferBody transfer(
+            *message.payload->getFrames().as<framing::MessageTransferBody>());
+        transfer.setDestination(UpdateClient::UPDATE);
+        
+        sb.get()->send(transfer, message.payload->getFrames(),
+                       !message.payload->isContentReleased());
+        if (message.payload->isContentReleased()){
+            uint16_t maxFrameSize = sb.get()->getConnection()->getNegotiatedSettings().maxFrameSize;
+            uint16_t maxContentSize = maxFrameSize - AMQFrame::frameOverhead();
+            bool morecontent = true;
+            for (uint64_t offset = 0; morecontent; offset += maxContentSize)
+            {            
+                AMQFrame frame((AMQContentBody()));
+                morecontent = message.payload->getContentFrame(*(message.queue), frame, maxContentSize, offset);
+                sb.get()->sendRawFrame(frame);
+            }
+        }
+    }
+
+    void updateMessage(const boost::intrusive_ptr<broker::Message>& message) {
+        updateQueuedMessage(broker::QueuedMessage(0, message, haveLastPos? lastPos.getValue()+1 : 1));
+    }
+};
+
+void UpdateClient::updateQueue(client::AsyncSession& s, const boost::shared_ptr<Queue>& q) {
+    broker::Exchange::shared_ptr alternateExchange = q->getAlternateExchange();
+    s.queueDeclare(
+        arg::queue = q->getName(),
+        arg::durable = q->isDurable(),
+        arg::autoDelete = q->isAutoDelete(),
+        arg::alternateExchange = alternateExchange ? alternateExchange->getName() : "",
+        arg::arguments = q->getSettings(),
+        arg::exclusive = q->hasExclusiveOwner()
+    );
+    MessageUpdater updater(q->getName(), s, expiry);
+    q->eachMessage(boost::bind(&MessageUpdater::updateQueuedMessage, &updater, _1));
+    q->eachBinding(boost::bind(&UpdateClient::updateBinding, this, s, q->getName(), _1));
+    ClusterConnectionProxy(s).queuePosition(q->getName(), q->getPosition());
+}
+
+void UpdateClient::updateExclusiveQueue(const boost::shared_ptr<broker::Queue>& q) {
+    QPID_LOG(debug, updaterId << " updating exclusive queue " << q->getName() << " on " << shadowSession.getId());
+    updateQueue(shadowSession, q);
+}
+
+void UpdateClient::updateNonExclusiveQueue(const boost::shared_ptr<broker::Queue>& q) {
+    if (!q->hasExclusiveOwner()) {
+        QPID_LOG(debug, updaterId << " updating queue " << q->getName());
+        updateQueue(session, q);
+    }//else queue will be updated as part of session state of owning session
+}
+
+void UpdateClient::updateBinding(client::AsyncSession& s, const std::string& queue, const QueueBinding& binding) {
+    s.exchangeBind(queue, binding.exchange, binding.key, binding.args);
+}
+
+void UpdateClient::updateOutputTask(const sys::OutputTask* task) {
+    const SemanticState::ConsumerImpl* cci =
+        boost::polymorphic_downcast<const SemanticState::ConsumerImpl*> (task);
+    SemanticState::ConsumerImpl* ci = const_cast<SemanticState::ConsumerImpl*>(cci);
+    uint16_t channel =  ci->getParent().getSession().getChannel();
+    ClusterConnectionProxy(shadowConnection).outputTask(channel,  ci->getName());
+    QPID_LOG(debug, updaterId << " updating output task " << ci->getName()
+             << " channel=" << channel);
+}
+
+void UpdateClient::updateConnection(const boost::intrusive_ptr<Connection>& updateConnection) {
+    QPID_LOG(debug, updaterId << " updating connection " << *updateConnection);
+    shadowConnection = catchUpConnection();
+
+    broker::Connection& bc = updateConnection->getBrokerConnection();
+    connectionSettings.maxFrameSize = bc.getFrameMax();
+    shadowConnection.open(updateeUrl, connectionSettings);
+    bc.eachSessionHandler(boost::bind(&UpdateClient::updateSession, this, _1));
+    // Safe to use decoder here because we are stalled for update.
+    std::pair<const char*, size_t> fragment = decoder.get(updateConnection->getId()).getFragment();
+    bc.getOutputTasks().eachOutput(
+        boost::bind(&UpdateClient::updateOutputTask, this, _1));
+    ClusterConnectionProxy(shadowConnection).shadowReady(
+        updateConnection->getId().getMember(),
+        updateConnection->getId().getNumber(),
+        bc.getUserId(),
+        string(fragment.first, fragment.second),
+        updateConnection->getOutput().getSendMax()
+    );
+    shadowConnection.close();
+    QPID_LOG(debug, updaterId << " updated connection " << *updateConnection);
+}
+
+void UpdateClient::updateSession(broker::SessionHandler& sh) {
+    broker::SessionState* ss = sh.getSession();
+    if (!ss) return;            // no session.
+
+    QPID_LOG(debug, updaterId << " updating session " << &sh.getConnection()
+             << "[" << sh.getChannel() << "] = " << ss->getId());
+
+    // Create a client session to update session state. 
+    boost::shared_ptr<client::ConnectionImpl> cimpl = client::ConnectionAccess::getImpl(shadowConnection);
+    boost::shared_ptr<client::SessionImpl> simpl = cimpl->newSession(ss->getId().getName(), ss->getTimeout(), sh.getChannel());
+    simpl->disableAutoDetach();
+    client::SessionBase_0_10Access(shadowSession).set(simpl);
+    AMQP_AllProxy::ClusterConnection proxy(simpl->out);
+
+    // Re-create session state on remote connection.
+
+    QPID_LOG(debug, updaterId << " updating exclusive queues.");
+    ss->getSessionAdapter().eachExclusiveQueue(boost::bind(&UpdateClient::updateExclusiveQueue, this, _1));
+
+    QPID_LOG(debug, updaterId << " updating consumers.");
+    ss->getSemanticState().eachConsumer(
+        boost::bind(&UpdateClient::updateConsumer, this, _1));
+
+    QPID_LOG(debug, updaterId << " updating unacknowledged messages.");
+    broker::DeliveryRecords& drs = ss->getSemanticState().getUnacked();
+    std::for_each(drs.begin(), drs.end(),  boost::bind(&UpdateClient::updateUnacked, this, _1));
+
+    updateTxState(ss->getSemanticState());           // Tx transaction state.
+
+    // Adjust command counter for message in progress, will be sent after state update.
+    boost::intrusive_ptr<Message> inProgress = ss->getMessageInProgress();
+    SequenceNumber received = ss->receiverGetReceived().command;
+    if (inProgress)  
+        --received;
+             
+    // Reset command-sequence state.
+    proxy.sessionState(
+        ss->senderGetReplayPoint().command,
+        ss->senderGetCommandPoint().command,
+        ss->senderGetIncomplete(),
+        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);
+    }
+    QPID_LOG(debug, updaterId << " updated session " << sh.getSession()->getId());
+}
+
+void UpdateClient::updateConsumer(
+    const broker::SemanticState::ConsumerImpl::shared_ptr& ci)
+{
+    QPID_LOG(debug, updaterId << " updating consumer " << ci->getName() << " on "
+             << shadowSession.getId());
+
+    using namespace message;
+    shadowSession.messageSubscribe(
+        arg::queue       = ci->getQueue()->getName(),
+        arg::destination = ci->getName(),
+        arg::acceptMode  = ci->isAckExpected() ? ACCEPT_MODE_EXPLICIT : ACCEPT_MODE_NONE,
+        arg::acquireMode = ci->isAcquire() ? ACQUIRE_MODE_PRE_ACQUIRED : ACQUIRE_MODE_NOT_ACQUIRED,
+        arg::exclusive   = ci->isExclusive(),
+        arg::resumeId    = ci->getResumeId(),
+        arg::resumeTtl   = ci->getResumeTtl(),
+        arg::arguments   = ci->getArguments()
+    );
+    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());
+    ClusterConnectionProxy(shadowSession).consumerState(
+        ci->getName(),
+        ci->isBlocked(),
+        ci->isNotifyEnabled(),
+        ci->position
+    );
+    consumerNumbering.add(ci);
+
+    QPID_LOG(debug, updaterId << " updated consumer " << ci->getName()
+             << " on " << shadowSession.getId());
+}
+    
+void UpdateClient::updateUnacked(const broker::DeliveryRecord& dr) {
+    if (!dr.isEnded() && dr.isAcquired() && dr.getMessage().payload) {
+        // If the message is acquired then it is no longer on the
+        // updatees queue, put it on the update queue for updatee to pick up.
+        //
+        MessageUpdater(UPDATE, shadowSession, expiry).updateQueuedMessage(dr.getMessage());
+    }
+    ClusterConnectionProxy(shadowSession).deliveryRecord(
+        dr.getQueue()->getName(),
+        dr.getMessage().position,
+        dr.getTag(),
+        dr.getId(),
+        dr.isAcquired(),
+        dr.isAccepted(),
+        dr.isCancelled(),
+        dr.isComplete(),
+        dr.isEnded(),
+        dr.isWindowing(),
+        dr.getQueue()->isEnqueued(dr.getMessage()),
+        dr.getCredit()
+    );
+}
+
+class TxOpUpdater : public broker::TxOpConstVisitor, public MessageUpdater {
+  public:
+    TxOpUpdater(UpdateClient& dc, client::AsyncSession s, ExpiryPolicy& expiry)
+        : MessageUpdater(UpdateClient::UPDATE, s, expiry), parent(dc), session(s), proxy(s) {}
+
+    void operator()(const broker::DtxAck& ) {
+        throw InternalErrorException("DTX transactions not currently supported by cluster.");
+    }
+    
+    void operator()(const broker::RecoveredDequeue& rdeq) {
+        updateMessage(rdeq.getMessage());
+        proxy.txEnqueue(rdeq.getQueue()->getName());
+    }
+
+    void operator()(const broker::RecoveredEnqueue& renq) {
+        updateMessage(renq.getMessage());
+        proxy.txEnqueue(renq.getQueue()->getName());
+    }
+
+    void operator()(const broker::TxAccept& txAccept) {
+        proxy.txAccept(txAccept.getAcked());
+    }
+
+    void operator()(const broker::TxPublish& txPub) {
+        updateMessage(txPub.getMessage());
+        typedef std::list<Queue::shared_ptr> QueueList;
+        const QueueList& qlist = txPub.getQueues();
+        Array qarray(TYPE_CODE_STR8);
+        for (QueueList::const_iterator i = qlist.begin(); i != qlist.end(); ++i) 
+            qarray.push_back(Array::ValuePtr(new Str8Value((*i)->getName())));
+        proxy.txPublish(qarray, txPub.delivered);
+    }
+
+  private:
+    UpdateClient& parent;
+    client::AsyncSession session;
+    ClusterConnectionProxy proxy;
+};
+    
+void UpdateClient::updateTxState(broker::SemanticState& s) {
+    QPID_LOG(debug, updaterId << " updating TX transaction state.");
+    ClusterConnectionProxy proxy(shadowSession);
+    proxy.accumulatedAck(s.getAccumulatedAck());
+    broker::TxBuffer::shared_ptr txBuffer = s.getTxBuffer();
+    if (txBuffer) {
+        proxy.txStart();
+        TxOpUpdater updater(*this, shadowSession, expiry);
+        txBuffer->accept(updater);
+        proxy.txEnd();
+    }
+}
+
+void UpdateClient::updateQueueListeners(const boost::shared_ptr<broker::Queue>& queue) {
+    queue->getListeners().eachListener(
+        boost::bind(&UpdateClient::updateQueueListener, this, queue->getName(), _1));
+}
+
+void UpdateClient::updateQueueListener(std::string& q,
+                                  const boost::shared_ptr<broker::Consumer>& c)
+{
+    const boost::shared_ptr<SemanticState::ConsumerImpl> ci =
+        boost::dynamic_pointer_cast<SemanticState::ConsumerImpl>(c);
+    size_t n = consumerNumbering[ci];
+    if (n >= consumerNumbering.size())
+        throw Exception(QPID_MSG("Unexpected listener on queue " << q));
+    ClusterConnectionProxy(session).addQueueListener(q, n);
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/UpdateClient.h b/cpp/src/qpid/cluster/UpdateClient.h
new file mode 100644
index 0000000000..29ef5f9df2
--- /dev/null
+++ b/cpp/src/qpid/cluster/UpdateClient.h
@@ -0,0 +1,119 @@
+#ifndef QPID_CLUSTER_UPDATECLIENT_H
+#define QPID_CLUSTER_UPDATECLIENT_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/ClusterMap.h"
+#include "qpid/cluster/Numbering.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/ConnectionSettings.h"
+#include "qpid/client/AsyncSession.h"
+#include "qpid/broker/SemanticState.h"
+#include "qpid/sys/Runnable.h"
+#include <boost/shared_ptr.hpp>
+
+
+namespace qpid {
+
+class Url;
+
+namespace broker {
+
+class Broker;
+class Queue;
+class Exchange;
+class QueueBindings;
+class QueueBinding;
+class QueuedMessage;
+class SessionHandler;
+class DeliveryRecord;
+class SessionState;
+class SemanticState;
+class Decoder;
+
+} // namespace broker
+
+namespace cluster {
+
+class Cluster;
+class Connection;
+class ClusterMap;
+class Decoder;
+class ExpiryPolicy;
+
+/**
+ * A client that updates the contents of a local broker to a remote one using AMQP.
+ */
+class UpdateClient : public sys::Runnable {
+  public:
+    static const std::string UPDATE; // Name for special update queue and exchange.
+    static client::Connection catchUpConnection();
+    
+    UpdateClient(const MemberId& updater, const MemberId& updatee, const Url&,
+                 broker::Broker& donor, const ClusterMap& map, ExpiryPolicy& expiry,
+                 const std::vector<boost::intrusive_ptr<Connection> >&, Decoder&,
+                 const boost::function<void()>& done,
+                 const boost::function<void(const std::exception&)>& fail,
+                 const client::ConnectionSettings& 
+    );
+
+    ~UpdateClient();
+    void update();
+    void run();                 // Will delete this when finished.
+
+    void updateUnacked(const broker::DeliveryRecord&);
+
+  private:
+    void updateQueue(client::AsyncSession&, const boost::shared_ptr<broker::Queue>&);
+    void updateNonExclusiveQueue(const boost::shared_ptr<broker::Queue>&);
+    void updateExclusiveQueue(const boost::shared_ptr<broker::Queue>&);
+    void updateExchange(const boost::shared_ptr<broker::Exchange>&);
+    void updateMessage(const broker::QueuedMessage&);
+    void updateMessageTo(const broker::QueuedMessage&, const std::string& queue, client::Session s);
+    void updateBinding(client::AsyncSession&, const std::string& queue, const broker::QueueBinding& binding);
+    void updateConnection(const boost::intrusive_ptr<Connection>& connection);
+    void updateSession(broker::SessionHandler& s);
+    void updateTxState(broker::SemanticState& s);
+    void updateOutputTask(const sys::OutputTask* task);
+    void updateConsumer(const broker::SemanticState::ConsumerImpl::shared_ptr&);
+    void updateQueueListeners(const boost::shared_ptr<broker::Queue>&);
+    void updateQueueListener(std::string& q, const boost::shared_ptr<broker::Consumer>& c);
+
+    Numbering<broker::SemanticState::ConsumerImpl::shared_ptr> consumerNumbering;
+    MemberId updaterId;
+    MemberId updateeId;
+    Url updateeUrl;
+    broker::Broker& updaterBroker;
+    ClusterMap map;
+    ExpiryPolicy& expiry;
+    std::vector<boost::intrusive_ptr<Connection> > connections;
+    Decoder& decoder;
+    client::Connection connection, shadowConnection;
+    client::AsyncSession session, shadowSession;
+    boost::function<void()> done;
+    boost::function<void(const std::exception& e)> failed;
+    client::ConnectionSettings connectionSettings;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_UPDATECLIENT_H*/
diff --git a/cpp/src/qpid/cluster/UpdateExchange.cpp b/cpp/src/qpid/cluster/UpdateExchange.cpp
new file mode 100644
index 0000000000..11937f296f
--- /dev/null
+++ b/cpp/src/qpid/cluster/UpdateExchange.cpp
@@ -0,0 +1,47 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/broker/Message.h"
+#include "UpdateExchange.h"
+
+namespace qpid {
+namespace cluster {
+
+using framing::MessageTransferBody;
+using framing::DeliveryProperties;
+
+UpdateExchange::UpdateExchange(management::Manageable* parent)
+    : broker::Exchange(UpdateClient::UPDATE, parent),
+      broker::FanOutExchange(UpdateClient::UPDATE, parent) {}
+
+
+void UpdateExchange::setProperties(const boost::intrusive_ptr<broker::Message>& msg) {
+    MessageTransferBody* transfer = msg->getMethod<MessageTransferBody>();
+    assert(transfer);
+    const DeliveryProperties* props = msg->getProperties<DeliveryProperties>();
+    assert(props);
+    if (props->hasExchange())
+        transfer->setDestination(props->getExchange());
+    else
+        transfer->clearDestinationFlag();
+}
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/UpdateExchange.h b/cpp/src/qpid/cluster/UpdateExchange.h
new file mode 100644
index 0000000000..9d7d9ee5fc
--- /dev/null
+++ b/cpp/src/qpid/cluster/UpdateExchange.h
@@ -0,0 +1,45 @@
+#ifndef QPID_CLUSTER_UPDATEEXCHANGE_H
+#define QPID_CLUSTER_UPDATEEXCHANGE_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/cluster/UpdateClient.h"
+#include "qpid/broker/FanOutExchange.h"
+
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * A keyless exchange (like fanout exchange) that does not modify
+ * delivery-properties.exchange but copies it to the MessageTransfer.
+ */
+class UpdateExchange : public broker::FanOutExchange
+{
+  public:
+    UpdateExchange(management::Manageable* parent);
+    void setProperties(const boost::intrusive_ptr<broker::Message>&);
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_UPDATEEXCHANGE_H*/
diff --git a/cpp/src/qpid/cluster/UpdateReceiver.h b/cpp/src/qpid/cluster/UpdateReceiver.h
new file mode 100644
index 0000000000..cc1ce0da8d
--- /dev/null
+++ b/cpp/src/qpid/cluster/UpdateReceiver.h
@@ -0,0 +1,42 @@
+#ifndef QPID_CLUSTER_UPDATESTATE_H
+#define QPID_CLUSTER_UPDATESTATE_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "Numbering.h"
+#include "qpid/broker/SemanticState.h"
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * Cluster-wide state used when receiving an update.
+ */
+class UpdateReceiver {
+  public:
+    /** Numbering used to identify Queue listeners as consumers */ 
+    typedef Numbering<boost::shared_ptr<broker::SemanticState::ConsumerImpl> > ConsumerNumbering;
+    ConsumerNumbering consumerNumbering;
+};
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_UPDATESTATE_H*/
diff --git a/cpp/src/qpid/cluster/WatchDogPlugin.cpp b/cpp/src/qpid/cluster/WatchDogPlugin.cpp
new file mode 100644
index 0000000000..fc2b830dac
--- /dev/null
+++ b/cpp/src/qpid/cluster/WatchDogPlugin.cpp
@@ -0,0 +1,136 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+/**@file
+
+   The watchdog plug-in will kill the qpidd broker process if it
+   becomes stuck for longer than a configured interval.
+
+   If the watchdog plugin is loaded and the --watchdog-interval=N
+   option is set then the broker starts a watchdog process and signals
+   it every N/2 seconds.
+
+   The watchdog process runs a very simple program that starts a timer
+   for N seconds, and resets the timer to N seconds whenever it is
+   signalled by the broker. If the timer ever reaches 0 the watchdog
+   kills the broker process (with kill -9) and exits.
+
+   This is useful in a cluster setting because in some insttances
+   (e.g. while resolving an error) it's possible for a stuck process
+   to hang other cluster members that are waiting for it to send a
+   message.  Using the watchdog, the stuck process is terminated and 
+   removed fromt the cluster allowing other members to continue and
+   clients of the stuck process to fail over to other members.
+
+*/
+#include "qpid/Plugin.h"
+#include "qpid/Options.h"
+#include "qpid/log/Statement.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/sys/Timer.h"
+#include "qpid/sys/Fork.h"
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <signal.h>
+
+namespace qpid {
+namespace cluster {
+
+using broker::Broker;
+
+struct Settings {
+    Settings() : interval(0) {}
+    int interval;
+};
+
+struct WatchDogOptions : public qpid::Options {
+    Settings& settings;
+
+    WatchDogOptions(Settings& s) : settings(s) {
+        addOptions()
+            ("watchdog-interval", optValue(settings.interval, "N"),
+             "broker is automatically killed if it is hung for more than \
+	      N seconds. 0 disables watchdog.");
+    }
+};
+
+struct WatchDogTask : public sys::TimerTask {
+    int pid;
+    sys::Timer& timer;
+    int interval;
+
+    WatchDogTask(int pid_, sys::Timer& t, int _interval)
+        : TimerTask(_interval*sys::TIME_SEC/2), pid(pid_), timer(t), interval(_interval) {}
+
+    void fire() {
+        timer.add (new WatchDogTask(pid, timer, interval));
+        QPID_LOG(debug, "Sending keepalive signal to watchdog");
+        ::kill(pid, SIGUSR1);
+    }
+};
+
+struct WatchDogPlugin : public qpid::Plugin, public qpid::sys::Fork {
+    Settings settings;
+    WatchDogOptions options;
+    Broker* broker;
+    int watchdogPid;
+
+    WatchDogPlugin() : options(settings), broker(0), watchdogPid(0) {}
+
+    ~WatchDogPlugin() {
+        if (watchdogPid) ::kill(watchdogPid, SIGTERM);
+        ::waitpid(watchdogPid, 0, 0);
+    }
+
+    Options* getOptions() { return &options; }
+
+    void earlyInitialize(qpid::Plugin::Target& target) {
+        broker = dynamic_cast<Broker*>(&target);
+        if (broker && settings.interval) {
+            QPID_LOG(notice, "Starting watchdog process with interval of " <<
+                     settings.interval << " seconds");
+            fork();
+        }
+    }
+
+    void initialize(Target&) {}
+
+  protected:
+
+    void child() {              // Child of fork
+        const char* watchdog = ::getenv("QPID_WATCHDOG_EXEC"); // For use in tests
+        if (!watchdog) watchdog=QPID_EXEC_DIR "/qpidd_watchdog";
+        std::string interval = boost::lexical_cast<std::string>(settings.interval);
+        ::execl(watchdog, watchdog, interval.c_str(), NULL);
+        QPID_LOG(critical, "Failed to exec watchdog program " << watchdog );
+        ::kill(::getppid(), SIGKILL);
+        exit(1);
+    }
+
+    void parent(int pid) {          // Parent of fork
+        watchdogPid = pid;
+        broker->getTimer().add(
+            new WatchDogTask(watchdogPid, broker->getTimer(), settings.interval));
+        // TODO aconway 2009-08-10: to be extra safe, we could monitor
+        // the watchdog child and re-start it if it exits.
+    }
+};
+
+static WatchDogPlugin instance; // Static initialization.
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/management-schema.xml b/cpp/src/qpid/cluster/management-schema.xml
new file mode 100644
index 0000000000..a6292e9113
--- /dev/null
+++ b/cpp/src/qpid/cluster/management-schema.xml
@@ -0,0 +1,61 @@
+<schema package="org.apache.qpid.cluster">
+
+  <!--
+      Licensed to the Apache Software Foundation (ASF) under one
+      or more contributor license agreements.  See the NOTICE file
+      distributed with this work for additional information
+      regarding copyright ownership.  The ASF licenses this file
+      to you under the Apache License, Version 2.0 (the
+      "License"); you may not use this file except in compliance
+      with the License.  You may obtain a copy of the License at
+      
+      http://www.apache.org/licenses/LICENSE-2.0
+      
+      Unless required by applicable law or agreed to in writing,
+      software distributed under the License is distributed on an
+      "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+      KIND, either express or implied.  See the License for the
+      specific language governing permissions and limitations
+      under the License.
+  -->
+
+  <!-- Type information:
+
+Numeric types with "_wm" suffix are watermarked numbers.  These are compound
+values containing a current value, and a low and high water mark for the reporting
+interval.  The low and high water marks are set to the current value at the
+beginning of each interval and track the minimum and maximum values of the statistic
+over the interval respectively.
+
+Access rights for configuration elements:
+
+RO => Read Only
+RC => Read/Create, can be set at create time only, read-only thereafter
+RW => Read/Write
+
+If access rights are omitted for a property, they are assumed to be RO.
+
+  -->
+
+  <class name="Cluster">
+    <property name="brokerRef"        type="objId"  references="Broker" access="RC" index="y" parentRef="y"/>
+    <property name="clusterName"      type="sstr"   access="RC" desc="Name of cluster this server is a member of"/>
+    <property name="clusterID"        type="sstr"   access="RO" desc="Globally unique ID (UUID) for this cluster instance"/>
+    <property name="memberID"         type="sstr"   access="RO" desc="ID of this member of the cluster"/>
+    <property name="publishedURL"     type="sstr"   access="RC" desc="URL this node advertizes itself as"/>
+    <property name="clusterSize"      type="uint16" access="RO" desc="Number of brokers currently in the cluster"/>
+    <property name="status"           type="sstr"   access="RO" desc="Cluster node status (STALLED,ACTIVE,JOINING)"/>
+    <property name="members"          type="lstr"   access="RO" desc="List of member URLs delimited by ';'"/> 
+    <property name="memberIDs"        type="lstr"   access="RO" desc="List of member IDs delimited by ';'"/> 
+
+    <method name="stopClusterNode">
+      <arg name="brokerId" type="sstr" dir="I"/>
+    </method>
+    <method name="stopFullCluster"/>
+
+  </class>
+
+
+
+</schema>
+
diff --git a/cpp/src/qpid/cluster/qpidd_watchdog.cpp b/cpp/src/qpid/cluster/qpidd_watchdog.cpp
new file mode 100644
index 0000000000..51c5ed4b3f
--- /dev/null
+++ b/cpp/src/qpid/cluster/qpidd_watchdog.cpp
@@ -0,0 +1,63 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+/** @file helper executable for WatchDogPlugin.cpp */
+
+#include <sys/types.h>
+#include <sys/time.h>
+#include <signal.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <limits.h>
+
+long timeout;
+
+void killParent(int) {
+    ::kill(getppid(), SIGKILL);
+    ::fprintf(stderr, "Watchdog killed unresponsive broker, pid=%d\n", ::getppid());
+    ::exit(1);
+}
+
+void resetTimer(int) {
+    struct ::itimerval itval = { { 0, 0 }, { timeout, 0 } };
+    if (::setitimer(ITIMER_REAL, &itval, 0) !=0) {
+        ::perror("Watchdog failed to set timer");
+        killParent(0);
+        ::exit(1);
+    }
+}
+
+/** Simple watchdog program: kill parent process if timeout
+ * expires without a SIGUSR1.
+ * Will be killed with SIGHUP when parent shuts down.
+ * Args: timeout in seconds.
+ */
+int main(int argc, char** argv) {
+    if(argc != 2 || (timeout = atoi(argv[1])) == 0) {
+        ::fprintf(stderr, "Usage: %s <timeout_seconds>\n", argv[0]);
+        ::exit(1);
+    }
+    ::signal(SIGUSR1, resetTimer);
+    ::signal(SIGALRM, killParent);
+    resetTimer(0);
+    while (true) { sleep(INT_MAX); }
+}
diff --git a/cpp/src/qpid/cluster/types.h b/cpp/src/qpid/cluster/types.h
new file mode 100644
index 0000000000..c25370b6b6
--- /dev/null
+++ b/cpp/src/qpid/cluster/types.h
@@ -0,0 +1,83 @@
+#ifndef QPID_CLUSTER_TYPES_H
+#define QPID_CLUSTER_TYPES_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "config.h"
+#include "qpid/Url.h"
+#include "qpid/sys/IntegerTypes.h"
+#include <boost/intrusive_ptr.hpp>
+#include <utility>
+#include <iosfwd>
+#include <string>
+
+extern "C" {
+#if defined (HAVE_OPENAIS_CPG_H)
+#  include <openais/cpg.h>
+#elif defined (HAVE_COROSYNC_CPG_H)
+#  include <corosync/cpg.h>
+#else
+#  error "No cpg.h header file available"
+#endif    
+}
+
+namespace qpid {
+namespace cluster {
+
+class Connection;
+typedef boost::intrusive_ptr<Connection> ConnectionPtr;
+
+/** Types of cluster event. */
+enum EventType { DATA, CONTROL };
+
+/** first=node-id, second=pid */
+struct MemberId : std::pair<uint32_t, uint32_t> {
+    MemberId(uint64_t n=0) : std::pair<uint32_t,uint32_t>( n >> 32, n & 0xffffffff) {}
+    MemberId(uint32_t node, uint32_t pid) : std::pair<uint32_t,uint32_t>(node, pid) {}
+    MemberId(const cpg_address& caddr) : std::pair<uint32_t,uint32_t>(caddr.nodeid, caddr.pid) {}
+    MemberId(const std::string&); // Decode from string.
+    uint32_t getNode() const { return first; }
+    uint32_t getPid() const { return second; }
+    operator uint64_t() const { return (uint64_t(first)<<32ull) + second; }
+
+    // AsMethodBody as string, network byte order.
+    std::string str() const;
+};
+
+inline bool operator==(const cpg_address& caddr, const MemberId& id) { return id == MemberId(caddr); }
+
+std::ostream& operator<<(std::ostream&, const MemberId&);
+
+struct ConnectionId : public std::pair<MemberId, uint64_t>  {
+    ConnectionId(const MemberId& m=MemberId(), uint64_t c=0) :  std::pair<MemberId, uint64_t> (m,c) {}
+    ConnectionId(uint64_t m, uint64_t c) : std::pair<MemberId, uint64_t>(MemberId(m), c) {}
+    MemberId getMember() const { return first; }
+    uint64_t getNumber() const { return second; }
+};
+
+std::ostream& operator<<(std::ostream&, const ConnectionId&);
+
+std::ostream& operator<<(std::ostream&, EventType);
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_TYPES_H*/