Make sure that DAEMON_EVENT_SHUTDOWN gets transmitted before close() (hacky, but works).

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>

Makefile

@@ -69,7 +69,7 @@ RM = rm -f
 
 all: help nDPId nDPIsrvd
 
-examples: examples/c-json-stdout/c-json-stdout examples/go-dashboard/go-dashboard
+examples: examples/c-captured/c-captured examples/c-json-stdout/c-json-stdout examples/go-dashboard/go-dashboard
 
 nDPId: nDPId.c utils.c
 	$(CC) $(PROJECT_CFLAGS) $(CFLAGS) $(PC_CFLAGS) $^ -o $@ $(LDFLAGS) $(PC_LDFLAGS) $(STATIC_NDPI_LIB) $(LIBS)
@@ -77,6 +77,9 @@ nDPId: nDPId.c utils.c
 nDPIsrvd: nDPIsrvd.c utils.c
 	$(CC) $(PROJECT_CFLAGS) $(CFLAGS) $^ -o $@ $(LDFLAGS) $(STATIC_NDPI_LIB) $(LIBS)
 
+examples/c-captured/c-captured: examples/c-captured/c-captured.c
+	$(CC) $(PROJECT_CFLAGS) $(CFLAGS) $(JSMN_CFLAGS) $@.c -o $@ $(LDFLAGS) $(LIBS)
+
 examples/c-json-stdout/c-json-stdout: examples/c-json-stdout/c-json-stdout.c
 	$(CC) $(PROJECT_CFLAGS) $(CFLAGS) $(JSMN_CFLAGS) $@.c -o $@ $(LDFLAGS) $(LIBS)
 

contrib/nDPIsrvd.h

@@ -0,0 +1,18 @@
+#ifndef NDPISRVD_H
+#define NDPISRVD_H 1
+
+struct nDPIsrvd_socket
+{
+    int fd;
+    union {
+        struct {
+            char const * dst_ip;
+            unsigned short dst_port;
+        } ip;
+        struct {
+            char * path;
+        } unix;
+    } address;
+};
+
+#endif

examples/c-captured/c-captured.c

@@ -0,0 +1,134 @@
+#include <arpa/inet.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "config.h"
+#include "jsmn.h"
+
+static char serv_listen_addr[INET_ADDRSTRLEN] = DISTRIBUTOR_HOST;
+static uint16_t serv_listen_port = DISTRIBUTOR_PORT;
+
+int main(void)
+{
+    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
+    struct sockaddr_in remote_addr = {};
+    socklen_t remote_addrlen = sizeof(remote_addr);
+    uint8_t buf[NETWORK_BUFFER_MAX_SIZE];
+    size_t buf_used = 0;
+    size_t json_start = 0;
+    unsigned long long int json_bytes = 0;
+    jsmn_parser parser;
+    jsmntok_t tokens[128];
+
+    if (sockfd < 0)
+    {
+        perror("socket");
+        return 1;
+    }
+
+    remote_addr.sin_family = AF_INET;
+    if (inet_pton(AF_INET, &serv_listen_addr[0], &remote_addr.sin_addr) != 1)
+    {
+        perror("inet_pton");
+        return 1;
+    }
+    remote_addr.sin_port = htons(serv_listen_port);
+
+    if (connect(sockfd, (struct sockaddr *)&remote_addr, remote_addrlen) != 0)
+    {
+        perror("connect");
+        return 1;
+    }
+
+    while (1)
+    {
+        errno = 0;
+        ssize_t bytes_read = read(sockfd, buf + buf_used, sizeof(buf) - buf_used);
+
+        if (bytes_read <= 0 || errno != 0)
+        {
+            fprintf(stderr, "Remote end disconnected.\n");
+            break;
+        }
+
+        buf_used += bytes_read;
+        while (buf_used >= nDPIsrvd_JSON_BYTES + 1)
+        {
+            if (buf[nDPIsrvd_JSON_BYTES] != '{')
+            {
+                fprintf(stderr, "BUG: JSON invalid opening character: '%c'\n", buf[nDPIsrvd_JSON_BYTES]);
+                exit(1);
+            }
+
+            char * json_str_start = NULL;
+            json_bytes = strtoull((char *)buf, &json_str_start, 10);
+            json_bytes += (uint8_t *)json_str_start - buf;
+            json_start = (uint8_t *)json_str_start - buf;
+
+            if (errno == ERANGE)
+            {
+                fprintf(stderr, "BUG: Size of JSON exceeds limit\n");
+                exit(1);
+            }
+            if ((uint8_t *)json_str_start == buf)
+            {
+                fprintf(stderr, "BUG: Missing size before JSON string: \"%.*s\"\n", nDPIsrvd_JSON_BYTES, buf);
+                exit(1);
+            }
+            if (json_bytes > sizeof(buf))
+            {
+                fprintf(stderr, "BUG: JSON string too big: %llu > %zu\n", json_bytes, sizeof(buf));
+                exit(1);
+            }
+            if (json_bytes > buf_used)
+            {
+                break;
+            }
+
+            if (buf[json_bytes - 1] != '}')
+            {
+                fprintf(stderr, "BUG: Invalid JSON string: %.*s\n", (int)json_bytes, buf);
+                exit(1);
+            }
+
+            int r;
+            jsmn_init(&parser);
+            r = jsmn_parse(&parser,
+                           (char *)(buf + json_start),
+                           json_bytes - json_start,
+                           tokens,
+                           sizeof(tokens) / sizeof(tokens[0]));
+            if (r < 0 || tokens[0].type != JSMN_OBJECT)
+            {
+                fprintf(stderr, "JSON parsing failed with return value %d at position %u\n", r, parser.pos);
+                fprintf(stderr, "JSON string: '%.*s'\n", (int)(json_bytes - json_start), (char *)(buf + json_start));
+                exit(1);
+            }
+
+            for (int i = 1; i < r; i++)
+            {
+                if (i % 2 == 1)
+                {
+                    printf("[%.*s : ", tokens[i].end - tokens[i].start, (char *)(buf + json_start) + tokens[i].start);
+                }
+                else
+                {
+                    printf("%.*s] ", tokens[i].end - tokens[i].start, (char *)(buf + json_start) + tokens[i].start);
+                }
+            }
+            printf("EoF\n");
+
+            memmove(buf, buf + json_bytes, buf_used - json_bytes);
+            buf_used -= json_bytes;
+            json_bytes = 0;
+            json_start = 0;
+        }
+    }
+
+    return 0;
+}

