diff -rup _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Broker.h _qpid-cpp-0.10-new/cpp/src/qpid/broker/Broker.h --- _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Broker.h 2012-08-14 09:21:34.031944153 +0200 +++ _qpid-cpp-0.10-new/cpp/src/qpid/broker/Broker.h 2012-08-14 09:22:23.132995577 +0200 @@ -230,6 +230,7 @@ public: SessionManager& getSessionManager() { return sessionManager; } const std::string& getFederationTag() const { return federationTag; } + void setFederationTag(const std::string& tag) { federationTag = tag; } management::ManagementObject* GetManagementObject (void) const; management::Manageable* GetVhostObject (void) const; diff -rup _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Connection.cpp _qpid-cpp-0.10-new/cpp/src/qpid/broker/Connection.cpp --- _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Connection.cpp 2012-08-14 09:21:34.034911239 +0200 +++ _qpid-cpp-0.10-new/cpp/src/qpid/broker/Connection.cpp 2012-08-14 09:22:50.201913557 +0200 @@ -103,8 +103,7 @@ Connection::Connection(ConnectionOutputH outboundTracker(*this) { outboundTracker.wrap(out); - if (isLink) - links.notifyConnection(mgmtId, this); + links.notifyConnection(mgmtId, this); // In a cluster, allow adding the management object to be delayed. if (!delayManagement) addManagementObject(); if (!isShadow()) broker.getConnectionCounter().inc_connectionCount(); diff -rup _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Link.cpp _qpid-cpp-0.10-new/cpp/src/qpid/broker/Link.cpp --- _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Link.cpp 2012-08-14 09:21:34.032974118 +0200 +++ _qpid-cpp-0.10-new/cpp/src/qpid/broker/Link.cpp 2012-08-14 09:45:41.635930246 +0200 @@ -132,7 +132,8 @@ Link::Link(LinkRegistry* _links, channelCounter(1), connection(0), agent(0), - failoverChannel(0) + failoverChannel(0), + passive(links->isPassive()) { if (parent != 0 && broker != 0) { @@ -183,7 +184,6 @@ void Link::setStateLH (int newState) case STATE_OPERATIONAL : mgmtObject->set_state("Operational"); break; case STATE_FAILED : mgmtObject->set_state("Failed"); break; case STATE_CLOSED : mgmtObject->set_state("Closed"); break; - case STATE_PASSIVE : mgmtObject->set_state("Passive"); break; } } @@ -193,9 +193,14 @@ void Link::startConnectionLH () // Set the state before calling connect. It is possible that connect // will fail synchronously and call Link::closed before returning. setStateLH(STATE_CONNECTING); - broker->connect (host, port, transport, - boost::bind (&Link::closed, this, _1, _2)); - QPID_LOG (debug, "Inter-broker link connecting to " << host << ":" << port); + if (!passive) { + // Passive links don't create their own connection. + // In a cluster the elder creates the connection and the non-elders + // pick up the shadow of this connection in established(). + broker->connect (host, port, transport, + boost::bind (&Link::closed, this, _1, _2)); + QPID_LOG (debug, "Inter-broker link connecting to " << host << ":" << port); + } } catch(std::exception& e) { setStateLH(STATE_WAITING); if (!hideManagement()) @@ -592,18 +597,12 @@ Manageable::status_t Link::ManagementMet return Manageable::STATUS_UNKNOWN_METHOD; } -void Link::setPassive(bool passive) +void Link::setPassive(bool p) { Mutex::ScopedLock mutex(lock); - if (passive) { - setStateLH(STATE_PASSIVE); - } else { - if (state == STATE_PASSIVE) { - setStateLH(STATE_WAITING); - } else { - QPID_LOG(warning, "Ignoring attempt to activate non-passive link"); - } - } + if (passive && !p) // transition passive->active + setStateLH(STATE_WAITING); + passive = p; } /** utility to clean up connection resources correctly */ Only in _qpid-cpp-0.10-new/cpp/src/qpid/broker: .Link.cpp.swp diff -rup _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Link.h _qpid-cpp-0.10-new/cpp/src/qpid/broker/Link.h --- _qpid-cpp-0.10-orig/cpp/src/qpid/broker/Link.h 2012-08-14 09:21:34.023029499 +0200 +++ _qpid-cpp-0.10-new/cpp/src/qpid/broker/Link.h 2012-08-14 09:25:35.546009198 +0200 @@ -89,10 +89,11 @@ namespace qpid { static const int STATE_OPERATIONAL = 3; static const int STATE_FAILED = 4; static const int STATE_CLOSED = 5; - static const int STATE_PASSIVE = 6; static const uint32_t MAX_INTERVAL = 32; + bool passive; + void setStateLH (int newState); void startConnectionLH(); // Start the IO Connection void destroy(); // Called when mgmt deletes this link diff -rup _qpid-cpp-0.10-orig/cpp/src/qpid/broker/LinkRegistry.h _qpid-cpp-0.10-new/cpp/src/qpid/broker/LinkRegistry.h --- _qpid-cpp-0.10-orig/cpp/src/qpid/broker/LinkRegistry.h 2012-08-14 09:21:34.020943150 +0200 +++ _qpid-cpp-0.10-new/cpp/src/qpid/broker/LinkRegistry.h 2012-08-14 09:25:59.046009904 +0200 @@ -149,6 +149,7 @@ namespace broker { * bridges won't therefore pull or push any messages. */ void setPassive(bool); + bool isPassive() const { return passive; } /** Iterate over each link in the registry. Used for cluster updates. */ diff -rup _qpid-cpp-0.10-orig/cpp/src/qpid/cluster/Connection.cpp _qpid-cpp-0.10-new/cpp/src/qpid/cluster/Connection.cpp --- _qpid-cpp-0.10-orig/cpp/src/qpid/cluster/Connection.cpp 2012-08-14 09:21:34.053038095 +0200 +++ _qpid-cpp-0.10-new/cpp/src/qpid/cluster/Connection.cpp 2012-08-14 09:26:36.494036465 +0200 @@ -697,6 +697,7 @@ void Connection::managementSetupState( agent->setBootSequence(bootSequence); agent->setUuid(id); agent->setName(vendor, product, instance); + cluster.getBroker().setFederationTag(id.str()); } void Connection::config(const std::string& encoded) { diff -rup _qpid-cpp-0.10-orig/cpp/src/tests/cluster_test_logs.py _qpid-cpp-0.10-new/cpp/src/tests/cluster_test_logs.py --- _qpid-cpp-0.10-orig/cpp/src/tests/cluster_test_logs.py 2012-08-14 09:21:33.941926929 +0200 +++ _qpid-cpp-0.10-new/cpp/src/tests/cluster_test_logs.py 2012-08-14 09:28:14.712911573 +0200 @@ -65,6 +65,7 @@ def filter_log(log): 'debug Sending keepalive signal to watchdog', # Watchdog timer thread 'last broker standing joined by 1 replicas, updating queue policies.', 'Connection .* timed out: closing', # heartbeat connection close + "info Connection is a federation link", # out of order with connection mgmt object. "org.apache.qpid.broker:bridge:" # ignore bridge index ]) # Regex to match a UUID @@ -97,14 +98,15 @@ def filter_log(log): out.write(l) out.close() -def verify_logs(): +def verify_logs(pattern="*.log"): """Compare log files from cluster brokers, verify that they correspond correctly.""" - for l in glob.glob("*.log"): filter_log(l) + for l in glob.glob(pattern): filter_log(l) checkpoints = set() - for l in glob.glob("*.filter"): checkpoints = checkpoints.union(set(split_log(l))) + for l in glob.glob("%s.filter"%(pattern)): + checkpoints = checkpoints.union(set(split_log(l))) errors=[] for c in checkpoints: - fragments = glob.glob("*.filter.%s"%(c)) + fragments = glob.glob("%s.filter.%s"%(pattern, c)) fragments.sort(reverse=True, key=os.path.getsize) while len(fragments) >= 2: a = fragments.pop(0) diff -rup _qpid-cpp-0.10-orig/cpp/src/tests/cluster_tests.py _qpid-cpp-0.10-new/cpp/src/tests/cluster_tests.py --- _qpid-cpp-0.10-orig/cpp/src/tests/cluster_tests.py 2012-08-14 09:21:33.941030557 +0200 +++ _qpid-cpp-0.10-new/cpp/src/tests/cluster_tests.py 2012-08-14 09:29:24.710989922 +0200 @@ -888,6 +888,83 @@ acl allow all all for i in range(2, len(src_cluster)): src_cluster[i].kill() for i in range(2, len(dst_cluster)): dst_cluster[i].kill() + def _verify_federation(self, src_broker, src, dst_broker, dst, timeout=30): + """ Prove that traffic can pass between two federated brokers. + """ + tot_time = 0 + active = False + send_session = src_broker.connect().session() + sender = send_session.sender(src) + receive_session = dst_broker.connect().session() + receiver = receive_session.receiver(dst) + while not active and tot_time < timeout: + sender.send(Message("Hello from Source!")) + try: + receiver.fetch(timeout = 1) + receive_session.acknowledge() + # Get this far without Empty exception, and the link is good! + active = True + while True: + # Keep receiving msgs, as several may have accumulated + receiver.fetch(timeout = 1) + receive_session.acknowledge() + except Empty: + if not active: + tot_time += 1 + receiver.close() + receive_session.close() + sender.close() + send_session.close() + return active + + def test_federation_bridges_consistent(self): + """FIXME JIRA: verify that federation bridges are constructed consistently.""" + # 2 node cluster source, 2 node cluster destination + args=["--mgmt-pub-interval=1", + "--log-enable=trace+:management", + "--log-enable=trace+:Bridge" + ] + src_cluster = self.cluster(2, args=args) + for b in src_cluster: b.ready(); + dst_cluster = self.cluster(2, args=args) + for b in dst_cluster: b.ready(); + + cmd = self.popen(["qpid-config", + "--broker", src_cluster[0].host_port(), + "add", "queue", "srcQ"], EXPECT_EXIT_OK) + cmd.wait() + + cmd = self.popen(["qpid-config", + "--broker", dst_cluster[0].host_port(), + "add", "exchange", "fanout", "destX"], EXPECT_EXIT_OK) + cmd.wait() + + cmd = self.popen(["qpid-config", + "--broker", dst_cluster[0].host_port(), + "add", "queue", "destQ"], EXPECT_EXIT_OK) + cmd.wait() + + cmd = self.popen(["qpid-config", + "--broker", dst_cluster[0].host_port(), + "bind", "destX", "destQ"], EXPECT_EXIT_OK) + cmd.wait() + + # federate the srcQ to the destination exchange + dst_cluster[0].startQmf() + dst_broker = dst_cluster[0].qmf_session.getObjects(_class="broker")[0] + result = dst_broker.connect(src_cluster[0].host(), src_cluster[0].port(), False, "PLAIN", + "guest", "guest", "tcp") + self.assertEqual(result.status, 0, result); + + link = dst_cluster[0].qmf_session.getObjects(_class="link")[0] + result = link.bridge(False, "srcQ", "destX", "", "", "", True, False, False, 10) + self.assertEqual(result.status, 0, result) + + # check that traffic passes + assert self._verify_federation(src_cluster[0], "srcQ", dst_cluster[0], "destQ") + + cluster_test_logs.verify_logs("*%s*.log"%(dst_cluster.name)) + cluster_test_logs.verify_logs("*%s*.log"%(src_cluster.name)) class LongTests(BrokerTest):