<commit_before>#include <stdlib.h>
#include <string.h>
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>

#include "uv.h"

#include "sad.h"
#include "bagride.h"
#include "mmtrace.h"

#include "mss_filter_config.h"

#define HELP_USAGE "usage: mss_filter cfg_file" 


static sad_filter_t filter;

static struct mss_filter_config_s config;

static char last_sentence[1024] = {0};
static char forward_sentence[1024] = {0};

static mmpool_t* udp_clients = NULL;
static mmpool_t* udp_servers = NULL;
static mmpool_t* http_servers = NULL;
static mmpool_t* tcp_servers = NULL;



static int on_stats_response(br_http_client_t* cli_) {
    cli_->m_resbuf.len = sad_stats_string(&cli_->m_resbuf.base, &filter);
    return 0;
}

static int on_udp_parse(ssize_t nread_, const uv_buf_t* inbuf_, br_udp_server_t* pserver_) {
    (void) pserver_;
    sad_decode_multiline(&filter, inbuf_->base, nread_);
    return 0;
}

static int on_ais_decoded(struct sad_filter_s * filter_) {

    struct ais_t * ais = &filter_->ais;
    sub0_substring_t* sentence = filter_->sentence;

    if (1 == ais->type || 2 == ais->type || 3 == ais->type) {

        if (0 == strncmp(last_sentence, sentence->start, sentence->n)) {
            /* drop duplicate */
#ifdef NDEBUG            
            printf("[KO] drop duplicate %08" PRIu64 " type:%02d mmsi:%09u %s",
                    filter_->sentences,
                    ais->type,
                    ais->mmsi,
                    filter_->sentence->start);
#endif            
            return 0;
        }
        strncpy(last_sentence, sentence->start, sentence->n);
        last_sentence[sentence->n + 1] = '0';

        const double lat = (double) ais->type1.lat / AIS_LATLON_DIV;
        const double lon = (double) ais->type1.lon / AIS_LATLON_DIV;

        if (lon > config.geofilter.x1 && lon < config.geofilter.x2 
          && lat < config.geofilter.y1 && lat > config.geofilter.y2) {

            strncpy(forward_sentence, sentence->start, sentence->n);
            forward_sentence[sentence->n] = '\n';
            forward_sentence[sentence->n + 1] = '\0';

#ifdef MM_ULTRADEBUG
            printf("[ok] %08" PRIu64 " type:%02d mmsi:%09u lat:%f lon:%f %s",
                    filter_->sentences, ais->type, ais->mmsi,lat,lon,forward_sentence);
#endif /* MM_ULTRADEBUG */

            br_udp_clients_send(udp_clients, forward_sentence);
            br_tcp_write_string((br_tcp_server_t*)(tcp_servers->items[0]->m_p), 
                    forward_sentence, sentence->n + 1);
        }
    }
    return 0;
}

static int on_tcp_parse(ssize_t nread_, const uv_buf_t* inbuf_, br_tcp_server_t* pserver_) {<commit_msg>fix crash on udp client send, when no udp client<commit_after>#include <stdlib.h>
#include <string.h>
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>

#include "uv.h"

#include "sad.h"
#include "bagride.h"
#include "mmtrace.h"

#include "mss_filter_config.h"

#define HELP_USAGE "usage: mss_filter cfg_file" 


static sad_filter_t filter;

static struct mss_filter_config_s config;

static char last_sentence[1024] = {0};
static char forward_sentence[1024] = {0};

static mmpool_t* udp_clients = NULL;
static mmpool_t* udp_servers = NULL;
static mmpool_t* http_servers = NULL;
static mmpool_t* tcp_servers = NULL;



static int on_stats_response(br_http_client_t* cli_) {
    cli_->m_resbuf.len = sad_stats_string(&cli_->m_resbuf.base, &filter);
    return 0;
}

static int on_udp_parse(ssize_t nread_, const uv_buf_t* inbuf_, br_udp_server_t* pserver_) {
    (void) pserver_;
    sad_decode_multiline(&filter, inbuf_->base, nread_);
    return 0;
}

static int on_ais_decoded(struct sad_filter_s * filter_) {

    struct ais_t * ais = &filter_->ais;
    sub0_substring_t* sentence = filter_->sentence;

    if (1 == ais->type || 2 == ais->type || 3 == ais->type) {

        if (0 == strncmp(last_sentence, sentence->start, sentence->n)) {
            /* drop duplicate */
#ifdef NDEBUG            
            printf("[KO] drop duplicate %08" PRIu64 " type:%02d mmsi:%09u %s",
                    filter_->sentences,
                    ais->type,
                    ais->mmsi,
                    filter_->sentence->start);
#endif            
            return 0;
        }
        strncpy(last_sentence, sentence->start, sentence->n);
        last_sentence[sentence->n + 1] = '0';

        const double lat = (double) ais->type1.lat / AIS_LATLON_DIV;
        const double lon = (double) ais->type1.lon / AIS_LATLON_DIV;

        if (lon > config.geofilter.x1 && lon < config.geofilter.x2 
          && lat < config.geofilter.y1 && lat > config.geofilter.y2) {

            strncpy(forward_sentence, sentence->start, sentence->n);
            forward_sentence[sentence->n] = '\n';
            forward_sentence[sentence->n + 1] = '\0';

#ifdef MM_ULTRADEBUG
            printf("[ok] %08" PRIu64 " type:%02d mmsi:%09u lat:%f lon:%f %s",
                    filter_->sentences, ais->type, ais->mmsi,lat,lon,forward_sentence);
#endif /* MM_ULTRADEBUG */

            if(0 < config.ais_out_udp.n){
              br_udp_clients_send(udp_clients, forward_sentence);
            }
            
            br_tcp_write_string((br_tcp_server_t*)(tcp_servers->items[0]->m_p), 
                    forward_sentence, sentence->n + 1);
        }
    }
    return 0;
}

static int on_tcp_parse(ssize_t nread_, const uv_buf_t* inbuf_, br_tcp_server_t* pserver_) {
{'fraction_non_alphanumeric': 0.13067552602436322, 'fraction_numerical': 0.017534145441122185, 'mean_word_length': 2.750173010380623, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2308580', 'n_tokens_mistral': 2049, 'n_tokens_neox': 1924, 'n_words': 393}