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 Bitterzbitter@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@/run/matahari-broker.pid # Source function library. . @sysconfdir@/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@/matahari-broker.conf --port $MATAHARI_PORT" +QPIDD_OPTIONS="--config @sysconfdir@/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 Bitterzbitter@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 Bitterzbitter@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 Bitterzbitter@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 Bitterzbitter@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 Bitterzbitter@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 Bitterzbitter@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@lists.fedorahosted.org https://fedorahosted.org/mailman/listinfo/matahari