You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gm...@apache.org on 2017/09/19 17:39:08 UTC
qpid-dispatch git commit: DISPATCH-818 - Added connection info list
to the connector which stores the initial connection information and the
backup connection information 1. If the primary connection goes down,
the primary and the backup connections will
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 233f23f15 -> af96b23fe
DISPATCH-818 - Added connection info list to the connector which stores the initial connection information and the backup connection information
1. If the primary connection goes down, the primary and the backup connections will be tried in quick succession
2. If the new backup connection does not provide failover-server-list, the list on the server is cleaned up to contain only the one pertinent connection info.
3. The connection info list always has at least one element in it which represents the initial successful connection information
(cherry picked from commit 2fc9e285f92d96f7d6c5a4e9e36bf356ab7cc256)
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/af96b23f
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/af96b23f
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/af96b23f
Branch: refs/heads/master
Commit: af96b23fefe0076391db35d31d612a1ab5e6f4a2
Parents: 233f23f
Author: Ganesh Murthy <gm...@redhat.com>
Authored: Fri Sep 15 09:57:16 2017 -0400
Committer: Ganesh Murthy <gm...@redhat.com>
Committed: Tue Sep 19 13:33:02 2017 -0400
----------------------------------------------------------------------
include/qpid/dispatch/failoverlist.h | 12 ++
python/qpid_dispatch/management/qdrouter.json | 8 +-
src/connection_manager.c | 88 ++++++++++++-
src/failoverlist.c | 10 --
src/router_node.c | 143 +++++++++++++++++++++
src/server.c | 67 +++++++++-
src/server_private.h | 4 +
tests/CMakeLists.txt | 1 +
tests/system_tests_handle_failover.py | 140 ++++++++++++++++++++
9 files changed, 457 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/include/qpid/dispatch/failoverlist.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/failoverlist.h b/include/qpid/dispatch/failoverlist.h
index c9e6fb5..d321445 100644
--- a/include/qpid/dispatch/failoverlist.h
+++ b/include/qpid/dispatch/failoverlist.h
@@ -1,5 +1,6 @@
#ifndef __failoverlist_h__
#define __failoverlist_h__ 1
+#include <qpid/dispatch/ctools.h>
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -24,6 +25,17 @@
*/
typedef struct qd_failover_list_t qd_failover_list_t;
+typedef struct qd_failover_item_t {
+ DEQ_LINKS(struct qd_failover_item_t);
+ char *scheme;
+ char *host;
+ char *port;
+ char *hostname;
+ char *host_port;
+} qd_failover_item_t;
+
+DEQ_DECLARE(qd_failover_item_t, qd_failover_item_list_t);
+
/**
* qd_failover_list
*
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/python/qpid_dispatch/management/qdrouter.json
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch/management/qdrouter.json b/python/qpid_dispatch/management/qdrouter.json
index ccedc94..70000bb 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -895,7 +895,13 @@
"default": "none",
"description": "A comma separated list that indicates which components of the message should be logged (no spaces allowed between list components). Defaults to 'none' (log nothing). If you want all properties and application properties of the message logged use 'all'. Specific components of the message can be logged by indicating the components via a comma separated list. The components are message-id, user-id, to, subject, reply-to, correlation-id, content-type, content-encoding, absolute-expiry-time, creation-time, group-id, group-sequence, reply-to-group-id, app-properties. The application-data part of the bare message will not be logged. This log message is written to the MESSAGE logging module. In the 'log' entity, set 'module' property to MESSAGE or DEFAULT and 'enable' to trace+ to see this log message",
"create": true
- }
+ },
+ "failoverList": {
+ "type": "string",
+ "description": "A read-only, comma-separated list of failover urls. ",
+ "create": false
+
+ }
}
},
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/src/connection_manager.c
----------------------------------------------------------------------
diff --git a/src/connection_manager.c b/src/connection_manager.c
index 797326c..1e08b7d 100644
--- a/src/connection_manager.c
+++ b/src/connection_manager.c
@@ -631,9 +631,77 @@ qd_error_t qd_entity_refresh_listener(qd_entity_t* entity, void *impl)
}
+static int get_failover_info_length(qd_failover_item_list_t conn_info_list)
+{
+ int arr_length = 0;
+ qd_failover_item_t *item = DEQ_HEAD(conn_info_list);
+
+ item = DEQ_NEXT(item);
+ while(item) {
+ if (item->scheme) {
+ // The +3 is for the '://'
+ arr_length += strlen(item->scheme) + 3;
+ }
+ if (item->host_port) {
+ arr_length += strlen(item->host_port);
+ }
+ item = DEQ_NEXT(item);
+ if (item) {
+ // This is for the comma between the items
+ arr_length += 2;
+ }
+ }
+
+ if (arr_length > 0)
+ // This is for the final '\0'
+ arr_length += 1;
+
+ return arr_length;
+}
+
qd_error_t qd_entity_refresh_connector(qd_entity_t* entity, void *impl)
{
- return QD_ERROR_NONE;
+ qd_connector_t *ct = (qd_connector_t*) impl;
+
+ if (DEQ_SIZE(ct->conn_info_list) > 1) {
+ qd_failover_item_list_t conn_info_list = ct->conn_info_list;
+
+ qd_failover_item_t *item = DEQ_HEAD(conn_info_list);
+
+ //
+ // As you can see we are skipping the head of the list. The
+ // first item in the list is always the original connection information
+ // and we dont want to display that information as part of the failover list.
+ //
+ int arr_length = get_failover_info_length(conn_info_list);
+ char failover_info[arr_length];
+ memset(failover_info, 0, sizeof(failover_info));
+
+ item = DEQ_NEXT(item);
+
+ while(item) {
+ if (item->scheme) {
+ strcat(failover_info, item->scheme);
+ strcat(failover_info, "://");
+ }
+ if (item->host_port) {
+ strcat(failover_info, item->host_port);
+ }
+ item = DEQ_NEXT(item);
+ if (item) {
+ strcat(failover_info, ", ");
+ }
+ }
+
+ if (qd_entity_set_string(entity, "failoverList", failover_info) == 0)
+ return QD_ERROR_NONE;
+ }
+ else {
+ if (qd_entity_clear(entity, "failoverList") == 0)
+ return QD_ERROR_NONE;
+ }
+
+ return qd_error_code();
}
@@ -645,6 +713,24 @@ qd_connector_t *qd_dispatch_configure_connector(qd_dispatch_t *qd, qd_entity_t *
DEQ_ITEM_INIT(ct);
DEQ_INSERT_TAIL(cm->connectors, ct);
log_config(cm->log_source, &ct->config, "Connector");
+
+ //
+ // Add the first item to the ct->conn_info_list
+ // The initial connection information and any backup connection information is stored in the conn_info_list
+ //
+ qd_failover_item_t *item = NEW(qd_failover_item_t);
+ ZERO(item);
+ item->scheme = 0;
+ item->host = strdup(ct->config.host);
+ item->port = strdup(ct->config.port);
+ item->hostname = 0;
+
+ int hplen = strlen(item->host) + strlen(item->port) + 2;
+ item->host_port = malloc(hplen);
+ snprintf(item->host_port, hplen, "%s:%s", item->host , item->port);
+
+ DEQ_INSERT_TAIL(ct->conn_info_list, item);
+
return ct;
}
qd_log(cm->log_source, QD_LOG_ERROR, "Unable to create connector: %s", qd_error_message());
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/src/failoverlist.c
----------------------------------------------------------------------
diff --git a/src/failoverlist.c b/src/failoverlist.c
index f350c1a..e21ea6e 100644
--- a/src/failoverlist.c
+++ b/src/failoverlist.c
@@ -23,16 +23,6 @@
#include <ctype.h>
#include <string.h>
-typedef struct qd_failover_item_t {
- DEQ_LINKS(struct qd_failover_item_t);
- const char *scheme;
- const char *host;
- const char *port;
- const char *hostname;
-} qd_failover_item_t;
-
-DEQ_DECLARE(qd_failover_item_t, qd_failover_item_list_t);
-
struct qd_failover_list_t {
qd_failover_item_list_t item_list;
char *text;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/src/router_node.c
----------------------------------------------------------------------
diff --git a/src/router_node.c b/src/router_node.c
index a7f50fa..43346da 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -767,6 +767,149 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
cost = remote_cost;
}
+ bool found_failover = false;
+
+ if (conn->connector && DEQ_SIZE(conn->connector->conn_info_list) > 1) {
+ // Here we are simply removing all other failover information except the one we used to make a successful connection.
+ int i = 1;
+ qd_failover_item_t *item = DEQ_HEAD(conn->connector->conn_info_list);
+ qd_failover_item_t *next_item = 0;
+ while(item) {
+ if (i != conn->connector->conn_index) {
+ next_item = DEQ_NEXT(item);
+ free(item->scheme);
+ free(item->host);
+ free(item->port);
+ free(item->hostname);
+ free(item->host_port);
+ DEQ_REMOVE(conn->connector->conn_info_list, item);
+
+ free(item);
+ item = next_item;
+ }
+ else {
+ item = DEQ_NEXT(item);
+ }
+ i += 1;
+ }
+ conn->connector->conn_index = 0;
+
+ // By the end of this loop we should be left with only one element in the conn_info_list.
+ }
+
+ if (props) {
+ pn_data_rewind(props);
+ pn_data_next(props);
+ if (props && pn_data_type(props) == PN_MAP) {
+ pn_data_enter(props);
+
+ //
+ // We are attempting to find a connection property called failover-server-list.
+ // which looks something like this
+ // :"failover-server-list"=[{:"network-host"="some-host", :port="35000"}, {:"network-host"="localhost", :port="25000"}]
+ // Note that the failover-list can contain one or more maps that contain failover connection information.
+ // In the following code, we are trying to get the contents of each map into the qd_failover_item_t object.
+ //
+ while (pn_data_next(props)) {
+ if (pn_data_type(props) == PN_SYMBOL) {
+ pn_bytes_t sym = pn_data_get_symbol(props);
+ if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY) == 0) {
+ found_failover = true;
+ }
+ }
+ else if (pn_data_type(props) == PN_LIST && found_failover) {
+ size_t list_num_items = pn_data_get_list(props);
+
+ if (list_num_items > 0) {
+
+ pn_data_enter(props); // enter list
+
+ for (int i=0; i < list_num_items; i++) {
+ pn_data_next(props);// this is the first element of the list, a map.
+ if (props && pn_data_type(props) == PN_MAP) {
+
+ size_t map_num_items = pn_data_get_map(props);
+ pn_data_enter(props);
+
+ qd_failover_item_t *item = NEW(qd_failover_item_t);
+ ZERO(item);
+
+ // We have found a map with the connection information. Step thru the map contents and create qd_failover_item_t
+
+ for (int j=0; j < map_num_items/2; j++) {
+
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_SYMBOL) {
+ pn_bytes_t sym = pn_data_get_symbol(props);
+ if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY) == 0) {
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_STRING) {
+ item->host = strdup(pn_data_get_string(props).start);
+ }
+ }
+ else if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY) == 0) {
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_STRING) {
+ item->port = strdup(pn_data_get_string(props).start);
+ }
+
+ }
+ else if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_SCHEME_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_SCHEME_KEY) == 0) {
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_STRING) {
+ item->scheme = strdup(pn_data_get_string(props).start);
+ }
+
+ }
+ else if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_HOSTNAME_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_HOSTNAME_KEY) == 0) {
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_STRING) {
+ item->hostname = strdup(pn_data_get_string(props).start);
+ }
+ }
+ }
+ }
+
+ int host_length = strlen(item->host);
+ //
+ // We will not even bother inserting the item if there is no host available.
+ //
+ if (host_length != 0) {
+ if (item->scheme == 0)
+ item->scheme = strdup("amqp");
+ if (item->port == 0)
+ item->port = strdup("5672");
+
+ int hplen = strlen(item->host) + strlen(item->port) + 2;
+ item->host_port = malloc(hplen);
+ snprintf(item->host_port, hplen, "%s:%s", item->host, item->port);
+
+ DEQ_INSERT_TAIL(conn->connector->conn_info_list, item);
+
+ qd_log(router->log_source, QD_LOG_DEBUG, "Added %s as backup host", item->host_port);
+ }
+ else {
+ free(item->scheme);
+ free(item->host);
+ free(item->port);
+ free(item->hostname);
+ free(item->host_port);
+ free(item);
+ }
+ }
+ pn_data_exit(props);
+ }
+ } // list_num_items > 0
+ }
+ }
+ }
+ }
+
if (multi_tenant)
vhost = pn_connection_remote_hostname(pn_conn);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/src/server.c
----------------------------------------------------------------------
diff --git a/src/server.c b/src/server.c
index 64bd8ff..2c2dd36 100644
--- a/src/server.c
+++ b/src/server.c
@@ -713,6 +713,14 @@ static void handle_listener(pn_event_t *e, qd_server_t *qd_server) {
}
+bool qd_connector_has_failover_info(qd_connector_t* ct)
+{
+ if (ct && DEQ_SIZE(ct->conn_info_list) > 1)
+ return true;
+ return false;
+}
+
+
void qd_connection_free(qd_connection_t *ctx)
{
qd_server_t *qd_server = ctx->server;
@@ -723,9 +731,25 @@ void qd_connection_free(qd_connection_t *ctx)
if (ctx->connector) {
sys_mutex_lock(ctx->connector->lock);
ctx->connector->ctx = 0;
+ // Increment the connection index by so that we can try connecting to the failover url (if any).
+ bool has_failover = qd_connector_has_failover_info(ctx->connector);
+ if (has_failover) {
+ if (DEQ_SIZE(ctx->connector->conn_info_list) == ctx->connector->conn_index)
+ // Start round robin again
+ ctx->connector->conn_index = 1;
+ else
+ ctx->connector->conn_index += 1;
+
+ // Go thru the failover list round robin.
+ // IMPORTANT: Note here that we set the re-try timer to 1 second.
+ // We want to quickly keep cycling thru the failover urls every second.
+ qd_timer_schedule(ctx->connector->timer, 1000);
+ }
+
ctx->connector->state = CXTR_STATE_CONNECTING;
sys_mutex_unlock(ctx->connector->lock);
- qd_timer_schedule(ctx->connector->timer, ctx->connector->delay);
+ if (!has_failover)
+ qd_timer_schedule(ctx->connector->timer, ctx->connector->delay);
}
// If counted for policy enforcement, notify it has closed
@@ -856,6 +880,19 @@ static void *thread_run(void *arg)
}
+qd_failover_item_t *qd_connector_get_conn_info(qd_connector_t *ct) {
+
+ qd_failover_item_t *item = DEQ_HEAD(ct->conn_info_list);
+
+ if (DEQ_SIZE(ct->conn_info_list) > 1) {
+ for (int i=1; i < ct->conn_index; i++) {
+ item = DEQ_NEXT(item);
+ }
+ }
+ return item;
+}
+
+
/* Timer callback to try/retry connection open */
static void try_open_lh(qd_connector_t *ct)
{
@@ -880,7 +917,13 @@ static void try_open_lh(qd_connector_t *ct)
// Set the hostname on the pn_connection. This hostname will be used by proton as the
// hostname in the open frame.
//
- pn_connection_set_hostname(ctx->pn_conn, config->host);
+
+ qd_failover_item_t *item = qd_connector_get_conn_info(ct);
+
+ char *current_host = item->host;
+ char *host_port = item->host_port;
+
+ pn_connection_set_hostname(ctx->pn_conn, current_host);
// Set the sasl user name and password on the proton connection object. This has to be
// done before pn_proactor_connect which will bind a transport to the connection
@@ -894,9 +937,9 @@ static void try_open_lh(qd_connector_t *ct)
ct->delay = 5000;
qd_log(ct->server->log_source, QD_LOG_TRACE,
- "[%"PRIu64"] Connecting to %s", ctx->connection_id, config->host_port);
+ "[%"PRIu64"] Connecting to %s", ctx->connection_id, host_port);
/* Note: the transport is configured in the PN_CONNECTION_BOUND event */
- pn_proactor_connect(ct->server->proactor, ctx->pn_conn, config->host_port);
+ pn_proactor_connect(ct->server->proactor, ctx->pn_conn, host_port);
}
static void setup_ssl_sasl_and_open(qd_connection_t *ctx)
@@ -1214,6 +1257,10 @@ qd_connector_t *qd_server_connector(qd_server_t *server)
if (!ct) return 0;
sys_atomic_init(&ct->ref_count, 1);
ct->server = server;
+ qd_failover_item_list_t conn_info_list;
+ DEQ_INIT(conn_info_list);
+ ct->conn_info_list = conn_info_list;
+ ct->conn_index = 0;
ct->lock = sys_mutex();
ct->timer = qd_timer(ct->server->qd, try_open_cb, ct);
if (!ct->lock || !ct->timer) {
@@ -1248,6 +1295,18 @@ void qd_connector_decref(qd_connector_t* ct)
sys_mutex_unlock(ct->lock);
qd_server_config_free(&ct->config);
qd_timer_free(ct->timer);
+
+ qd_failover_item_t *item = DEQ_HEAD(ct->conn_info_list);
+ while (item) {
+ DEQ_REMOVE_HEAD(ct->conn_info_list);
+ free(item->scheme);
+ free(item->host);
+ free(item->port);
+ free(item->hostname);
+ free(item->host_port);
+ free(item);
+ item = DEQ_HEAD(ct->conn_info_list);
+ }
free_qd_connector_t(ct);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/src/server_private.h
----------------------------------------------------------------------
diff --git a/src/server_private.h b/src/server_private.h
index 11b1d31..2e66587 100644
--- a/src/server_private.h
+++ b/src/server_private.h
@@ -119,6 +119,10 @@ struct qd_connector_t {
sys_mutex_t *lock;
cxtr_state_t state;
qd_connection_t *ctx;
+
+ /* This conn_list contains all the connection information needed to make a connection. It also includes failover connection information */
+ qd_failover_item_list_t conn_info_list;
+ int conn_index; // Which connection in the connection list to connect to next.
DEQ_LINKS(qd_connector_t);
};
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index fc9d548..6727d9a 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -81,6 +81,7 @@ foreach(py_test_module
system_tests_drain
system_tests_management
system_tests_one_router
+ system_tests_handle_failover
system_tests_default_distribution
system_tests_policy
system_tests_protocol_family
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/af96b23f/tests/system_tests_handle_failover.py
----------------------------------------------------------------------
diff --git a/tests/system_tests_handle_failover.py b/tests/system_tests_handle_failover.py
new file mode 100644
index 0000000..0204a48
--- /dev/null
+++ b/tests/system_tests_handle_failover.py
@@ -0,0 +1,140 @@
+#
+# 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.
+#
+
+import json, re
+from time import sleep
+import system_test
+from system_test import TestCase, Qdrouterd, Process, TIMEOUT
+from subprocess import PIPE, STDOUT
+
+class FailoverTest(TestCase):
+ inter_router_port = None
+
+ @classmethod
+ def setUpClass(cls):
+ super(FailoverTest, cls).setUpClass()
+
+ def router(name, config):
+ config = Qdrouterd.Config(config)
+
+ cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))
+
+ cls.routers = []
+
+ inter_router_port = cls.tester.get_port()
+ cls.inter_router_port_1 = cls.tester.get_port()
+ cls.backup_port = cls.tester.get_port()
+ cls.failover_list = 'amqp://third-host:5671, ' + 'amqp://localhost:' + str(cls.backup_port)
+
+ #
+ # Router A tries to connect to Router B via its connectorToB. Router B responds with an open frame which will
+ # have the failover-server-list as one of its connection properties like the following -
+ # [0x13024d0]:0 <- @open(16) [container-id="Router.A", max-frame-size=16384, channel-max=32767,
+ # idle-time-out=8000, offered-capabilities=:"ANONYMOUS-RELAY",
+ # properties={:product="qpid-dispatch-router", :version="1.0.0",
+ # :"failover-server-list"=[{:"network-host"="some-host", :port="35000"},
+ # {:"network-host"="localhost", :port="25000"}]}]
+ #
+ # The suite of tests determine if the router receiving this open frame stores it properly and if the
+ # original connection goes down, check that the router is trying to make connections to the failover urls.
+ #
+ router('QDR.B', [
+ ('router', {'mode': 'interior', 'id': 'QDR.B'}),
+ ('listener', {'role': 'inter-router', 'port': inter_router_port,
+ 'failoverList': cls.failover_list}),
+ ('listener', {'role': 'normal', 'port': cls.tester.get_port()}),
+ ]
+ )
+ router('QDR.A',
+ [
+ ('router', {'mode': 'interior', 'id': 'QDR.A'}),
+ ('listener', {'role': 'normal', 'port': cls.tester.get_port()}),
+ ('connector', {'name': 'connectorToB', 'role': 'inter-router',
+ 'port': inter_router_port, 'verifyHostName': 'no'}),
+ ]
+ )
+
+ router('QDR.C', [
+ ('router', {'mode': 'interior', 'id': 'QDR.C'}),
+ ('listener', {'role': 'inter-router', 'port': cls.backup_port}),
+ ('listener', {'role': 'normal', 'port': cls.tester.get_port()}),
+ ]
+ )
+
+ cls.routers[1].wait_router_connected('QDR.B')
+
+ def address(self):
+ return self.routers[1].addresses[0]
+
+ def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK, address=None):
+ p = self.popen(
+ ['qdmanage'] + cmd.split(' ') + ['--bus', address or self.address(), '--indent=-1', '--timeout', str(TIMEOUT)],
+ stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect)
+ out = p.communicate(input)[0]
+ try:
+ p.teardown()
+ except Exception, e:
+ raise Exception("%s\n%s" % (e, out))
+ return out
+
+ def run_qdstat(self, args, regexp=None, address=None):
+ p = self.popen(
+ ['qdstat', '--bus', str(address or self.router.addresses[0]), '--timeout', str(system_test.TIMEOUT) ] + args,
+ name='qdstat-'+self.id(), stdout=PIPE, expect=None)
+
+ out = p.communicate()[0]
+ assert p.returncode == 0, \
+ "qdstat exit status %s, output:\n%s" % (p.returncode, out)
+ if regexp: assert re.search(regexp, out, re.I), "Can't find '%s' in '%s'" % (regexp, out)
+ return out
+
+ def test_connector_has_failover_list(self):
+ """
+ Makes a qdmanage connector query and checks if Router A is storing the failover information received from
+ Router B.
+ :return:
+ """
+ long_type = 'org.apache.qpid.dispatch.connector'
+ query_command = 'QUERY --type=' + long_type
+ output = json.loads(self.run_qdmanage(query_command))
+ self.assertIn(FailoverTest.failover_list, output[0]['failoverList'])
+
+ def test_remove_router_B(self):
+ # First make sure there are no inter-router connections on router C
+ outs = self.run_qdstat(['--connections'], address=self.routers[2].addresses[1])
+ self.assertNotIn('inter-router', outs)
+
+ # Kill the router B
+ FailoverTest.routers[0].teardown()
+
+ # Make sure that the router B is gone
+ # You need to sleep 5 seconds for the router to cycle thru the failover urls and make a successful connection
+ # to Router C
+ sleep(4)
+
+ long_type = 'org.apache.qpid.dispatch.connector'
+ query_command = 'QUERY --type=' + long_type
+ output = json.loads(self.run_qdmanage(query_command, address=self.routers[1].addresses[0]))
+ # The failoverList must now be gone since the backup router does not send a failoverList in its
+ # connection properties.
+ self.assertIsNone(output[0].get('failoverList'))
+
+ # Since router B has been killed, router A should now try to connect to a listener on router C.
+ # Use qdstat to connect to router C and determine that there is an inter-router connection with router A.
+ self.run_qdstat(['--connections'], regexp=r'QDR.A.*inter-router.*', address=self.routers[2].addresses[1])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org