diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/qpid/cluster/Cluster.cpp | 11 | ||||
| -rw-r--r-- | cpp/src/qpid/cluster/Connection.cpp | 8 | ||||
| -rw-r--r-- | cpp/src/qpid/cluster/Connection.h | 3 | ||||
| -rw-r--r-- | cpp/src/tests/cluster_test.cpp | 3 | ||||
| -rwxr-xr-x | cpp/src/tests/run_test | 2 | ||||
| -rw-r--r-- | cpp/src/tests/vg_check | 5 |
6 files changed, 28 insertions, 4 deletions
diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp index 70c73191ee..f6022aa5b8 100644 --- a/cpp/src/qpid/cluster/Cluster.cpp +++ b/cpp/src/qpid/cluster/Cluster.cpp @@ -565,6 +565,17 @@ void Cluster::memberUpdate(Lock& l) { } mgmtObject->set_members(urlstr); } + + //close connections belonging to members that have now been excluded + for (ConnectionMap::iterator i = connections.begin(); i != connections.end();) { + MemberId member = i->first.getMember(); + if (member != myId && !map.isMember(member)) { + i->second->left(); + connections.erase(i++); + } else { + i++; + } + } } std::ostream& operator<<(std::ostream& o, const Cluster& cluster) { diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp index 513816735d..bcdc4ffe27 100644 --- a/cpp/src/qpid/cluster/Connection.cpp +++ b/cpp/src/qpid/cluster/Connection.cpp @@ -141,6 +141,7 @@ void Connection::delivered(framing::AMQFrame& f) { } } +// A local connection is closed by the network layer. void Connection::closed() { try { if (catchUp) { @@ -165,12 +166,19 @@ void Connection::closed() { } } +// Self-delivery of close message, close the connection. void Connection::deliverClose () { assert(!catchUp); connection.closed(); cluster.erase(self); } +// Member of a shadow connection left the cluster. +void Connection::left() { + assert(isShadow()); + connection.closed(); +} + // Decode data from local clients. size_t Connection::decode(const char* buffer, size_t size) { if (catchUp) { // Handle catch-up locally. diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h index 2eafa90f32..06176bf81d 100644 --- a/cpp/src/qpid/cluster/Connection.h +++ b/cpp/src/qpid/cluster/Connection.h @@ -89,6 +89,9 @@ class Connection : void idleOut() { connection.idleOut(); } void idleIn() { connection.idleIn(); } + /** Called if the connectors member has left the cluster */ + void left(); + // ConnectionCodec methods size_t decode(const char* buffer, size_t size); diff --git a/cpp/src/tests/cluster_test.cpp b/cpp/src/tests/cluster_test.cpp index cf6c2e73de..5fc513bb28 100644 --- a/cpp/src/tests/cluster_test.cpp +++ b/cpp/src/tests/cluster_test.cpp @@ -117,12 +117,13 @@ ClusterFixture::ClusterFixture(size_t n, int localIndex_) : name(Uuid(true).str( void ClusterFixture::add() { if (size() != size_t(localIndex)) { // fork a broker process. std::ostringstream os; os << "fork" << size(); + std::string prefix = os.str(); const char* argv[] = { "qpidd " __FILE__ , "--load-module=../.libs/cluster.so", "--cluster-name", name.c_str(), "--auth=no", "--no-data-dir", - "--log-prefix", os.str().c_str(), + "--log-prefix", prefix.c_str(), }; size_t argc = sizeof(argv)/sizeof(argv[0]); forkedBrokers.push_back(shared_ptr<ForkedBroker>(new ForkedBroker(argc, argv))); diff --git a/cpp/src/tests/run_test b/cpp/src/tests/run_test index 4d0da15d4c..c4789bb391 100755 --- a/cpp/src/tests/run_test +++ b/cpp/src/tests/run_test @@ -49,7 +49,7 @@ if grep -l "^# Generated by .*libtool" "$1" >/dev/null 2>&1; then test -n "$VALGRIND" && VALGRIND="$VALGRIND $VALGRIND_OPTS" # Hide output unless there's an error. libtool --mode=execute $VALGRIND "$@" 2>&1 || ERROR=1 - test -n "$VALGRIND" && { vg_check $VG_LOG* || ERROR=1 ; } + test -n "$VALGRIND" && { vg_check $VG_LOG || ERROR=1 ; } else # This is a non-libtool shell script, just execute it. exec "$@" diff --git a/cpp/src/tests/vg_check b/cpp/src/tests/vg_check index e9a691abe6..c5a1e6d2d0 100644 --- a/cpp/src/tests/vg_check +++ b/cpp/src/tests/vg_check @@ -1,6 +1,7 @@ # Check for valgrind errors. Sourced by test scripts. vg_failed() { + echo "Valgrind error log in $VG_LOG." 1>&2 cat $VG_LOG 1>&2 echo $1 1>&2 exit 1 @@ -12,10 +13,10 @@ vg_check() test -f $VG_LOG || vg_failed Valgrind log file $VG_LOG missing. # Ensure there is an ERROR SUMMARY line. grep -E '^==[0-9]+== ERROR SUMMARY:' $VG_LOG > /dev/null || \ - vg_failed "No valgrind ERROR SUMMARY line in $$vg_failed." + vg_failed "No valgrind ERROR SUMMARY line in $VG_LOG." # Ensure that the number of errors is 0. grep -E '^==[0-9]+== ERROR SUMMARY: [^0]' $VG_LOG > /dev/null && \ - vg_failed "Valgrind reported errors in $vg_out; see above." + vg_failed "Valgrind reported errors in $VG_LOG; see above." # Check for leaks. grep -E '^==[0-9]+== +.* lost: [^0]' $VG_LOG && \ vg_failed "Found memory leaks (see log file, $VG_LOG); see above." |