nDPId.c

@@ -736,6 +736,7 @@ static void jsonize_daemon(struct nDPId_reader_thread * const reader_thread, enu
                                     "max-packets-per-flow-to-send",
                                     max_packets_per_flow_to_send);
     }
+    serialize_and_send(reader_thread);
 }
 
 static void jsonize_flow(struct nDPId_workflow * const workflow, struct nDPId_flow_info const * const flow)
@@ -1038,10 +1039,8 @@ static void jsonize_flow_event(struct nDPId_reader_thread * const reader_thread,
 
         case FLOW_EVENT_NOT_DETECTED:
         case FLOW_EVENT_GUESSED:
-            if (ndpi_dpi2json(workflow->ndpi_struct,
-                              flow->ndpi_flow,
-                              flow->guessed_l7_protocol,
-                              &workflow->ndpi_serializer) != 0)
+            if (ndpi_dpi2json(
+                    workflow->ndpi_struct, flow->ndpi_flow, flow->guessed_l7_protocol, &workflow->ndpi_serializer) != 0)
             {
                 syslog(LOG_DAEMON | LOG_ERR,
                        "[%8llu, %4u] ndpi_dpi2json failed for not-detected/guessed flow",
@@ -1366,7 +1365,8 @@ static void ndpi_process_packet(uint8_t * const args,
     /* process datalink layer */
     switch (pcap_datalink(workflow->pcap_handle))
     {
-        case DLT_NULL: {
+        case DLT_NULL:
+        {
             uint32_t dlt_hdr = ntohl(*((uint32_t *)&packet[eth_offset]));
 
             if (dlt_hdr == 0x00000002)
@@ -1919,7 +1919,6 @@ static void * processing_thread(void * const ndpi_thread_arg)
 
     run_pcap_loop(reader_thread);
     fcntl(reader_thread->json_sockfd, F_SETFL, fcntl(reader_thread->json_sockfd, F_GETFL, 0) & ~O_NONBLOCK);
-    jsonize_daemon(reader_thread, DAEMON_EVENT_SHUTDOWN);
     reader_thread->workflow->error_or_eof = 1;
     return NULL;
 }
@@ -2059,6 +2058,11 @@ static int stop_reader_threads(void)
             process_idle_flow(&reader_threads[i], idle_scan_index);
         }
 
+        jsonize_daemon(&reader_threads[i], DAEMON_EVENT_SHUTDOWN);
+        fsync(reader_threads[i].json_sockfd);
+        struct timespec ts = {.tv_sec = 0, .tv_nsec = 50000};
+        nanosleep(&ts, NULL); // ugly; make sure that DAEMON_EVENT_SHUTDOWN gets transmitted before close()
+
         close(reader_threads[i].json_sockfd);
         reader_threads[i].json_sockfd = -1;
     }