On 09/06/2011 11:01 AM, Zane Bitter wrote:
Replace the broker wrapper script with a C wrapper that calls
execve() on
qpidd. Implement parsing of the MATAHARI_PORT environment variable and the
Broker Federation options in the C wrapper.
Broker federation is configured using the BROKER_FEDERATION environment
variable set in /etc/sysconfig/matahari-broker. The contents of the
variable should be a list of hosts or DNS SRV queries separated by any
combinations of spaces, commas and/or semicolons.
Signed-off-by: Zane Bitter<zbitter(a)redhat.com>
---
CMakeLists.txt | 12 ++--
matahari-broker.init.in | 9 +++
matahari-broker.service | 19 -------
matahari-broker.service.in | 22 ++++++++
matahari-broker.sysconf.in | 23 ++++----
matahari-brokerd.in | 8 ---
src/CMakeLists.txt | 1
src/broker/CMakeLists.txt | 16 ++++++
src/broker/broker.c | 88 +++++++++++++++++++++++++++++++
src/broker/broker.h | 35 +++++++++++++
src/broker/broker_federation.c | 112 ++++++++++++++++++++++++++++++++++++++++
src/broker/broker_federation.h | 33 ++++++++++++
src/broker/broker_linux.c | 71 +++++++++++++++++++++++++
src/broker/broker_os.h | 51 ++++++++++++++++++
src/include/config.h.in | 9 +++
15 files changed, 463 insertions(+), 46 deletions(-)
delete mode 100644 matahari-broker.service
create mode 100644 matahari-broker.service.in
delete mode 100644 matahari-brokerd.in
create mode 100644 src/broker/CMakeLists.txt
create mode 100644 src/broker/broker.c
create mode 100644 src/broker/broker.h
create mode 100644 src/broker/broker_federation.c
create mode 100644 src/broker/broker_federation.h
create mode 100644 src/broker/broker_linux.c
create mode 100644 src/broker/broker_os.h
diff --git a/CMakeLists.txt b/CMakeLists.txt
index f7555b4..4ec2d1e 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -109,9 +109,6 @@ else(WIN32)
set (CPACK_GENERATOR "TGZ")
- configure_file(${CMAKE_CURRENT_SOURCE_DIR}/matahari-brokerd.in
- ${CMAKE_CURRENT_BINARY_DIR}/matahari-brokerd)
-
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/matahari.init.in
${CMAKE_CURRENT_BINARY_DIR}/matahari.init)
@@ -126,8 +123,6 @@ else(WIN32)
GROUP_READ GROUP_WRITE GROUP_EXECUTE)
install(CODE "execute_process(COMMAND chown root:qpidd
${localstatedir}/lib/matahari)")
- install(PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/matahari-brokerd DESTINATION sbin)
-
install(PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/matahari-broker DESTINATION ${initdir})
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/matahari.sysconf DESTINATION
${sysconfdir}/sysconfig RENAME matahari)
@@ -142,8 +137,13 @@ else(WIN32)
if(systemd_FOUND)
set(systemdunitdir "/lib/systemd/system")
- install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/matahari-broker.service DESTINATION
${systemdunitdir})
+
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/matahari-broker.service.in
+ ${CMAKE_CURRENT_BINARY_DIR}/matahari-broker.service)
+
+ install(FILES ${CMAKE_CURRENT_BINARY_DIR}/matahari-broker.service DESTINATION
${systemdunitdir})
endif(systemd_FOUND)
+
endif(WIN32)
INCLUDE(InstallRequiredSystemLibraries)
diff --git a/matahari-broker.init.in b/matahari-broker.init.in
index e74926c..d9ca4ee 100755
--- a/matahari-broker.init.in
+++ b/matahari-broker.init.in
@@ -46,12 +46,19 @@ pidfile=@localstatedir(a)/run/matahari-broker.pid
# Source function library.
. @sysconfdir(a)/rc.d/init.d/functions
+if [ -f @sysconfdir@/sysconfig/matahari ]; then
+ . @sysconfdir@/sysconfig/matahari
+fi
+if [ -f @sysconfdir@/sysconfig/matahari-broker ]; then
+ . @sysconfdir@/sysconfig/matahari-broker
+fi
+
RETVAL=0
start() {
echo -n $"Starting Matahari broker daemon: "
install -o root -g qpidd -m 775 -d @localstatedir@/run/matahari
- daemon --pidfile $pidfile --check $proc --user qpidd /usr/sbin/$prog --daemon
+ daemon --pidfile $pidfile --check $proc --user qpidd /usr/sbin/$prog $QPIDD_OPTIONS
--daemon
RETVAL=$?
echo
[ $RETVAL = 0 ]&& touch $lockfile
diff --git a/matahari-broker.service b/matahari-broker.service
deleted file mode 100644
index 7e6e9f7..0000000
--- a/matahari-broker.service
+++ /dev/null
@@ -1,19 +0,0 @@
-[Unit]
-Description=Matahari Broker is an AMQP broker. It receives, stores, routes and forwards
messages using the AMQP protcol.
-After=syslog.target network.target
-Requires=syslog.target network.target
-
-[Install]
-WantedBy=multi-user.target
-
-[Service]
-Type=simple
-NotifyAccess=all
-SysVStartPriority=85
-User=qpidd
-Group=qpidd
-
-ExecStart=/usr/sbin/matahari-brokerd
-
-# Restart options include: no, on-success, on-failure, on-abort or always
-Restart=on-failure
diff --git a/matahari-broker.service.in b/matahari-broker.service.in
new file mode 100644
index 0000000..62898fb
--- /dev/null
+++ b/matahari-broker.service.in
@@ -0,0 +1,22 @@
+[Unit]
+Description=Matahari Broker is an AMQP broker. It receives, stores, routes and forwards
messages using the AMQP protcol.
+After=syslog.target network.target
+Requires=syslog.target network.target
+
+[Install]
+WantedBy=multi-user.target
+
+[Service]
+Type=simple
+NotifyAccess=all
+SysVStartPriority=85
+User=qpidd
+Group=qpidd
+
+EnvironmentFile=-@sysconfdir@/sysconfig/matahari
+EnvironmentFile=-@sysconfdir@/sysconfig/matahari-broker
+
+ExecStart=/usr/sbin/matahari-brokerd $QPIDD_OPTIONS
+
+# Restart options include: no, on-success, on-failure, on-abort or always
+Restart=on-failure
diff --git a/matahari-broker.sysconf.in b/matahari-broker.sysconf.in
index 9b1c2b5..367e00c 100644
--- a/matahari-broker.sysconf.in
+++ b/matahari-broker.sysconf.in
@@ -1,13 +1,12 @@
-# Try to use the centrally defined port (also used by the agents)
-# Otherwise, use the default port specified at build-time
-
-if [ -f @sysconfdir@/sysconfig/matahari ] ; then
- . @sysconfdir@/sysconfig/matahari
-fi
-
-if [ -z $MATAHARI_PORT ]; then
- MATAHARI_PORT=@MATAHARI_PORT@
-fi
-
-QPIDD_OPTIONS="--config @sysconfdir(a)/matahari-broker.conf --port
$MATAHARI_PORT"
+QPIDD_OPTIONS="--config @sysconfdir(a)/matahari-broker.conf"
+
+# Federated Broker configuration - list the brokers to federate with
+# The following formats are allowed:
+# Hostname and port
+#<fqdn>:<port>
+# DNS SRV record format ("_matahari._tcp" is REQUIRED)
+# _matahari._tcp.<fqdn>
+#
+# Addresses need not all be in the same format
+#FEDERATED_BROKERS="broker1.example.com:49000, _matahari._tcp.example.com"
It would be good to update the comment here to make it clear that only
the highest priority SRV record will be used.
diff --git a/matahari-brokerd.in b/matahari-brokerd.in
deleted file mode 100644
index 0bab652..0000000
--- a/matahari-brokerd.in
+++ /dev/null
@@ -1,8 +0,0 @@
-#!/bin/sh
-
-if [ -f @sysconfdir@/sysconfig/matahari-broker ]; then
- . @sysconfdir@/sysconfig/matahari-broker
-fi
-
-/usr/sbin/qpidd --data-dir @localstatedir@/lib/matahari $QPIDD_OPTIONS $*
-
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index f045887..e3a7e38 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -162,6 +162,7 @@ endif(WITH-QMF)
### Subdirectories
+add_subdirectory(broker)
add_subdirectory(lib)
add_subdirectory(host)
add_subdirectory(network)
diff --git a/src/broker/CMakeLists.txt b/src/broker/CMakeLists.txt
new file mode 100644
index 0000000..7b48387
--- /dev/null
+++ b/src/broker/CMakeLists.txt
@@ -0,0 +1,16 @@
+SET(BROKER_DAEMON "matahari-brokerd")
+
+if(WITH-QMF)
+if(NOT WIN32)
+ add_executable(${BROKER_DAEMON} broker.c broker_federation.c broker_${VARIANT}.c)
+
+ if(HAVE_RESOLV_H)
+ target_link_libraries(${BROKER_DAEMON} mcommon)
+ target_link_libraries(${BROKER_DAEMON} resolv)
+ endif(HAVE_RESOLV_H)
+
+ install(TARGETS ${BROKER_DAEMON} DESTINATION sbin)
+
+endif(NOT WIN32)
+endif(WITH-QMF)
+
diff --git a/src/broker/broker.c b/src/broker/broker.c
new file mode 100644
index 0000000..f6d7d73
--- /dev/null
+++ b/src/broker/broker.c
@@ -0,0 +1,88 @@
+/* Copyright (C) 2011 Red Hat, Inc.
+ * Written by Zane Bitter<zbitter(a)redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+#include "config.h"
+
+#include<stdlib.h>
+#include<stdint.h>
+#include<string.h>
+#include<stdio.h>
+
+#include "broker_os.h"
+#include "broker.h"
+
+
+uint16_t broker_get_port(void)
+{
+ static uint16_t port = 0;
+
+ if (!port) {
+ char *port_env = getenv("MATAHARI_PORT");
+ if (port_env) {
+ sscanf(port_env, "%hu",&port);
How about a check for invalid input?
+ }
+ }
+
+ return port ? port : MATAHARI_PORT;
+}
+
+static char **broker_args(int argc, char * const argv[])
+{
+ int i = 0, o = 0;
+ char portarg[13];
+ char **newargs = malloc(sizeof(char *) * (argc + 2 + 1));
+
+ if (argc) {
+ newargs[o++] = strdup(argv[i++]);
+ }
+
+ snprintf(portarg, sizeof(portarg), "--port=%hu", broker_get_port());
+ newargs[o++] = strdup(portarg);
+ newargs[o++] = strdup("--data-dir=" LOCAL_STATE_DIR
"/lib/matahari");
+
+ while (i< argc) {
+ newargs[o++] = strdup(argv[i++]);
+ }
+
+ newargs[o] = NULL;
+
+ return newargs;
+}
+
+static void broker_args_free(char **args)
+{
+ char **p = args;
+
+ while (*args) {
+ free(*args++);
+ }
+
+ free(p);
+}
+
+int main(int argc, char *argv[])
+{
+ char **arglist = broker_args(argc, argv);
+
+ int ret = broker_os_start_broker(arglist);
+
+ broker_args_free(arglist);
+
+ return ret;
+}
+
diff --git a/src/broker/broker.h b/src/broker/broker.h
new file mode 100644
index 0000000..3aff4d3
--- /dev/null
+++ b/src/broker/broker.h
@@ -0,0 +1,35 @@
+/* Copyright (C) 2011 Red Hat, Inc.
+ * Written by Zane Bitter<zbitter(a)redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+/**
+ * @file broker.h
+ * Interface to the broker configuration code.
+ */
+
+#ifndef MH_BROKER_H
+#define MH_BROKER_H
+
+#include<stdint.h>
+
+/**
+ * Get the port for the broker to listen on.
+ * @return the TCP port for the local broker
+ */
+uint16_t broker_get_port(void);
+
+#endif
diff --git a/src/broker/broker_federation.c b/src/broker/broker_federation.c
new file mode 100644
index 0000000..5140657
--- /dev/null
+++ b/src/broker/broker_federation.c
@@ -0,0 +1,112 @@
+/* Copyright (C) 2011 Red Hat, Inc.
+ * Written by Zane Bitter<zbitter(a)redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+#include "config.h"
+
+#include<stdlib.h>
+#include<string.h>
+#include<stdio.h>
+
+#include<matahari/dnssrv.h>
+#include<matahari/utilities.h>
+#include<matahari/logging.h>
+
+#include "broker.h"
+#include "broker_os.h"
+#include "broker_federation.h"
+
+
+#define DNS_SRV_PREFIX "_matahari._tcp."
+
+#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_federate(const char *local, const char *remote)
+{
+ int i;
+ struct mh_qpid_route routes[] = ROUTE_LIST(local, remote);
+
+ mh_trace("Adding routes for federated broker \"%s\"", remote);
+
+ 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);
+ return ret;
+ }
+ }
+
+ return 0;
+}
+
+static char *broker_lookup(const char *query)
+{
+#ifdef HAVE_RESOLV_H
+ static char broker[1025 + 6];
+ char *host;
+ mh_trace("Looking up DNS SRV record for \"%s\"", query);
+
+ host = mh_dnssrv_lookup(query);
+
+ if (host) {
+ snprintf(broker, sizeof(broker), "%s:%hu", host, MATAHARI_PORT);
+ return broker;
+ }
+ mh_warn("Failed to find DNS SRV record for \"%s\"", query);
+#else
+ mh_warn("DNS SRV lookups are not supported");
+#endif
+
+ return NULL;
+}
+
+void broker_federation_configure(void)
+{
+ const char *brokers = getenv("FEDERATED_BROKERS");
+ char local[16];
+ char peers[1024];
+ char *p = NULL;
+ char *peer;
+
+ if (!brokers) {
+ return;
+ }
+
+ snprintf(local, sizeof(local), "localhost:%hu", broker_get_port());
+
+ strncpy(peers, brokers, sizeof(peers) - 1);
+ peers[sizeof(peers) - 1] = '\0';
+
+ while ((peer = strtok_r(p ? NULL : peers, ",; ",&p)) != NULL) {
+ if (strncmp(peer, DNS_SRV_PREFIX, strlen(DNS_SRV_PREFIX)) == 0) {
+ peer = broker_lookup(peer);
+ }
+
+ if (peer) {
+ broker_federate(local, peer);
+ }
+ }
+}
+
diff --git a/src/broker/broker_federation.h b/src/broker/broker_federation.h
new file mode 100644
index 0000000..0cacaef
--- /dev/null
+++ b/src/broker/broker_federation.h
@@ -0,0 +1,33 @@
+/* Copyright (C) 2011 Red Hat, Inc.
+ * Written by Zane Bitter<zbitter(a)redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+/**
+ * @file broker_federation.h
+ * Interface to the broker federation configuration code.
+ */
+
+#ifndef MH_BROKER_FEDERATION_H
+#define MH_BROKER_FEDERATION_H
+
+/**
+ * Read the broker federation settings from the environment and set up broker
+ * federation.
+ */
+void broker_federation_configure(void);
+
+#endif
diff --git a/src/broker/broker_linux.c b/src/broker/broker_linux.c
new file mode 100644
index 0000000..6dd98e8
--- /dev/null
+++ b/src/broker/broker_linux.c
@@ -0,0 +1,71 @@
+/* Copyright (C) 2011 Red Hat, Inc.
+ * Written by Zane Bitter<zbitter(a)redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+#include<stdlib.h>
+#include<string.h>
+#include<stdio.h>
+#include<errno.h>
+#include<unistd.h>
+
+#include "broker_federation.h"
+#include "broker_os.h"
+
+
+int broker_os_start_broker(char * const args[])
+{
+ pid_t pid = fork();
+
+ if (pid< 0) {
+ perror("fork()");
+ return errno;
+ }
+
+ if (pid) {
+ // exec() in parent process so that qpidd keeps the initial pid.
+ // Note that this means the child will become a zombie. That's just the
+ // way it goes.
+
+ // execvp() will find the path to qpidd from $PATH.
+ execvp("qpidd", args);
+
+ // It's an error if we get here
+ perror("exec()");
+ return errno;
+ } else {
+ // Configure federated brokers in the child process
+ broker_federation_configure();
+ return 0;
+ }
+}
+
+int broker_os_add_qpid_route(const struct mh_qpid_route *route)
+{
+ char cmd[1024];
+ int ret;
+
+ 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) {
+ perror("system()");
+ }
+ return ret;
+}
+
diff --git a/src/broker/broker_os.h b/src/broker/broker_os.h
new file mode 100644
index 0000000..788fe78
--- /dev/null
+++ b/src/broker/broker_os.h
@@ -0,0 +1,51 @@
+/* Copyright (C) 2011 Red Hat, Inc.
+ * Written by Zane Bitter<zbitter(a)redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+/**
+ * @file broker_os.h
+ * Interface to the OS-specific broker configuration code.
+ */
+
+#ifndef MH_BROKER_OS_H
+#define MH_BROKER_OS_H
+
+/**
+ * A route between two Qpid brokers.
+ */
+struct mh_qpid_route {
+ const char *dest; /**< The route destination */
+ const char *src; /**< The route source */
+ const char *exchange; /**< The exchange */
+};
+
+
+/**
+ * Start the broker with the supplied argument list.
+ * @param args a list of args to pass to the broker. The last item in the list
+ * must be NULL.
+ */
+int broker_os_start_broker(char * const args[]);
+
+/**
+ * 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.
+ */
+int broker_os_add_qpid_route(const struct mh_qpid_route *route);
+
+#endif
diff --git a/src/include/config.h.in b/src/include/config.h.in
index 6231970..fd248ad 100644
--- a/src/include/config.h.in
+++ b/src/include/config.h.in
@@ -1,3 +1,6 @@
+#ifndef MH_CONFIG_H
+#define MH_CONFIG_H
+
#cmakedefine HAVE_STRING_H 1
#cmakedefine HAVE_SYS_IOCTL_H 1
#cmakedefine HAVE_ASPRINTF 1
@@ -5,3 +8,9 @@
#cmakedefine HAVE_TIME 1
#cmakedefine HAVE_G_LIST_FREE_FULL 1
#cmakedefine HAVE_PK_GET_SYNC 1
+
+#define LOCAL_STATE_DIR "@localstatedir@"
+
+#define MATAHARI_PORT @MATAHARI_PORT@
+
+#endif
_______________________________________________
Matahari mailing list
Matahari(a)lists.fedorahosted.org
https://fedorahosted.org/mailman/listinfo/matahari
--
Russell Bryant