You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2016/04/22 23:55:26 UTC
[5/7] qpid-dispatch git commit: DISPATCH-10 - Add negotiation of
connection-cost for inter-router connections.
DISPATCH-10 - Add negotiation of connection-cost for inter-router connections.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/ff83fc26
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/ff83fc26
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/ff83fc26
Branch: refs/heads/master
Commit: ff83fc265b45893ef1f13f4ab75a431c160aa00a
Parents: 2cb107b
Author: Ted Ross <tr...@redhat.com>
Authored: Thu Apr 21 16:17:55 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Apr 22 17:51:18 2016 -0400
----------------------------------------------------------------------
include/qpid/dispatch/amqp.h | 8 +++++---
src/amqp.c | 6 ++++--
src/router_node.c | 33 +++++++++++++++++++++++++++++++++
src/server.c | 27 +++++++++++++++++----------
4 files changed, 59 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/include/qpid/dispatch/amqp.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/amqp.h b/include/qpid/dispatch/amqp.h
index 54773b5..99daf50 100644
--- a/include/qpid/dispatch/amqp.h
+++ b/include/qpid/dispatch/amqp.h
@@ -114,10 +114,12 @@ extern const char * const QD_CAPABILITY_ROUTER_DATA;
extern const char * const QD_DYNAMIC_NODE_PROPERTY_ADDRESS; ///< Address for routing dynamic sources
/// @}
-/** @name Miscellaneous Strings */
+/** @name Connection Properties */
/// @{
-extern const char * const QD_INTERNODE_LINK_NAME_1;
-extern const char * const QD_INTERNODE_LINK_NAME_2;
+extern const char * const QD_CONNECTION_PROPERTY_PRODUCT_KEY;
+extern const char * const QD_CONNECTION_PROPERTY_PRODUCT_VALUE;
+extern const char * const QD_CONNECTION_PROPERTY_VERSION_KEY;
+extern const char * const QD_CONNECTION_PROPERTY_COST_KEY;
/// @}
/** @name AMQP error codes. */
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/src/amqp.c
----------------------------------------------------------------------
diff --git a/src/amqp.c b/src/amqp.c
index 656f7ef..d3b02f2 100644
--- a/src/amqp.c
+++ b/src/amqp.c
@@ -31,8 +31,10 @@ const char * const QD_CAPABILITY_ANONYMOUS_RELAY = "ANONYMOUS-RELAY";
const char * const QD_DYNAMIC_NODE_PROPERTY_ADDRESS = "x-opt-qd.address";
-const char * const QD_INTERNODE_LINK_NAME_1 = "qd.internode.1";
-const char * const QD_INTERNODE_LINK_NAME_2 = "qd.internode.2";
+const char * const QD_CONNECTION_PROPERTY_PRODUCT_KEY = "product";
+const char * const QD_CONNECTION_PROPERTY_PRODUCT_VALUE = "qpid-dispatch-router";
+const char * const QD_CONNECTION_PROPERTY_VERSION_KEY = "version";
+const char * const QD_CONNECTION_PROPERTY_COST_KEY = "qd.inter-router-cost";
const qd_amqp_error_t QD_AMQP_OK = { 200, "OK" };
const qd_amqp_error_t QD_AMQP_CREATED = { 201, "Created" };
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/src/router_node.c
----------------------------------------------------------------------
diff --git a/src/router_node.c b/src/router_node.c
index 8cba85e..a6431fe 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -500,6 +500,7 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
{
qdr_connection_role_t role = 0;
int cost = 1;
+ int remote_cost = 1;
bool strip_annotations_in = false;
bool strip_annotations_out = false;
int link_capacity = 1;
@@ -509,6 +510,38 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
qd_router_connection_get_config(conn, &role, &cost, &name,
&strip_annotations_in, &strip_annotations_out, &link_capacity);
+ if (role == QDR_ROLE_INTER_ROUTER) {
+ //
+ // Check the remote properties for an inter-router cost value.
+ //
+ pn_data_t *props = pn_conn ? pn_connection_remote_properties(pn_conn) : 0;
+ if (props) {
+ pn_data_rewind(props);
+ pn_data_next(props);
+ if (props && pn_data_type(props) == PN_MAP) {
+ pn_data_enter(props);
+ 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_COST_KEY) &&
+ strcmp(sym.start, QD_CONNECTION_PROPERTY_COST_KEY) == 0) {
+ pn_data_next(props);
+ if (pn_data_type(props) == PN_INT)
+ remote_cost = pn_data_get_int(props);
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ //
+ // Use the larger of the local and remote costs for this connection
+ //
+ if (remote_cost > cost)
+ cost = remote_cost;
+ }
+
qdr_connection_t *qdrc = qdr_connection_opened(router->router_core, inbound, role, cost, name,
pn_connection_remote_container(pn_conn),
strip_annotations_in, strip_annotations_out, link_capacity);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/ff83fc26/src/server.c
----------------------------------------------------------------------
diff --git a/src/server.c b/src/server.c
index 785bb31..7750056 100644
--- a/src/server.c
+++ b/src/server.c
@@ -454,12 +454,9 @@ static const char *log_incoming(char *buf, size_t size, qdpn_connector_t *cxtr)
}
-static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn)
+static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn, const qd_server_config_t *config)
{
size_t clen = strlen(QD_CAPABILITY_ANONYMOUS_RELAY);
- static char *product_key = "product";
- static char *product_val = "qpid-dispatch-router";
- static char *version_key = "version";
//
// Set the container name
@@ -477,11 +474,21 @@ static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn)
pn_data_put_map(pn_connection_properties(conn));
pn_data_enter(pn_connection_properties(conn));
- pn_data_put_symbol(pn_connection_properties(conn), pn_bytes(strlen(product_key), product_key));
- pn_data_put_string(pn_connection_properties(conn), pn_bytes(strlen(product_val), product_val));
+ pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_PRODUCT_KEY), QD_CONNECTION_PROPERTY_PRODUCT_KEY));
+ pn_data_put_string(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_PRODUCT_VALUE), QD_CONNECTION_PROPERTY_PRODUCT_VALUE));
- pn_data_put_symbol(pn_connection_properties(conn), pn_bytes(strlen(version_key), version_key));
- pn_data_put_string(pn_connection_properties(conn), pn_bytes(strlen(QPID_DISPATCH_VERSION), QPID_DISPATCH_VERSION));
+ pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_VERSION_KEY), QD_CONNECTION_PROPERTY_VERSION_KEY));
+ pn_data_put_string(pn_connection_properties(conn),
+ pn_bytes(strlen(QPID_DISPATCH_VERSION), QPID_DISPATCH_VERSION));
+
+ if (config && config->inter_router_cost > 1) {
+ pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_COST_KEY), QD_CONNECTION_PROPERTY_COST_KEY));
+ pn_data_put_int(pn_connection_properties(conn), config->inter_router_cost);
+ }
pn_data_exit(pn_connection_properties(conn));
}
@@ -536,7 +543,7 @@ static void thread_process_listeners_LH(qd_server_t *qd_server)
pn_connection_t *conn = pn_connection();
ctx->collector = pn_collector();
pn_connection_collect(conn, ctx->collector);
- decorate_connection(qd_server, conn);
+ decorate_connection(qd_server, conn, ctx->listener->config);
qdpn_connector_set_connection(cxtr, conn);
pn_connection_set_context(conn, ctx);
ctx->pn_conn = conn;
@@ -1086,7 +1093,7 @@ static void cxtr_try_open(void *context)
qd_log(ct->server->log_source, QD_LOG_TRACE, "Connecting to %s:%s", ct->config->host, ct->config->port);
pn_connection_collect(ctx->pn_conn, ctx->collector);
- decorate_connection(ctx->server, ctx->pn_conn);
+ decorate_connection(ctx->server, ctx->pn_conn, ct->config);
//
// qdpn_connector is not thread safe
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org