Aligned nDPIsrvd.py to nDPIsrvd.h for consistency.

 * Simplified Python interface as well.
 * c-captured and flow-undetected-to-pcap.py produce similiar results
 * Removed overloaded nDPIsrvd.py event structures.
 * flow-info.py prints (with a color-hash) additional information e.g. alias/source and midstream

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

5 files changed, 173 insertions, 177 deletions

diff --git a/examples/c-captured/c-captured.c b/examples/c-captured/c-captured.c
index d0e60521d..a192acb31 100644
--- a/examples/c-captured/c-captured.c
+++ b/examples/c-captured/c-captured.c
@@ -10,6 +10,7 @@
 #include <string.h>
 #include <sys/socket.h>
 #include <sys/types.h>
+#include <time.h>
 #include <unistd.h>
 
 #include "nDPIsrvd.h"
@@ -37,8 +38,6 @@ struct flow_user_data
     uint8_t detected;
     nDPIsrvd_ull flow_datalink;
     nDPIsrvd_ull flow_max_packets;
-    nDPIsrvd_ull flow_l4_header_len;
-    nDPIsrvd_ull flow_total_l4_payload_len;
     UT_array * packets;
 };
 
@@ -47,6 +46,11 @@ static int main_thread_shutdown = 0;
 static char const serv_listen_path[] = DISTRIBUTOR_UNIX_SOCKET;
 static char const serv_listen_addr[INET_ADDRSTRLEN] = DISTRIBUTOR_HOST;
 static uint16_t const serv_listen_port = DISTRIBUTOR_PORT;
