forwarder device added

Makefile.am

@@ -14,8 +14,8 @@ if BUILD_PERF
 DIR_PERF = perf
 endif
 
-SUBDIRS = src $(DIR_P) $(DIR_R) $(DIR_J) $(DIR_PERF)
-DIST_SUBDIRS = src python ruby java perf
+SUBDIRS = src $(DIR_P) $(DIR_R) $(DIR_J) $(DIR_PERF) devices
+DIST_SUBDIRS = src python ruby java perf devices
 
 EXTRA_DIST =  $(top_srcdir)/foreign/openpgm/@pgm_basename@.tar.bz2
 

configure.in

@@ -441,6 +441,16 @@ if test "x$with_pgm_ext" != "xno"; then
     pgm_ext="yes"
 fi
 
+# forwarder device
+forwarder="no"
+AC_ARG_WITH([forwarder], [AS_HELP_STRING([--with-forwarder], 
+    [build forwarder device [default=no]])], [with_forwarder=yes], [with_forwarder=no])
+
+if test "x$with_forwarder" != "xno"; then
+    forwarder="yes"
+fi
+
+
 # Perf
 perf="no"
 AC_ARG_WITH([perf], [AS_HELP_STRING([--with-perf], 
@@ -462,6 +472,7 @@ AM_CONDITIONAL(BUILD_RUBY, test "x$rbzmq" = "xyes")
 AM_CONDITIONAL(BUILD_C, test "x$czmq" = "xyes")
 AM_CONDITIONAL(BUILD_CPP, test "x$cppzmq" = "xyes")
 AM_CONDITIONAL(BUILD_PGM, test "x$pgm_ext" = "xyes")
+AM_CONDITIONAL(BUILD_FORWARDER, test "x$forwarder" = "xyes") 
 AM_CONDITIONAL(BUILD_PERF, test "x$perf" = "xyes") 
 
 AC_SUBST(stdint)
@@ -479,7 +490,8 @@ AC_CHECK_FUNCS(perror gettimeofday memset socket getifaddrs freeifaddrs)
 
 AC_OUTPUT(Makefile src/Makefile python/Makefile python/setup.py ruby/Makefile \
     java/Makefile perf/Makefile perf/c/Makefile perf/cpp/Makefile \
-    perf/python/Makefile perf/ruby/Makefile perf/java/Makefile src/libzmq.pc)
+    perf/python/Makefile perf/ruby/Makefile perf/java/Makefile src/libzmq.pc \
+    devices/Makefile devices/forwarder/Makefile)
 
 AC_MSG_RESULT([])
 AC_MSG_RESULT([ ******************************************************** ])
@@ -492,19 +504,26 @@ AC_MSG_RESULT([  license text. ])
 AC_MSG_RESULT([ ******************************************************** ])
 AC_MSG_RESULT([])
 AC_MSG_RESULT([ 0MQ install dir: $prefix])
-AC_MSG_RESULT([   C language binding: $czmq])
-AC_MSG_RESULT([   C++ language binding: $cppzmq])
-AC_MSG_RESULT([   Python language binding: $pyzmq])
-AC_MSG_RESULT([   Ruby language binding: $rbzmq])
+AC_MSG_RESULT([   Language bindings:])
+AC_MSG_RESULT([     C: $czmq])
+AC_MSG_RESULT([     C++: $cppzmq])
+AC_MSG_RESULT([     Java: $jzmq])
+AC_MSG_RESULT([     Python: $pyzmq])
+AC_MSG_RESULT([     Ruby: $rbzmq])
 if test "x$rbzmq" = "xyes"; then
-AC_MSG_RESULT([   Ruby library install dir: $rubydir])
+AC_MSG_RESULT([       Ruby library install dir: $rubydir])
 fi
-AC_MSG_RESULT([   Java language binding: $jzmq])
+AC_MSG_RESULT([   Network protocols:])
+AC_MSG_RESULT([     TCP: yes])
 if test "x$pgm_ext" = "xyes"; then
-AC_MSG_RESULT([   PGM extension: $pgm_ext ($pgm_name)])
+AC_MSG_RESULT([     PGM: $pgm_ext ($pgm_name)])
 else
-AC_MSG_RESULT([   PGM extension: $pgm_ext])
+AC_MSG_RESULT([     PGM: $pgm_ext])
 fi
-AC_MSG_RESULT([   performance tests: $perf]) 
+AC_MSG_RESULT([   Devices:])
+AC_MSG_RESULT([     forwarder: $forwarder])
+AC_MSG_RESULT([   Performance tests: $perf]) 
+AC_MSG_RESULT([])
+AC_MSG_RESULT([ ******************************************************** ])
 AC_MSG_RESULT([])
 

devices/Makefile.am

+if BUILD_FORWARDER
+FORWARDER_DIR = forwarder
+endif
+
+SUBDIRS = $(FORWARDER_DIR)
+DIST_SUBDIRS = forwarder

devices/forwarder/Makefile.am

+INCLUDES = -I$(top_builddir)/c -I$(top_builddir)/cpp
+
+bin_PROGRAMS = forwarder
+
+forwarder_LDADD = $(top_builddir)/src/libzmq.la
+forwarder_SOURCES = forwarder.cpp
+forwarder_CXXFLAGS = -Wall -pedantic -Werror
+
+

devices/forwarder/forwarder.cpp

+/*
+    Copyright (c) 2007-2009 FastMQ Inc.
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the Lesser GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    0MQ 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
+    Lesser GNU General Public License for more details.
+
+    You should have received a copy of the Lesser GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "../../cpp/zmq.hpp"
+#include "../../foreign/xmlParser/xmlParser.cpp"
+
+int main (int argc, char *argv [])
+{
+    if (argc != 2) {
+        fprintf (stderr, "usage: forwarder <config-file>\n");
+        return 1;
+    }
+
+    XMLNode root = XMLNode::parseFile (argv [1]);
+    if (root.isEmpty ()) {
+        fprintf (stderr, "configuration file not found\n");
+        return 1;
+    }
+
+    if (strcmp (root.getName (), "forwarder") != 0) {
+        fprintf (stderr, "root element in the configuration file should be "
+            "named 'forwarder'\n");
+        return 1;
+    }
+
+    XMLNode in_node = root.getChildNode ("in");
+    if (in_node.isEmpty ()) {
+        fprintf (stderr, "'in' node is missing in the configuration file\n");
+        return 1;
+    }
+
+    XMLNode out_node = root.getChildNode ("out");
+    if (out_node.isEmpty ()) {
+        fprintf (stderr, "'out' node is missing in the configuration file\n");
+        return 1;
+    }
+
+    //  TODO: make the number of I/O threads configurable.
+    zmq::context_t ctx (1, 1);
+    zmq::socket_t in_socket (ctx, ZMQ_P2P);
+    zmq::socket_t out_socket (ctx, ZMQ_P2P);
+
+    int n = 0;
+    while (true) {
+        XMLNode bind = in_node.getChildNode ("bind", n);
+        if (bind.isEmpty ())
+            break;
+        const char *addr = bind.getAttribute ("addr");
+        if (!addr) {
+            fprintf (stderr, "'bind' node is missing 'addr' attribute\n");
+            return 1;
+        }
+        in_socket.bind (addr);
+        n++;
+    }
+
+    n = 0;
+    while (true) {
+        XMLNode connect = in_node.getChildNode ("connect", n);
+        if (connect.isEmpty ())
+            break;
+        const char *addr = connect.getAttribute ("addr");
+        if (!addr) {
+            fprintf (stderr, "'connect' node is missing 'addr' attribute\n");
+            return 1;
+        }
+        in_socket.connect (addr);
+        n++;
+    }
+
+    n = 0;
+    while (true) {
+        XMLNode bind = out_node.getChildNode ("bind", n);
+        if (bind.isEmpty ())
+            break;
+        const char *addr = bind.getAttribute ("addr");
+        if (!addr) {
+            fprintf (stderr, "'bind' node is missing 'addr' attribute\n");
+            return 1;
+        }
+        out_socket.bind (addr);
+        n++;
+    }
+
+    n = 0;
+    while (true) {
+        XMLNode connect = out_node.getChildNode ("connect", n);
+        if (connect.isEmpty ())
+            break;
+        const char *addr = connect.getAttribute ("addr");
+        if (!addr) {
+            fprintf (stderr, "'connect' node is missing 'addr' attribute\n");
+            return 1;
+        }
+        out_socket.connect (addr);
+        n++;
+    }
+
+    zmq::message_t msg;
+    while (true) {
+        in_socket.recv (&msg);
+        out_socket.send (msg);
+    }
+
+    return 0;
+}