diff options
| author | Alan Conway <aconway@apache.org> | 2012-05-08 15:04:51 +0000 |
|---|---|---|
| committer | Alan Conway <aconway@apache.org> | 2012-05-08 15:04:51 +0000 |
| commit | 6f68a7a7c39eb412d07d459574f1d1ab52fed7db (patch) | |
| tree | 81480793b64678e54bf091e35bb4e16f3da93cb4 /cpp/src/qpid/ha | |
| parent | e20364fc61452aef02f6f30639b7e6c5638ec7a4 (diff) | |
| download | qpid-python-6f68a7a7c39eb412d07d459574f1d1ab52fed7db.tar.gz | |
QPID-3603: Add ErrorListener to Bridge to handle session errors.
Will be used in HA code.
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1335563 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src/qpid/ha')
| -rw-r--r-- | cpp/src/qpid/ha/Primary.cpp | 86 |
1 files changed, 86 insertions, 0 deletions
diff --git a/cpp/src/qpid/ha/Primary.cpp b/cpp/src/qpid/ha/Primary.cpp new file mode 100644 index 0000000000..bf17d27ca3 --- /dev/null +++ b/cpp/src/qpid/ha/Primary.cpp @@ -0,0 +1,86 @@ +/* + * + * 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 "Backup.h" +#include "ConnectionExcluder.h" +#include "HaBroker.h" +#include "Primary.h" +#include "qpid/broker/Broker.h" +#include "qpid/broker/Queue.h" +#include "qpid/framing/FieldTable.h" +#include <boost/bind.hpp> + +namespace qpid { +namespace ha { + +Primary* Primary::instance = 0; + +Primary::Primary(HaBroker& b) : + haBroker(b), logPrefix(b), + expected(b.getSettings().expectedBackups), + unready(0), activated(false) +{ + instance = this; // Let queue replicators find us. + if (expected == 0) // No need for ready check + activate(*(sys::Mutex::ScopedLock*)0); // fake lock, ok in ctor. + else { + // Set up the primary-ready check: ready when all queues have + // expected number of backups. Note clients are excluded at this point + // so dont't have to worry about changes to the set of queues. + HaBroker::QueueNames names = haBroker.getActiveBackups(); + for (HaBroker::QueueNames::const_iterator i = names.begin(); i != names.end(); ++i) + { + queues[*i] = 0; + ++unready; + QPID_LOG(debug, logPrefix << "Need backup of " << *i + << ", " << unready << " unready queues"); + } + QPID_LOG(critical, "FIXME Primary " << queues.size() << " queues"); + if (queues.empty()) + activate(*(sys::Mutex::ScopedLock*)0); // fake lock, ok in ctor. + else { + QPID_LOG(debug, logPrefix << "Waiting for " << expected + << " backups on " << queues.size() << " queues"); + // Allow backups to connect. + haBroker.getExcluder()->setBackupAllowed(true); + } + } +} + +void Primary::addReplica(const std::string& q) { + sys::Mutex::ScopedLock l(lock); + if (!activated) { + QueueCounts::iterator i = queues.find(q); + if (i != queues.end()) { + ++i->second; + if (i->second == expected) --unready; + QPID_LOG(debug, logPrefix << i->second << " backups caught up on " << q + << ", " << unready << " unready queues"); + if (unready == 0) activate(l); + } + } +} + +void Primary::activate(sys::Mutex::ScopedLock&) { + activated = true; + haBroker.activate(); +} + +}} // namespace qpid::ha |