+#ifdef pcap_dump_open_append
+static time_t pcap_filename_rotation = 600;
+static time_t pcap_filename_last_rotation = 0;
+static struct tm pcap_filename_last_rotation_tm = {};
+#endif
 
 static void packet_data_copy(void * dst, const void * src)
 {
@@ -82,10 +86,41 @@ static char * generate_pcap_filename(struct nDPIsrvd_flow const * const flow,
                                      char * const dest,
                                      size_t size)
 {
+    char appendix[32] = {};
+
+#ifdef pcap_dump_open_append
+    if (pcap_filename_rotation > 0)
+    {
+        time_t current_time = time(NULL);
+
+        if (current_time >= pcap_filename_last_rotation + pcap_filename_rotation)
+        {
+            pcap_filename_last_rotation = current_time;
+            if (localtime_r(&pcap_filename_last_rotation, &pcap_filename_last_rotation_tm) == NULL)
+            {
+                return NULL;
+            }
+        }
+
+        if (strftime(appendix, sizeof(appendix), "%d_%m_%y-%H_%M_%S", &pcap_filename_last_rotation_tm) == 0)
+        {
+            return NULL;
+        }
+    } else
+#endif
+    {
+        if (snprintf(appendix, sizeof(appendix), "%s", flow->id) <= 0)
+        {
+            return NULL;
+        }
+    }
+
     if (flow_user->guessed != 0 || flow_user->detected == 0)
     {
         int ret =
-            snprintf(dest, size, "flow-%s-%s.pcap", (flow_user->guessed != 0 ? "guessed" : "undetected"), flow->id);
+            snprintf(dest, size, "flow-%s-%s.pcap",
+                     (flow_user->guessed != 0 ? "guessed" : "undetected"),
+                     appendix);
         if (ret <= 0 || (size_t)ret > size)
         {
             return NULL;
@@ -119,7 +154,12 @@ static int packet_write_pcap_file(UT_array const * const pd_array, int pkt_datal
     {
         return 1;
     }
+
+#ifdef pcap_dump_open_append
+    pcap_dumper_t * pd = pcap_dump_open_append(p, filename);
+#else
     pcap_dumper_t * pd = pcap_dump_open(p, filename);
+#endif
     if (pd == NULL)
     {
         fprintf(stderr, "pcap error %s\n", pcap_geterr(p));
@@ -258,6 +298,9 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
         nDPIsrvd_ull pkt_l4_len = 0ull;
         perror_ull(TOKEN_VALUE_TO_ULL(TOKEN_GET_SZ(sock, "pkt_l4_len"), &pkt_l4_len), "pkt_l4_len");
 
+        nDPIsrvd_ull pkt_l4_offset = 0ull;
+        perror_ull(TOKEN_VALUE_TO_ULL(TOKEN_GET_SZ(sock, "pkt_l4_offset"), &pkt_l4_offset), "pkt_l4_offset");
+
         struct packet_data pd = {
             .packet_ts_sec = pkt_ts_sec,
             .packet_ts_usec = pkt_ts_usec,
@@ -266,31 +309,16 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
             .base64_packet_const = pkt->value
         };
         utarray_push_back(flow_user->packets, &pd);
-        flow_user->flow_total_l4_payload_len += pkt_l4_len - flow_user->flow_l4_header_len;
     }
 
     {
         struct nDPIsrvd_json_token const * const flow_event_name = TOKEN_GET_SZ(sock, "flow_event_name");
         if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "new") != 0)
         {
+            flow_user->flow_new_seen = 1;
             perror_ull(TOKEN_VALUE_TO_ULL(TOKEN_GET_SZ(sock, "flow_datalink"), &flow_user->flow_datalink), "flow_datalink");
             perror_ull(TOKEN_VALUE_TO_ULL(TOKEN_GET_SZ(sock, "flow_max_packets"), &flow_user->flow_max_packets), "flow_max_packets");
 
-            struct nDPIsrvd_json_token const * const l4_proto = TOKEN_GET_SZ(sock, "l4_proto");
-            if (TOKEN_VALUE_EQUALS_SZ(l4_proto, "tcp") != 0)
-            {
-                flow_user->flow_l4_header_len = sizeof(struct tcphdr);
-            } else if (TOKEN_VALUE_EQUALS_SZ(l4_proto, "udp") != 0)
-            {
-                flow_user->flow_l4_header_len = sizeof(struct udphdr);
-            } else if (TOKEN_VALUE_EQUALS_SZ(l4_proto, "icmp") != 0 ||
-                       TOKEN_VALUE_EQUALS_SZ(l4_proto, "icmp6") != 0)
-            {
-                flow_user->flow_l4_header_len = sizeof(struct icmphdr);
-            }
-
-            flow_user->flow_new_seen = 1;
-
             return CALLBACK_OK;
         } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "guessed") != 0)
         {
@@ -329,7 +357,6 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
             (flow_user->guessed != 0 || flow_user->detected == 0))
         {
             packet_data_print(flow_user->packets);
-            if (flow_user->flow_total_l4_payload_len > 0)
             {
                 char pcap_filename[64];
                 if (generate_pcap_filename(flow, flow_user, pcap_filename, sizeof(pcap_filename)) == NULL)
@@ -342,8 +369,6 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
                 {
                     return CALLBACK_ERROR;
                 }
-            } else {
-                printf("flow %s: captured packets do not have any l4 payload\n", flow->id);
             }
 
             utarray_free(flow_user->packets);
@@ -376,6 +401,31 @@ static void captured_flow_end_callback(struct nDPIsrvd_socket * const sock, stru
     }
 }
 
+// TODO: argv parsing
+#if 0
+static int parse_options(int argc, char ** argv)
+{
+    int opt;
+
+    static char const usage[] =
+        "Usage: %s "
+        "[-d] [-s host] [-R rotate-every-n-seconds] [-g] [-u]\n";
+
+    while ((opt = getopt(argc, argv, "hds:R:gu")) != -1)
+    {
+    }
+
+    if (optind < argc)
+    {
+        fprintf(stderr, "Unexpected argument after options\n\n");
+        fprintf(stderr, usage, argv[0]);
+        return 1;
+    }
+
+    return 0;
+}
+#endif
+
 int main(int argc, char ** argv)
 {
     sock = nDPIsrvd_init(0, sizeof(struct flow_user_data), captured_json_callback, captured_flow_end_callback);
diff --git a/examples/py-flow-info/flow-info.py b/examples/py-flow-info/flow-info.py
index c13e9cead..62e43f74b 100755
--- a/examples/py-flow-info/flow-info.py
+++ b/examples/py-flow-info/flow-info.py
@@ -7,34 +7,33 @@ sys.path.append(os.path.dirname(sys.argv[0]) + '/../../dependencies')
 import nDPIsrvd
 from nDPIsrvd import nDPIsrvdSocket, TermColor
 
-
-def parse_json_str(json_str):
-
-    j = nDPIsrvd.JsonParseBytes(json_str[0])
-    nDPIdEvent = nDPIsrvd.nDPIdEvent.validateJsonEventTypes(j)
-    if nDPIdEvent.isValid is False:
-        raise RuntimeError('Missing event id or event name invalid in the JSON string: {}'.format(j))
-    if nDPIdEvent.BasicEventID != -1:
-        print('{:>21}: {}'.format(TermColor.WARNING + TermColor.BLINK + 'BASIC-EVENT' + TermColor.END,
-                                  nDPIdEvent.BasicEventPrettyName))
-        return
-    elif nDPIdEvent.FlowEventID == -1:
-        return
+def prettifyEvent(color_list, whitespaces, text):
+    term_attrs = str()
+    for color in color_list:
+        term_attrs += str(color)
+    return '{}{:>' + str(whitespaces) + '}{}'.format(term_attrs, text, TermColor.END)
+
+def onJsonLineRecvd(json_dict, current_flow, global_user_data):
+    if 'basic_event_id' in json_dict:
+        print('{}: {}'.format(prettifyEvent([TermColor.WARNING, TermColor.BLINK], 16, 'BASIC-EVENT'), json_dict['basic_event_name']))
+        return True
+    elif 'flow_event_id' not in json_dict:
+        return True
 
     ndpi_proto_categ = ''
     ndpi_frisk = ''
 
-    if 'ndpi' in j:
-        if 'proto' in j['ndpi']:
-            ndpi_proto_categ += '[' + str(j['ndpi']['proto']) + ']'
+    if 'ndpi' in json_dict:
+        if 'proto' in json_dict['ndpi']:
+            ndpi_proto_categ += '[' + str(json_dict['ndpi']['proto']) + ']'
 
-        if 'category' in j['ndpi']:
-            ndpi_proto_categ += '[' + str(j['ndpi']['category']) + ']'
+        if 'category' in json_dict['ndpi']:
+            ndpi_proto_categ += '[' + str(json_dict['ndpi']['category']) + ']'
 
-        if 'flow_risk' in j['ndpi']:
+        if 'flow_risk' in json_dict['ndpi']:
             cnt = 0
-            for key in j['ndpi']['flow_risk']:
-                ndpi_frisk += str(j['ndpi']['flow_risk'][key]) + ', '
+            for key in json_dict['ndpi']['flow_risk']:
+                ndpi_frisk += str(json_dict['ndpi']['flow_risk'][key]) + ', '
                 cnt += 1
             ndpi_frisk = '{}: {}'.format(
                 TermColor.WARNING + TermColor.BOLD + 'RISK' + TermColor.END if cnt < 2
@@ -42,39 +41,43 @@ def parse_json_str(json_str):
                 ndpi_frisk[:-2])
 
     instance_and_source = ''
-    instance_and_source += '[{}]'.format(TermColor.setColorByString(j['alias']))
-    instance_and_source += '[{}]'.format(TermColor.setColorByString(j['source']))
+    instance_and_source += '[{}]'.format(TermColor.setColorByString(json_dict['alias']))
+    instance_and_source += '[{}]'.format(TermColor.setColorByString(json_dict['source']))
 
+    line_suffix = ''
     flow_event_name = ''
-    if nDPIdEvent.FlowEventName == 'guessed' or nDPIdEvent.FlowEventName == 'undetected':
-        flow_event_name += '{}{:>16}{}'.format(TermColor.HINT, nDPIdEvent.FlowEventPrettyName, TermColor.END)
+    if json_dict['flow_event_name'] == 'guessed' or json_dict['flow_event_name'] == 'not-detected':
+        flow_event_name += '{}{:>16}{}'.format(TermColor.HINT, json_dict['flow_event_name'], TermColor.END)
     else:
-        flow_event_name += '{:>16}'.format(nDPIdEvent.FlowEventPrettyName)
+        if json_dict['flow_event_name'] == 'new' and json_dict['midstream'] != 0:
+            line_suffix = '[{}]'.format(TermColor.WARNING + TermColor.BLINK + 'MIDSTREAM' + TermColor.END)
+        flow_event_name += '{:>16}'.format(json_dict['flow_event_name'])
 
-    if j['l3_proto'] == 'ip4':
-        print('{} {}: [{:.>6}] [{}][{:.>5}] [{:.>15}]{} -> [{:.>15}]{} {}' \
+    if json_dict['l3_proto'] == 'ip4':
+        print('{} {}: [{:.>6}] [{}][{:.>5}] [{:.>15}]{} -> [{:.>15}]{} {}{}' \
               ''.format(instance_and_source, flow_event_name, 
-              j['flow_id'], j['l3_proto'], j['l4_proto'],
-              j['src_ip'].lower(),
-              '[{:.>5}]'.format(j['src_port']) if 'src_port' in j else '',
-              j['dst_ip'].lower(),
-              '[{:.>5}]'.format(j['dst_port']) if 'dst_port' in j else '',
-              ndpi_proto_categ))
-    elif j['l3_proto'] == 'ip6':
-        print('{} {}: [{:.>6}] [{}][{:.>5}] [{:.>39}]{} -> [{:.>39}]{} {}' \
+              json_dict['flow_id'], json_dict['l3_proto'], json_dict['l4_proto'],
+              json_dict['src_ip'].lower(),
+              '[{:.>5}]'.format(json_dict['src_port']) if 'src_port' in json_dict else '',
+              json_dict['dst_ip'].lower(),
+              '[{:.>5}]'.format(json_dict['dst_port']) if 'dst_port' in json_dict else '',
+              ndpi_proto_categ, line_suffix))
+    elif json_dict['l3_proto'] == 'ip6':
+        print('{} {}: [{:.>6}] [{}][{:.>5}] [{:.>39}]{} -> [{:.>39}]{} {}{}' \
                 ''.format(instance_and_source, flow_event_name,
-              j['flow_id'], j['l3_proto'], j['l4_proto'],
-              j['src_ip'].lower(),
-              '[{:.>5}]'.format(j['src_port']) if 'src_port' in j else '',
-              j['dst_ip'].lower(),
-              '[{:.>5}]'.format(j['dst_port']) if 'dst_port' in j else '',
-              ndpi_proto_categ))
+              json_dict['flow_id'], json_dict['l3_proto'], json_dict['l4_proto'],
+              json_dict['src_ip'].lower(),
+              '[{:.>5}]'.format(json_dict['src_port']) if 'src_port' in json_dict else '',
+              json_dict['dst_ip'].lower(),
+              '[{:.>5}]'.format(json_dict['dst_port']) if 'dst_port' in json_dict else '',
+              ndpi_proto_categ, line_suffix))
     else:
-        raise RuntimeError('unsupported l3 protocol: {}'.format(j['l3_proto']))
+        raise RuntimeError('unsupported l3 protocol: {}'.format(json_dict['l3_proto']))
 
     if len(ndpi_frisk) > 0:
         print('{} {:>18}{}'.format(instance_and_source, '', ndpi_frisk))
 
+    return True
 
 if __name__ == '__main__':
     argparser = nDPIsrvd.defaultArgumentParser()
@@ -86,9 +89,4 @@ if __name__ == '__main__':
 
     nsock = nDPIsrvdSocket()
     nsock.connect(address)
-
-    while True:
-        received = nsock.receive()
-        for received_json_pkt in received:
-            parse_json_str(received_json_pkt)
-
+    nsock.loop(onJsonLineRecvd, None)
diff --git a/examples/py-flow-undetected-to-pcap/flow-undetected-to-pcap.py b/examples/py-flow-undetected-to-pcap/flow-undetected-to-pcap.py
index f04e4d955..961adc3cf 100755
--- a/examples/py-flow-undetected-to-pcap/flow-undetected-to-pcap.py
+++ b/examples/py-flow-undetected-to-pcap/flow-undetected-to-pcap.py
@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 
-import base64
 import os
 import sys
 
@@ -8,56 +7,44 @@ sys.path.append(os.path.dirname(sys.argv[0]) + '/../../dependencies')
 import nDPIsrvd
 from nDPIsrvd import TermColor, nDPIsrvdSocket, PcapPacket
 
-FLOWS = dict()
+def onJsonLineRecvd(json_dict, current_flow, global_user_data):
+    if current_flow is None:
 
-def parse_json_str(json_str):
+        if 'packet_event_name' in json_dict and json_dict['packet_event_name'] == 'packet':
+            fake_flow = Flow()
+            fake_flow.pkt = PcapPacket()
+            PcapPacket.handleJSON(json_dict, fake_flow)
+            fake_flow.pkt.doDump()
+            fake_flow.pkt.setSuffix('packet_undetected')
+            fake_flow.pkt.fin()
 
-    j = nDPIsrvd.JsonParseBytes(json_str[0])
+        return True
 
-    global FLOWS
+    PcapPacket.handleJSON(json_dict, current_flow)
 
-    if 'flow_event_name' in j:
+    if 'flow_event_name' in json_dict and PcapPacket.isInitialized(current_flow) and \
+        (json_dict['flow_event_name'] == 'guessed' or json_dict['flow_event_name'] == 'not-detected'):
 
-        event = j['flow_event_name'].lower()
-        flow_id = j['flow_id']
+        current_flow.pcap_packet.doDump()
+        if json_dict['flow_event_name'] == 'guessed':
+            current_flow.pcap_packet.setSuffix('guessed')
 
-        if 'midstream' in j and j['midstream'] == 1:
-            return
+            try:
+                if current_flow.pcap_packet.fin() is True:
+                    print('Guessed flow with id {}, dumped'.format(current_flow.flow_id))
+            except RuntimeError as err:
+                print('Guessed flow with id {} excepted: {}'.format(current_flow.flow_id, str(err)))
 
-        if event == 'new':
-            FLOWS[flow_id] = PcapPacket(flow_id)
-        elif flow_id not in FLOWS:
-            return
-        elif event == 'end' or event == 'idle':
-            del FLOWS[flow_id]
-        elif event == 'detected' or event == 'detection-update':
-            FLOWS[flow_id].detected()
-        elif event == 'guessed' or event == 'not-detected':
-            if event == 'guessed':
-                print('Guessed flow with id {}, PCAP dump returned: {}'.format(flow_id, FLOWS[flow_id].fin('guessed')))
-            else:
-                print('Not-detected flow with id {}: PCAP dump returned {}'.format(flow_id, FLOWS[flow_id].fin('undetected')))
         else:
-            raise RuntimeError('unknown flow event name: {}'.format(event))
+            current_flow.pcap_packet.setSuffix('undetected')
 
-    elif 'packet_event_name' in j:
-
-        buffer_decoded = base64.b64decode(j['pkt'], validate=True)
-
-        if j['packet_event_name'] == 'packet-flow':
-
-            flow_id = j['flow_id']
-
-            if flow_id not in FLOWS:
-                return
-
-            FLOWS[flow_id].addPacket(buffer_decoded, j['pkt_type'], j['pkt_l3_offset'])
-
-        if j['packet_event_name'] == 'packet':
-
-            flow = PcapPacket()
-            flow.addPacket(buffer_decoded, j['pkt_type'], j['pkt_l3_offset'])
+            try:
+                if current_flow.pcap_packet.fin() is True:
+                    print('Not-detected flow with id {}, dumped'.format(current_flow.flow_id))
+            except RuntimeError as err:
+                print('Not-detected flow with id {} excepted: {}'.format(current_flow.flow_id, str(err)))
 
+    return True
 
 if __name__ == '__main__':
     argparser = nDPIsrvd.defaultArgumentParser()
@@ -69,9 +56,4 @@ if __name__ == '__main__':
 
     nsock = nDPIsrvdSocket()
     nsock.connect(address)
-
-    while True:
-        received = nsock.receive()
-        for received_json_pkt in received:
-            parse_json_str(received_json_pkt)
-
+    nsock.loop(onJsonLineRecvd, None)
diff --git a/examples/py-json-stdout/json-stdout.py b/examples/py-json-stdout/json-stdout.py
index ac94d707f..9f58d161e 100755
--- a/examples/py-json-stdout/json-stdout.py
+++ b/examples/py-json-stdout/json-stdout.py
@@ -8,13 +8,9 @@ import nDPIsrvd
 from nDPIsrvd import nDPIsrvdSocket, TermColor
 
 
-def parse_json_str(json_str):
-
-    j = nDPIsrvd.JsonParseBytes(json_str[0])
-    nDPIdEvent = nDPIsrvd.nDPIdEvent.validateJsonEventTypes(j)
-    if nDPIdEvent.isValid is False:
-        raise RuntimeError('Missing event id or event name invalid in the JSON string: {}'.format(j))
-    print(j)
+def onJsonLineRecvd(json_dict, current_flow, global_user_data):
+    print(json_dict)
+    return True
 
 if __name__ == '__main__':
     argparser = nDPIsrvd.defaultArgumentParser()
@@ -26,9 +22,4 @@ if __name__ == '__main__':
 
     nsock = nDPIsrvdSocket()
     nsock.connect(address)
-
-    while True:
-        received = nsock.receive()
-        for received_json_pkt in received:
-            parse_json_str(received_json_pkt)
-
+    nsock.loop(onJsonLineRecvd, None)
diff --git a/examples/py-risky-flow-to-pcap/risky-flow-to-pcap.py b/examples/py-risky-flow-to-pcap/risky-flow-to-pcap.py
index 1a07e2e90..33c0be810 100755
--- a/examples/py-risky-flow-to-pcap/risky-flow-to-pcap.py
+++ b/examples/py-risky-flow-to-pcap/risky-flow-to-pcap.py
@@ -8,52 +8,32 @@ sys.path.append(os.path.dirname(sys.argv[0]) + '/../../dependencies')
 import nDPIsrvd
 from nDPIsrvd import TermColor, nDPIsrvdSocket, PcapPacket
 
-FLOWS = dict()
+def onJsonLineRecvd(json_dict, current_flow, global_user_data):
+    if current_flow is None:
+        return True
 
-def parse_json_str(json_str):
+    PcapPacket.handleJSON(json_dict, current_flow)
 
-    j = nDPIsrvd.JsonParseBytes(json_str[0])
+    if 'flow_event_name' in json_dict and PcapPacket.isInitialized(current_flow) and \
+        'ndpi' in json_dict and 'flow_risk' in json_dict['ndpi'] and not hasattr(current_flow, 'is_risky_flow'):
 
-    global FLOWS
+        current_flow.pcap_packet.doDump()
+        current_flow.pcap_packet.setSuffix('risky')
+        current_flow.is_risky_flow = True
+        print('Risky flow with id {} marked for dumping.'.format(current_flow.flow_id))
 
-    if 'flow_event_name' in j:
+    if hasattr(current_flow, 'is_risky_flow') and \
+        (current_flow.pcap_packet.current_packet < current_flow.pcap_packet.max_packets or \
+         ('flow_event_name' in json_dict and \
+          (json_dict['flow_event_name'] == 'end' or json_dict['flow_event_name'] == 'idle'))):
 
-        event = j['flow_event_name'].lower()
-        flow_id = j['flow_id']
-
-        if 'midstream' in j and j['midstream'] == 1:
-            return
-
-        if event == 'new':
-            FLOWS[flow_id] = PcapPacket(flow_id)
-        elif flow_id not in FLOWS:
-            return
-        elif event == 'end' or event == 'idle':
-            del FLOWS[flow_id]
-        elif event == 'detected' or event == 'detection-update' or event == 'guessed' or event == 'not-detected':
-            if 'ndpi' in j and 'flow_risk' in j['ndpi']:
-                print('Risky flow with id {}, PCAP dump returned: {}'.format(flow_id, FLOWS[flow_id].fin('risky')))
-        else:
-            raise RuntimeError('unknown flow event name: {}'.format(event))
-
-    elif 'packet_event_name' in j:
-
-        buffer_decoded = base64.b64decode(j['pkt'], validate=True)
-
-        if j['packet_event_name'] == 'packet-flow':
-
-            flow_id = j['flow_id']
-
-            if flow_id not in FLOWS:
-                return
-
-            FLOWS[flow_id].addPacket(buffer_decoded, j['pkt_type'], j['pkt_l3_offset'])
-
-        if j['packet_event_name'] == 'packet':
-
-            flow = PcapPacket()
-            flow.addPacket(buffer_decoded, j['pkt_type'], j['pkt_l3_offset'])
+        try:
+            if current_flow.pcap_packet.fin() is True:
+                print('Risky flow with id {} dumped.'.format(current_flow.flow_id))
+        except RuntimeError as err:
+            pass
 
+    return True
 
 if __name__ == '__main__':
     argparser = nDPIsrvd.defaultArgumentParser()
@@ -65,9 +45,4 @@ if __name__ == '__main__':
 
     nsock = nDPIsrvdSocket()
     nsock.connect(address)
-
-    while True:
-        received = nsock.receive()
-        for received_json_pkt in received:
-            parse_json_str(received_json_pkt)
-
+    nsock.loop(onJsonLineRecvd, None)