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 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 | 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(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"
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(a)lists.fedorahosted.org
https://fedorahosted.org/mailman/listinfo/matahari
--
Russell Bryant