This looks like it's on the right track. I'll assume you know what
you're doing w.r.t. the actual federation setup ;)
See inline for comments. You'll also need to make sure the
matahari-vios-proxy sysconf file is sourced in the init script for the
broker and added as an environment file in the service file. You should
also adjust the startup dependencies to ensure that matahari-broker
starts after vios-proxy-guest if the latter is enabled.
cheers,
Zane.
On 19/10/11 00:46, Adam Stokes wrote:
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);
+}
There's no reason these functions can't be static.
-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) {
route_type is not used anywhere else, so you might as well just pass a
boolean instead of a string.
+ 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;
}
}
Actually, I'm not clear why we need to put this into a GList at all. Why
not just call broker_os_add_qpid_route() from inside
mh_add_broker_route()? Or if error handling is a problem, just use a
local array.
+ 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) {
This condition is not enough to tell if we are a vios-proxy guest, since
VP_G_SERVICE_PORT will usually always be defined in the config file.
We should check for the existence of the appropriate file inside
$VP_G_HOST_DIR.
+ 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);
As per previous comment, these don't need to be exported.
#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? */
Just use int; no need to drag glib into it.
};
@@ -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.