Setting up some different routes for either using bi-directional or the push-route federation which is primarily used within the vios-proxy setup.
Signed-off-by: Adam Stokes astokes@fedoraproject.org --- src/broker/broker_federation.c | 101 +++++++++++++++++++++++++++++---------- src/broker/broker_federation.h | 8 +++ src/broker/broker_linux.c | 29 ++++++++++- src/broker/broker_os.h | 12 +++++ 4 files changed, 121 insertions(+), 29 deletions(-)
diff --git a/src/broker/broker_federation.c b/src/broker/broker_federation.c index e775d3e..fedceae 100644 --- a/src/broker/broker_federation.c +++ b/src/broker/broker_federation.c @@ -21,6 +21,7 @@ #include <stdlib.h> #include <string.h> #include <stdio.h> +#include <glib.h>
#include <matahari/dnssrv.h> #include <matahari/utilities.h> @@ -38,35 +39,75 @@ DNS_SRV_PREFIX_##TYPE, \ strlen(DNS_SRV_PREFIX_##TYPE)) == 0)
-#define ROUTE_LIST(LOCAL, REMOTE) { \ - {REMOTE, LOCAL, "amq.direct"}, \ - {LOCAL, REMOTE, "amq.direct"}, \ - {REMOTE, LOCAL, "qmf.default.direct"}, \ - {LOCAL, REMOTE, "qmf.default.direct"}, \ - {REMOTE, LOCAL, "qmf.default.topic"}, \ - {LOCAL, REMOTE, "qmf.default.topic"} \ +GList * +mh_add_broker_route(GList *routes, const char *local, const char *remote, + const char *exchange, const char *route_key, gboolean srclocal, + gboolean aggregate) +{ + struct mh_qpid_route *route; + + route = malloc(sizeof(*route)); + route->src = local; + route->dest = remote; + route->exchange = exchange; + route->route_key = route_key; + route->srclocal = srclocal; + route->aggregate = aggregate; + free(route); + + return g_list_append(routes, route); }
+void +mh_broker_route_free(gpointer data) +{ + struct mh_qpid_route *route = data; + + free(route); +}
-static int broker_federate(const char *local, const char *remote) +static int +broker_federate(const char *local, const char *remote, const char *route_type) { - int i; - struct mh_qpid_route routes[] = ROUTE_LIST(local, remote); + GList *routes = NULL; + + int ret = broker_os_add_qpid_route_link(local, remote); + if (ret) { + mh_err("Failed to link brokers "%s" to "%s"", local, remote); + return ret; + }
- mh_trace("Adding routes for federated broker "%s"", remote); + if (strcasecmp(route_type, "aggregate") == 0) { + mh_add_broker_route(routes, local, remote, "qmf.default.topic", "direct-agent.#", FALSE, TRUE); + mh_add_broker_route(routes, local, remote, "qmf.default.topic", "console.#", FALSE, TRUE); + mh_add_broker_route(routes, remote, local, "qmf.default.topic", "direct-console.#", TRUE, TRUE); + mh_add_broker_route(routes, remote, local, "qmf.default.topic", "agent.#", TRUE, TRUE); + } else { + mh_add_broker_route(routes, remote, local, "amq.direct", NULL, FALSE, FALSE); + mh_add_broker_route(routes, local, remote, "amq.direct", NULL, FALSE, FALSE); + mh_add_broker_route(routes, remote, local, "qmf.default.direct", NULL, FALSE, FALSE); + mh_add_broker_route(routes, local, remote, "qmf.default.direct", NULL, FALSE, FALSE); + mh_add_broker_route(routes, remote, local, "qmf.default.topic", NULL, FALSE, FALSE); + mh_add_broker_route(routes, local, remote, "qmf.default.topic", NULL, FALSE, FALSE); + }
- for (i = 0; i < DIMOF(routes); i++) { - int ret = broker_os_add_qpid_route(&routes[i]); + mh_trace("Adding routes for federated broker "%s"", remote); + + for (; routes; routes = g_list_next(routes)) { + struct mh_qpid_route *route = routes->data; + ret = broker_os_add_qpid_route(route); if (ret) { mh_err("Failed to add federated broker "%s"", remote); return ret; } } + g_list_free_full(routes, mh_broker_route_free);
return 0; }
-static char *broker_lookup(const char *query) +static char * +broker_lookup(const char *query) { #ifdef HAVE_RESOLV_H static char broker[1025 + 6]; @@ -90,30 +131,38 @@ static char *broker_lookup(const char *query) return NULL; }
-void broker_federation_configure(void) +void +broker_federation_configure(void) { const char *brokers = getenv("FEDERATED_BROKERS"); + const char *proxy_service_port = getenv("VP_G_SERVICE_PORT"); char local[16]; char peers[1024]; char *p = NULL; char *peer;
- if (!brokers) { - return; - } + if (proxy_service_port) { + char remote[16]; + snprintf(local, sizeof(local), "localhost:%s", proxy_service_port); + snprintf(remote, sizeof(remote), "localhost:%hu", broker_get_port()); + broker_federate(local, remote, "aggregate"); + } else if (brokers) { + snprintf(local, sizeof(local), "localhost:%hu", broker_get_port());
- snprintf(local, sizeof(local), "localhost:%hu", broker_get_port()); + mh_string_copy(peers, brokers, sizeof(peers));
- mh_string_copy(peers, brokers, sizeof(peers)); + while ((peer = strtok_r(p ? NULL : peers, ",; ", &p)) != NULL) { + if (IS_DNS_SRV(peer, TCP) || IS_DNS_SRV(peer, TLS)) { + peer = broker_lookup(peer); + }
- while ((peer = strtok_r(p ? NULL : peers, ",; ", &p)) != NULL) { - if (IS_DNS_SRV(peer, TCP) || IS_DNS_SRV(peer, TLS)) { - peer = broker_lookup(peer); + if (peer) { + broker_federate(local, peer, NULL); + } }
- if (peer) { - broker_federate(local, peer); - } + } else { + return; } }
diff --git a/src/broker/broker_federation.h b/src/broker/broker_federation.h index 0cacaef..d4806d4 100644 --- a/src/broker/broker_federation.h +++ b/src/broker/broker_federation.h @@ -24,10 +24,18 @@ #ifndef MH_BROKER_FEDERATION_H #define MH_BROKER_FEDERATION_H
+#include <glib.h> /** * Read the broker federation settings from the environment and set up broker * federation. */ void broker_federation_configure(void);
+void +mh_broker_route_free(gpointer data); + +GList * +mh_add_broker_route(GList *routes, const char *local, const char *remote, + const char *exchange, const char *route_key, gboolean srclocal, + gboolean aggregate); #endif diff --git a/src/broker/broker_linux.c b/src/broker/broker_linux.c index 997734f..962cdc7 100644 --- a/src/broker/broker_linux.c +++ b/src/broker/broker_linux.c @@ -55,14 +55,37 @@ int broker_os_start_broker(char * const args[]) } }
-int broker_os_add_qpid_route(const struct mh_qpid_route *route) +int broker_os_add_qpid_route_link(const char *local, const char *remote) { char cmd[1024]; int ret;
- snprintf(cmd, sizeof(cmd), "qpid-route --timeout=5 dynamic add %s %s %s", - route->dest, route->src, route->exchange); + snprintf(cmd, sizeof(cmd), "qpid-route link add %s %s", + local, remote); + + ret = system(cmd); + + if (ret < 0) { + perror("system()"); + } + return ret; +} + +int broker_os_add_qpid_route(const struct mh_qpid_route *route) +{ + char cmd[1024]; + int ret;
+ if (route->aggregate && route->srclocal) { + snprintf(cmd, sizeof(cmd), "qpid-route --src-local route add %s %s %s %s", + route->dest, route->src, route->exchange, route->route_key); + } else if (route->aggregate) { + snprintf(cmd, sizeof(cmd), "qpid-route route add %s %s %s %s", + route->dest, route->src, route->exchange, route->route_key); + } else { + snprintf(cmd, sizeof(cmd), "qpid-route --timeout=5 dynamic add %s %s %s", + route->dest, route->src, route->exchange); + } ret = system(cmd);
if (ret < 0) { diff --git a/src/broker/broker_os.h b/src/broker/broker_os.h index 788fe78..7c556ab 100644 --- a/src/broker/broker_os.h +++ b/src/broker/broker_os.h @@ -24,6 +24,7 @@ #ifndef MH_BROKER_OS_H #define MH_BROKER_OS_H
+#include <glib.h> /** * A route between two Qpid brokers. */ @@ -31,6 +32,9 @@ struct mh_qpid_route { const char *dest; /**< The route destination */ const char *src; /**< The route source */ const char *exchange; /**< The exchange */ + const char *route_key; /**< The routing key */ + gboolean srclocal; /**< Is this route static? */ + gboolean aggregate; /**< Is this route part of aggregate? */ };
@@ -42,6 +46,14 @@ struct mh_qpid_route { int broker_os_start_broker(char * const args[]);
/** + * Add a route link between two Qpid brokers. + * @param local local broker to link + * @param remote remote broker to link + * @return -1 on error, otherwise the result of the external route add command. + */ +int broker_os_add_qpid_route_link(const char *local, const char *remote); + +/** * Add a route between two Qpid brokers. * @param route pointer to the route to add * @return -1 on error, otherwise the result of the external route add command.