Merge pull request #29 from lukbaj/ft-zmq-input

Add ZeroMQ input

Author: kristrev

CMakeLists.txt

@@ -12,6 +12,7 @@ set(SOURCE
     metadata_utils.c
     system_helpers.c
     metadata_input_netlink.c
+    metadata_input_nl_zmq_common.c
     lib/minmea.c)
 
 
@@ -57,13 +58,20 @@ if (NEAT)
     add_definitions("-DNEAT_SUPPORT")
 endif()
 
-if (ZEROMQ)
+if (ZEROMQ_WRITER)
     set(LIBS ${LIBS} zmq)
     set(SOURCE ${SOURCE}
         metadata_writer_zeromq.c
         metadata_writer_zeromq_monroe.c
         metadata_writer_zeromq_nne.c)
-    add_definitions("-DZEROMQ_SUPPORT")
+    add_definitions("-DZEROMQ_SUPPORT_WRITER")
+endif()
+
+if (ZEROMQ_INPUT)
+    set(LIBS ${LIBS} zmq)
+    set(SOURCE ${SOURCE}
+        metadata_input_zeromq.c)
+    add_definitions("-DZEROMQ_SUPPORT_INPUT")
 endif()
 
 if (GPSD)

metadata_exporter.c

@@ -44,6 +44,9 @@
 #ifdef GPSD_SUPPORT
     #include "metadata_input_gpsd.h"
 #endif
+#ifdef ZEROMQ_SUPPORT_INPUT
+    #include "metadata_input_zeromq.h"
+#endif
 #ifdef MUNIN_SUPPORT
     #include "metadata_input_munin.h"
 #endif
@@ -58,7 +61,7 @@
 #ifdef SQLITE_SUPPORT
     #include "metadata_writer_sqlite.h"
 #endif
-#ifdef ZEROMQ_SUPPORT
+#ifdef ZEROMQ_SUPPORT_WRITER
     #include "metadata_writer_zeromq.h"
 #endif
 #ifdef NNE_SUPPORT
@@ -258,16 +261,16 @@ static struct json_object *create_fake_restart_obj()
 static struct json_object *create_fake_conn_obj(uint64_t l3_id, uint64_t l4_id,
         uint8_t event_param, char *event_value_str, uint64_t tstamp)
 {
-	struct json_object *obj = NULL, *obj_add = NULL;
+    struct json_object *obj = NULL, *obj_add = NULL;
     uint8_t rand_value = 0;
     uint64_t rand_value_64 = 0;
     struct timeval tv;
 
-	if (!(obj = json_object_new_object()))
-		return NULL;
+    if (!(obj = json_object_new_object()))
+        return NULL;
 
     gettimeofday(&tv, NULL);
-	if (!(obj_add = json_object_new_int64(tv.tv_sec))) {
+    if (!(obj_add = json_object_new_int64(tv.tv_sec))) {
         json_object_put(obj);
         return NULL;
     }
@@ -354,7 +357,6 @@ static struct json_object *create_fake_conn_obj(uint64_t l3_id, uint64_t l4_id,
     }
     json_object_object_add(obj, "interface_id", obj_add);
 
-    
     if (!(obj_add = json_object_new_string("1234567"))) {
         json_object_put(obj);
         return NULL;
@@ -578,7 +580,7 @@ static void test_netlink(uint32_t packets)
     //
     //When testing, there is no need to multicast. We can just send to the PID
     netlink_addr.nl_pid = getpid();
-    
+
     srand(time(NULL));
 
     //TODO: Specify number of packets from command line
@@ -684,8 +686,8 @@ static void print_usage()
 #ifdef SQLITE_SUPPORT
     md_sqlite_usage();
 #endif
-#ifdef ZEROMQ_SUPPORT
-    md_zeromq_usage();
+#ifdef ZEROMQ_SUPPORT_WRITER
+    md_zeromq_writer_usage();
 #endif
 }
 
@@ -784,6 +786,19 @@ int main(int argc, char *argv[])
             num_inputs++;
         }
 #endif
+#ifdef ZEROMQ_SUPPORT_INPUT
+        else if (!strcmp(key, "zmq_input")) {
+            mde->md_inputs[MD_INPUT_ZEROMQ] = calloc(sizeof(struct md_input_zeromq), 1);
+
+            if (mde->md_inputs[MD_INPUT_ZEROMQ] == NULL) {
+                META_PRINT_SYSLOG(mde, LOG_ERR, "Could not allocate ZeroMQ input\n");
+                exit(EXIT_FAILURE);
+            }
+
+            md_zeromq_input_setup(mde, (struct md_input_zeromq*) mde->md_inputs[MD_INPUT_ZEROMQ]);
+            num_inputs++;
+        }
+#endif
 #ifdef NNE_SUPPORT
         else if (!strcmp(key, "nne")) {
             mde->md_writers[MD_WRITER_NNE] = calloc(sizeof(struct md_writer_nne), 1);
@@ -862,7 +877,7 @@ int main(int argc, char *argv[])
             num_writers++;
         }
 #endif
-#ifdef ZEROMQ_SUPPORT
+#ifdef ZEROMQ_SUPPORT_WRITER
         else if (!strcmp(key, "zmq")) {
             mde->md_writers[MD_WRITER_ZEROMQ] = calloc(sizeof(struct md_writer_zeromq), 1);
 
@@ -871,7 +886,7 @@ int main(int argc, char *argv[])
                 exit(EXIT_FAILURE);
             }
 
-            md_zeromq_setup(mde, (struct md_writer_zeromq*) mde->md_writers[MD_WRITER_ZEROMQ]);
+            md_zeromq_writer_setup(mde, (struct md_writer_zeromq*) mde->md_writers[MD_WRITER_ZEROMQ]);
             num_writers++;
         }
 #endif

metadata_exporter.h

@@ -111,6 +111,7 @@ enum md_inputs {
     MD_INPUT_GPS_NSB,
     MD_INPUT_MUNIN,
     MD_INPUT_SYSEVENT,
+    MD_INPUT_ZEROMQ,
     __MD_INPUT_MAX
 };