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>
---
matahari-broker.init.in | 5 ++
matahari-broker.service.in | 3 +-
matahari.spec.in | 26 +++++++++++
src/broker/broker_federation.c | 91 +++++++++++++++++++++++++++-------------
src/broker/broker_linux.c | 29 +++++++++++-
src/broker/broker_os.h | 11 +++++
src/lib/mh_agent.cpp | 2 +-
7 files changed, 132 insertions(+), 35 deletions(-)
diff --git a/matahari-broker.init.in b/matahari-broker.init.in
index d9ca4ee..0f03b44 100755
--- a/matahari-broker.init.in
+++ b/matahari-broker.init.in
@@ -27,6 +27,7 @@
# Required-Stop: $local_fs
# Default-Start: 2 3 4 5
# Default-Stop: 0 1 6
+# Should-Start: matahari-vios-proxy-guest
# Short-Description: start or stop matahari broker
# Description: Matahari Broker is an AMQP broker. It receives, stores, routes and
forwards messages using the AMQP protcol.
### END INIT INFO
@@ -53,6 +54,10 @@ if [ -f @sysconfdir@/sysconfig/matahari-broker ]; then
. @sysconfdir@/sysconfig/matahari-broker
fi
+if [ -f @sysconfdir@/sysconfig/matahari-vios-proxy ]; then
+ . @sysconfdir@/sysconfig/matahari-vios-proxy
+fi
+
RETVAL=0
start() {
diff --git a/matahari-broker.service.in b/matahari-broker.service.in
index 62898fb..5f28f48 100644
--- a/matahari-broker.service.in
+++ b/matahari-broker.service.in
@@ -1,6 +1,6 @@
[Unit]
Description=Matahari Broker is an AMQP broker. It receives, stores, routes and forwards
messages using the AMQP protcol.
-After=syslog.target network.target
+After=syslog.target network.target matahari-vios-proxy-guest.service
Requires=syslog.target network.target
[Install]
@@ -15,6 +15,7 @@ Group=qpidd
EnvironmentFile=-@sysconfdir@/sysconfig/matahari
EnvironmentFile=-@sysconfdir@/sysconfig/matahari-broker
+EnvironmentFile=-@sysconfdir@/sysconfig/matahari-vios-proxy
ExecStart=/usr/sbin/matahari-brokerd $QPIDD_OPTIONS
diff --git a/matahari.spec.in b/matahari.spec.in
index 130c4d8..a2934a2 100644
--- a/matahari.spec.in
+++ b/matahari.spec.in
@@ -100,6 +100,7 @@ Requires: %{name}-core = %{version}-%{release}
Requires: qpid-cpp-server > 0.7
Requires: qpid-cpp-server-ssl > 0.7
Requires: qmf > 0.7
+Requires: qpid-tools
Requires(post): chkconfig
Requires(preun): chkconfig
Requires(preun): initscripts
@@ -114,6 +115,9 @@ Summary: Network proxy using virtioserial for QEMU host
Group: Applications/System
Requires: %{name}-broker = %{version}-%{release}
Requires: vios-proxy-host
+Requires(post): chkconfig
+Requires(preun): chkconfig
+Requires(preun): initscripts
%description vios-proxy-host
%{name} specific host initialization scripts
@@ -124,6 +128,9 @@ Summary: Network proxy using virtioserial for QEMU guest
Group: Applications/System
Requires: %{name}-broker = %{version}-%{release}
Requires: vios-proxy-guest
+Requires(post): chkconfig
+Requires(preun): chkconfig
+Requires(preun): initscripts
%description vios-proxy-guest
%{name} specific guest initialization scripts
@@ -399,6 +406,25 @@ if [ "$1" -ge "1" ]; then
/sbin/service matahari-vios-proxy-host condrestart >/dev/null 2>&1 || :
fi
+#== Vios Proxy guest
+%post vios-proxy-guest
+%if %{systemd}
+systemctl --system daemon-reload
+%endif
+/sbin/service matahari-vios-proxy-guest condrestart
+
+%preun vios-proxy-guest
+if [ $1 = 0 ]; then
+ /sbin/service matahari-vios-proxy-guest stop >/dev/null 2>&1 || :
+%if !%{systemd}
+ chkconfig --del matahari-vios-proxy-guest
+%endif
+fi
+
+%postun vios-proxy-guest
+if [ "$1" -ge "1" ]; then
+ /sbin/service matahari-vios-proxy-guest condrestart >/dev/null 2>&1 || :
+fi
#== Core
%pre core
diff --git a/src/broker/broker_federation.c b/src/broker/broker_federation.c
index e775d3e..efdb6fa 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,27 +39,47 @@
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"} \
+static int
+broker_add_route(const char *local, const char *remote,
+ const char *exchange, const char *route_key, int srclocal,
+ int aggregate)
+{
+ int ret;
+ struct mh_qpid_route route[] = {
+ { local, remote, exchange, route_key, srclocal, aggregate }
+ };
+
+ mh_info("Adding routes for federated broker \"%s\"", remote);
+ ret = broker_os_add_qpid_route(route);
+ if (ret) {
+ mh_err("Failed to route %s => %s", local, remote);
+ }
+ return ret;
}
-
-static int broker_federate(const char *local, const char *remote)
+static int
+broker_federate(const char *local, const char *remote, int route_type)
{
- int i;
- struct mh_qpid_route routes[] = ROUTE_LIST(local, remote);
-
- mh_trace("Adding routes for federated broker \"%s\"", remote);
+ 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;
+ }
- for (i = 0; i < DIMOF(routes); i++) {
- int ret = broker_os_add_qpid_route(&routes[i]);
- if (ret) {
- mh_err("Failed to add federated broker \"%s\"", remote);
+ if (route_type) {
+ if ((broker_add_route(local, remote, "qmf.default.topic",
"direct-agent.#", FALSE, TRUE) ||
+ broker_add_route(local, remote, "qmf.default.topic",
"console.#", FALSE, TRUE) ||
+ broker_add_route(remote, local, "qmf.default.topic",
"direct-console.#", TRUE, TRUE) ||
+ broker_add_route(remote, local, "qmf.default.topic",
"agent.#", TRUE, TRUE)) < 0) {
+ return ret;
+ }
+ } else {
+ if ((broker_add_route(remote, local, "amq.direct", NULL, FALSE, FALSE)
||
+ broker_add_route(local, remote, "amq.direct", NULL, FALSE, FALSE)
||
+ broker_add_route(remote, local, "qmf.default.direct", NULL, FALSE,
FALSE) ||
+ broker_add_route(local, remote, "qmf.default.direct", NULL, FALSE,
FALSE) ||
+ broker_add_route(remote, local, "qmf.default.topic", NULL, FALSE,
FALSE) ||
+ broker_add_route(local, remote, "qmf.default.topic", NULL, FALSE,
FALSE)) < 0) {
return ret;
}
}
@@ -66,7 +87,8 @@ static int broker_federate(const char *local, const char *remote)
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 +112,39 @@ 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");
+ const char *proxy_service_dir = getenv("VP_G_HOST_DIR");
char local[16];
char peers[1024];
char *p = NULL;
char *peer;
- if (!brokers) {
- return;
- }
+ if (g_file_test(proxy_service_dir, G_FILE_TEST_IS_DIR)) {
+ char remote[16];
+ snprintf(remote, sizeof(local), "localhost:%s", proxy_service_port);
+ snprintf(local, sizeof(remote), "localhost:%hu", broker_get_port());
+ broker_federate(local, remote, TRUE);
+ } 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, FALSE);
+ }
}
- if (peer) {
- broker_federate(local, peer);
- }
+ } else {
+ return;
}
}
diff --git a/src/broker/broker_linux.c b/src/broker/broker_linux.c
index 997734f..8d5689d 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 untrusted/untrusted@%s
PLAIN",
+ 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..7f8f8a9 100644
--- a/src/broker/broker_os.h
+++ b/src/broker/broker_os.h
@@ -31,6 +31,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 */
+ int srclocal; /**< Is this route static? */
+ int aggregate; /**< Is this route part of aggregate? */
};
@@ -42,6 +45,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.
diff --git a/src/lib/mh_agent.cpp b/src/lib/mh_agent.cpp
index 5417680..49098b0 100644
--- a/src/lib/mh_agent.cpp
+++ b/src/lib/mh_agent.cpp
@@ -449,7 +449,7 @@ mh_parse_options(const char *proc_name, int argc, char **argv,
OptionsMap &optio
/* Force local-only handling */
mh_add_option('b', required_argument, "broker",
"specify broker host name", &options, map_option);
mh_add_option('D', no_argument, "dns-srv",
"interpret the value of --broker as a domain name for DNS SRV lookups",
&options, map_option);
- mh_add_option('p', required_argument, "port",
"specify broker ", &options, map_option);
+ mh_add_option('p', required_argument, "port",
"specify broker port", &options, map_option);
mh_add_option('v', no_argument, "verbose",
"Increase the log level", NULL, map_option);
mh_add_option('u', required_argument, "username", "username
to use for authentication to the broker", &amqp_options, connection_option);
--
1.7.6.4