Author: aconway
Date: Mon Jul 23 21:29:17 2012
New Revision: 1364806
URL: http://svn.apache.org/viewvc?rev=1364806&view=rev
Log:
NO-JIRA: Fix typos, update comments, update log messages.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h
qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/HaPlugin.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.h
qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp
qpid/trunk/qpid/cpp/src/tests/ha_tests.py
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h Mon Jul 23 21:29:17 2012
@@ -54,7 +54,9 @@ class Consumer
bool preAcquires() const { return acquires; }
const std::string& getName() const { return name; }
+ /**@return the position of the last message seen by this consumer */
virtual framing::SequenceNumber getPosition() const { return position; }
+
virtual void setPosition(framing::SequenceNumber pos) { position = pos; }
virtual bool deliver(QueuedMessage& msg) = 0;
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp Mon Jul 23 21:29:17 2012
@@ -71,6 +71,7 @@ HaBroker::HaBroker(broker::Broker& b, co
// otherwise there's a window for a client to connect before we get to
// initialize()
if (settings.cluster) {
+ QPID_LOG(debug, logPrefix << "Rejecting client connections.");
observer->setObserver(boost::shared_ptr<broker::ConnectionObserver>(
new BackupConnectionExcluder));
broker.getConnectionObservers().add(observer);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaPlugin.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaPlugin.cpp?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaPlugin.cpp Mon Jul 23 21:29:17 2012
@@ -66,7 +66,7 @@ struct HaPlugin : public Plugin {
broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
if (broker) {
// Must create the HaBroker in earlyInitialize so it can set up its
- // connection observer before clients start conneting.
+ // connection observer before clients start connecting.
haBroker.reset(new ha::HaBroker(*broker, settings));
broker->addFinalizer(boost::bind(&HaPlugin::finalize, this));
}
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Primary.cpp Mon Jul 23 21:29:17 2012
@@ -130,9 +130,13 @@ void Primary::checkReady(BackupMap::iter
if (i != backups.end() && i->second->isReady()) {
BrokerInfo info = i->second->getBrokerInfo();
info.setStatus(READY);
- QPID_LOG(info, "Expected backup is ready: " << info);
haBroker.addBroker(info);
- if (expectedBackups.erase(i->second)) checkReady(l);
+ if (expectedBackups.erase(i->second)) {
+ QPID_LOG(info, logPrefix << "Expected backup is ready: " << info);
+ checkReady(l);
+ }
+ else
+ QPID_LOG(info, logPrefix << "Backup is ready: " << info);
}
}
@@ -146,7 +150,7 @@ void Primary::timeoutExpectedBackups() {
boost::shared_ptr<RemoteBackup> rb = *i;
if (!rb->isConnected()) {
BrokerInfo info = rb->getBrokerInfo();
- QPID_LOG(error, "Expected backup timed out: " << info);
+ QPID_LOG(error, logPrefix << "Expected backup timed out: " <<
info);
expectedBackups.erase(i++);
backups.erase(info.getSystemId());
rb->cancel();
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.cpp?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.cpp Mon Jul 23 21:29:17 2012
@@ -101,7 +101,7 @@ void QueueGuard::cancel() {
void QueueGuard::attach(ReplicatingSubscription& rs) {
Mutex::ScopedLock l(lock);
- subscription = &rs;
+ subscription = &rs;
}
namespace {
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.h?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/QueueGuard.h Mon Jul 23 21:29:17 2012
@@ -79,13 +79,20 @@ class QueueGuard {
void attach(ReplicatingSubscription&);
/**
- * Return the queue range at the time the QueueGuard was created. The
- * QueueGuard is created before the queue becomes active: either when a
- * backup is promoted, or when a new queue is created on the primary.
+ * Return the un-guarded queue range at the time the QueueGuard was
created.
+ *
+ * The first position guaranteed to be protected by the guard is
+ * getRange().getBack()+1. It is possible that the guard has protected some
+ * messages before that point. Any such messages are dealt with in
subscriptionStart
+ *
+ * The QueueGuard is created in 3 situations
+ * - when a backup is promoted, guards are created for expected backups.
+ * - when a new queue is created on the primary
+ * - when a new backup joins.
+ *
+ * In the last situation the queue is active while the guard is being
+ * created.
*
- * NOTE: The first position guaranteed to be protected by the guard is
- * getRange().getBack()+1. It is possible that the guard has protected
- * some messages before that point.
*/
const QueueRange& getRange() const { return range; } // range is
immutable, no lock needed.
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp Mon Jul 23
21:29:17 2012
@@ -175,13 +175,13 @@ ReplicatingSubscription::ReplicatingSubs
logPrefix = os.str();
// NOTE: Once the guard is attached we can have concurrent
- // calls to dequeued so we need to lock use of this->deques.
+ // calls to dequeued so we need to lock use of this->dequeues.
//
// However we must attach the guard _before_ we scan for
// initial dequeues to be sure we don't miss any dequeues
// between the scan and attaching the guard.
if (Primary::get()) guard = Primary::get()->getGuard(queue, info);
- if (!guard) guard.reset(new QueueGuard(*queue, getBrokerInfo()));
+ if (!guard) guard.reset(new QueueGuard(*queue, info));
guard->attach(*this);
QueueRange backup(arguments); // The remote backup state.
@@ -213,6 +213,9 @@ ReplicatingSubscription::ReplicatingSubs
scan.finish();
position = backup.back;
}
+ // NOTE: we are assuming that the messages that are on the backup are
+ // consistent with those on the primary. If the backup is a replica
+ // queue and hasn't been tampered with then that will be the case.
QPID_LOG(debug, logPrefix << "Subscribed:"
<< " backup:" << backup
@@ -332,7 +335,7 @@ void ReplicatingSubscription::sendDequeu
void ReplicatingSubscription::dequeued(const QueuedMessage& qm)
{
assert (qm.queue == getQueue().get());
- QPID_LOG(trace, logPrefix << "Dequeued " << qm);
+ QPID_LOG(trace, logPrefix << "Dequeued " << qm);
{
Mutex::ScopedLock l(lock);
dequeues.add(qm.position);
Modified: qpid/trunk/qpid/cpp/src/tests/ha_tests.py
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_tests.py?rev=1364806&r1=1364805&r2=1364806&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_tests.py Mon Jul 23 21:29:17 2012
@@ -758,7 +758,7 @@ acl deny all all
s1.sender("ex").send("foo");
self.assertEqual(s1.receiver("q").fetch().content, "foo")
- def test_alterante_exchange(self):
+ def test_alternate_exchange(self):
"""Verify that alternate-exchange on exchanges and queues is propagated
to new members of a cluster. """
cluster = HaCluster(self, 2)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
