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@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 | 74 ++++++++++++++++++++++++++++++++ src/broker/broker.h | 12 +++++ src/broker/broker_federation.c | 93 ++++++++++++++++++++++++++++++++++++++++ src/broker/broker_federation.h | 6 +++ src/broker/broker_linux.c | 53 +++++++++++++++++++++++ src/broker/broker_os.h | 15 ++++++ src/include/config.h.in | 2 + 15 files changed, 319 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"
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 cb47956..58ba8a2 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -161,6 +161,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..ffb2587 --- /dev/null +++ b/src/broker/broker.c @@ -0,0 +1,74 @@ +#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 *end; + char *port_env = getenv("MATAHARI_PORT"); + if (port_env) { + unsigned long port_l = strtoul(port_env, &end, 10); + if (port_env != end && port_l < (1 << 16)) { + port = port_l; + } + } + } + + 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..4c7340d --- /dev/null +++ b/src/broker/broker.h @@ -0,0 +1,12 @@ +#ifndef MH_BROKER_H +#define MH_BROKER_H + +#include <stdint.h> + + +#define MATAHARI_PORT 49000 + + +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..ae8dccf --- /dev/null +++ b/src/broker/broker_federation.c @@ -0,0 +1,93 @@ +#include "config.h" + +#include <stdlib.h> +#include <string.h> +#include <stdio.h> + +#include <matahari/dnssrv.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 < (sizeof(routes) / sizeof(*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..12cfe61 --- /dev/null +++ b/src/broker/broker_federation.h @@ -0,0 +1,6 @@ +#ifndef MH_BROKER_FEDERATION_H +#define MH_BROKER_FEDERATION_H + +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..54e3bff --- /dev/null +++ b/src/broker/broker_linux.c @@ -0,0 +1,53 @@ +#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..a961625 --- /dev/null +++ b/src/broker/broker_os.h @@ -0,0 +1,15 @@ +#ifndef MH_OS_BROKER_H +#define MH_OS_BROKER_H + +struct mh_qpid_route { + const char *dest; + const char *src; + const char *exchange; +}; + + +int broker_os_start_broker(char * const args[]); + +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..64a64db 100644 --- a/src/include/config.h.in +++ b/src/include/config.h.in @@ -5,3 +5,5 @@ #cmakedefine HAVE_TIME 1 #cmakedefine HAVE_G_LIST_FREE_FULL 1 #cmakedefine HAVE_PK_GET_SYNC 1 + +#define LOCAL_STATE_DIR "@localstatedir@"
On 09/05/2011 01:26 PM, 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 | 74 ++++++++++++++++++++++++++++++++ src/broker/broker.h | 12 +++++ src/broker/broker_federation.c | 93 ++++++++++++++++++++++++++++++++++++++++ src/broker/broker_federation.h | 6 +++ src/broker/broker_linux.c | 53 +++++++++++++++++++++++ src/broker/broker_os.h | 15 ++++++ src/include/config.h.in | 2 + 15 files changed, 319 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"
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 cb47956..58ba8a2 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -161,6 +161,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..ffb2587 --- /dev/null +++ b/src/broker/broker.c @@ -0,0 +1,74 @@ +#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 *end;
char *port_env = getenv("MATAHARI_PORT");
if (port_env) {
unsigned long port_l = strtoul(port_env,&end, 10);
if (port_env != end&& port_l< (1<< 16)) {
port = port_l;
}
Perhaps sscanf would be a little simpler?
if (sscanf(port_env, "%hu", %port) == -1) { /* invalid input */ port = 0; }
}
- }
- 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++]);
- }
Do you really want to copy over argv[0] ?
- 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..4c7340d --- /dev/null +++ b/src/broker/broker.h @@ -0,0 +1,12 @@ +#ifndef MH_BROKER_H +#define MH_BROKER_H
+#include<stdint.h>
+#define MATAHARI_PORT 49000
I'm not sure why it would ever change, but right now we have MATAHARI_POT as a cmake variable.
+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..ae8dccf --- /dev/null +++ b/src/broker/broker_federation.c @@ -0,0 +1,93 @@ +#include "config.h"
+#include<stdlib.h> +#include<string.h> +#include<stdio.h>
+#include<matahari/dnssrv.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< (sizeof(routes) / sizeof(*routes)); i++) {
There is a macro you can use to get the array length in utilities.h (I think) - DIMOF()
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;
- }
I don't know if our DNS SRV code suppots IPv6, but it will need to eventually. If the address that comes back is an IPv6 address, I suspect the host part will need to be in square brackets. It would be good to check qpidd for the proper syntax, though.
- 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);
I'm not sure how this should work once the SRV lookup API provides multiple records back. The only thing that makes sense to me is to continue doing what is being doing here, which is to use one record.
Ideally for the broker, we shouldn't be doing SRV lookups at all. qpidd should be handling it, because it should iterate over the results if connecting to one fails. Does qpidd have any SRV support that we may be able to take advantage of?
}
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..12cfe61 --- /dev/null +++ b/src/broker/broker_federation.h @@ -0,0 +1,6 @@
Add copyright headers to the top of new files.
+#ifndef MH_BROKER_FEDERATION_H +#define MH_BROKER_FEDERATION_H
+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..54e3bff --- /dev/null +++ b/src/broker/broker_linux.c @@ -0,0 +1,53 @@ +#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..a961625 --- /dev/null +++ b/src/broker/broker_os.h @@ -0,0 +1,15 @@ +#ifndef MH_OS_BROKER_H +#define MH_OS_BROKER_H
+struct mh_qpid_route {
- const char *dest;
- const char *src;
- const char *exchange;
+};
Doxygen comments in all the headers would be nice. :-)
+int broker_os_start_broker(char * const args[]);
+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..64a64db 100644 --- a/src/include/config.h.in +++ b/src/include/config.h.in @@ -5,3 +5,5 @@ #cmakedefine HAVE_TIME 1 #cmakedefine HAVE_G_LIST_FREE_FULL 1 #cmakedefine HAVE_PK_GET_SYNC 1
+#define LOCAL_STATE_DIR "@localstatedir@"
Matahari mailing list Matahari@lists.fedorahosted.org https://fedorahosted.org/mailman/listinfo/matahari
On 06/09/11 16:14, Russell Bryant wrote:
On 09/05/2011 01:26 PM, 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 | 74 ++++++++++++++++++++++++++++++++ src/broker/broker.h | 12 +++++ src/broker/broker_federation.c | 93 ++++++++++++++++++++++++++++++++++++++++ src/broker/broker_federation.h | 6 +++ src/broker/broker_linux.c | 53 +++++++++++++++++++++++ src/broker/broker_os.h | 15 ++++++ src/include/config.h.in | 2 + 15 files changed, 319 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"
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 cb47956..58ba8a2 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -161,6 +161,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..ffb2587 --- /dev/null +++ b/src/broker/broker.c @@ -0,0 +1,74 @@ +#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 *end;
char *port_env = getenv("MATAHARI_PORT");
if (port_env) {
unsigned long port_l = strtoul(port_env,&end, 10);
if (port_env != end&& port_l< (1<< 16)) {
port = port_l;
}
Perhaps sscanf would be a little simpler?
if (sscanf(port_env, "%hu", %port) == -1) { /* invalid input */ port = 0; }
Ahem, yeah. Thanks :)
}
- }
- 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++]);
- }
Do you really want to copy over argv[0] ?
As discussed, yes.
- 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..4c7340d --- /dev/null +++ b/src/broker/broker.h @@ -0,0 +1,12 @@ +#ifndef MH_BROKER_H +#define MH_BROKER_H
+#include<stdint.h>
+#define MATAHARI_PORT 49000
I'm not sure why it would ever change, but right now we have MATAHARI_POT as a cmake variable.
OK
+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..ae8dccf --- /dev/null +++ b/src/broker/broker_federation.c @@ -0,0 +1,93 @@ +#include "config.h"
+#include<stdlib.h> +#include<string.h> +#include<stdio.h>
+#include<matahari/dnssrv.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< (sizeof(routes) / sizeof(*routes)); i++) {
There is a macro you can use to get the array length in utilities.h (I think) - DIMOF()
OK (that's a _weird_ definition for it, isn't it?)
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;
- }
I don't know if our DNS SRV code suppots IPv6, but it will need to eventually. If the address that comes back is an IPv6 address, I suspect the host part will need to be in square brackets. It would be good to check qpidd for the proper syntax, though.
I think the DNS SRV code ought to handle this if it is required.
- 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);
I'm not sure how this should work once the SRV lookup API provides multiple records back. The only thing that makes sense to me is to continue doing what is being doing here, which is to use one record.
Ideally for the broker, we shouldn't be doing SRV lookups at all. qpidd should be handling it, because it should iterate over the results if connecting to one fails. Does qpidd have any SRV support that we may be able to take advantage of?
It's qpid-route that needs to handle it, and give that we have to call it 6 times with the same value each time, I don't think there's even a good way for it to do that. I agree that we seem to be stuck with what is happening here.
}
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..12cfe61 --- /dev/null +++ b/src/broker/broker_federation.h @@ -0,0 +1,6 @@
Add copyright headers to the top of new files.
Yeah, just remembered that right after I emailed the patch ;)
+#ifndef MH_BROKER_FEDERATION_H +#define MH_BROKER_FEDERATION_H
+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..54e3bff --- /dev/null +++ b/src/broker/broker_linux.c @@ -0,0 +1,53 @@ +#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..a961625 --- /dev/null +++ b/src/broker/broker_os.h @@ -0,0 +1,15 @@ +#ifndef MH_OS_BROKER_H +#define MH_OS_BROKER_H
+struct mh_qpid_route {
- const char *dest;
- const char *src;
- const char *exchange;
+};
Doxygen comments in all the headers would be nice. :-)
Done.
+int broker_os_start_broker(char * const args[]);
+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..64a64db 100644 --- a/src/include/config.h.in +++ b/src/include/config.h.in @@ -5,3 +5,5 @@ #cmakedefine HAVE_TIME 1 #cmakedefine HAVE_G_LIST_FREE_FULL 1 #cmakedefine HAVE_PK_GET_SYNC 1
+#define LOCAL_STATE_DIR "@localstatedir@"
Matahari mailing list Matahari@lists.fedorahosted.org https://fedorahosted.org/mailman/listinfo/matahari
On 09/06/2011 11:00 AM, Zane Bitter wrote:
On 06/09/11 16:14, Russell Bryant wrote:
On 09/05/2011 01:26 PM, Zane Bitter wrote:
+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;
- }
I don't know if our DNS SRV code suppots IPv6, but it will need to eventually. If the address that comes back is an IPv6 address, I suspect the host part will need to be in square brackets. It would be good to check qpidd for the proper syntax, though.
I think the DNS SRV code ought to handle this if it is required.
Actually, my comments don't really make sense since it's going to return a hostname, not an IP address.
However, I just noticed something else. SRV records include the port number you're supposed to use. The API doesn't give it to you, though. I'll address this when I work on this code later today.
matahari@lists.fedorahosted.org