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(a)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.
--
1.7.6.4