<commit_before> /*
** Copyright (C) 2015 Eneo Tecnologia S.L.
** Author: <NAME> <<EMAIL>>
**
** This program is free software; you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program 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
** GNU Affero General Public License for more details.
**
** You should have received a copy of the GNU General Public License
** along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "util.h"
#include "global_config.h"
#include "librd/rdfile.h"
#include "librd/rdsysqueue.h"

#ifdef HAVE_LIBMICROHTTPD
#include "http.h"
#endif
#include "socket.h"

#include <errno.h>
#include <librd/rdlog.h>
#include <string.h>
#include <jansson.h>
#include <arpa/inet.h>
#include <assert.h>

#ifndef LIST_FOREACH_SAFE
#define	LIST_FOREACH_SAFE(var, head, field, tvar)			\
	for ((var) = LIST_FIRST((head));				\
	    (var) && ((tvar) = LIST_NEXT((var), field), 1);		\
	    (var) = (tvar))
#endif

#define CONFIG_LISTENERS_ARRAY "listeners"
#define CONFIG_PROTO_KEY "proto"
#define CONFIG_THREADS_KEY "threads"
#define CONFIG_TOPIC_KEY "topic"
#define CONFIG_BROKERS_KEY "brokers"
#define CONFIG_PORT_KEY "port"
#define CONFIG_DEBUG_KEY "debug"
#define CONFIG_RESPONSE_KEY "response"
#define CONFIG_BLACKLIST_KEY "blacklist"
#define CONFIG_MSE_SENSORS_KEY "mse-sensors"
#define CONFIG_MERAKI_SECRETS_KEY "meraki-secrets"
#define CONFIG_RBHTTP2K_CONFIG "rb_http2k_config"
#define CONFIG_RDKAFKA_KEY "rdkafka."
#define CONFIG_TCP_KEEPALIVE "tcp_keepalive"

#define CONFIG_PROTO_TCP  "tcp"
#define CONFIG_PROTO_UDP  "udp"
#define CONFIG_PROTO_HTTP "http"

#define CONFIG_DECODE_AS_NULL           ""
#define CONFIG_DECODE_AS_MSE            "MSE"
#define CONFIG_DECODE_AS_MERAKI         "meraki"
#define CONFIG_DECODE_AS_RBHTTP2K       "rb_http2k"

struct n2kafka_config global_config;

static const struct registered_decoder{
	const char *decode_as;
	const char *config_parameters;
	listener_callback cb;
	listener_opaque_creator opaque_creator;
	listener_opaque_reload opaque_reload;
	listener_opaque_destructor opaque_destructor;
} registered_decoders[] = {
	{CONFIG_DECODE_AS_NULL,NULL,dumb_decoder,NULL,NULL,NULL},
	// @TODO destructors
	{CONFIG_DECODE_AS_MSE,CONFIG_MSE_SENSORS_KEY,mse_decode,mse_opaque_creator,mse_opaque_reload,
	                                                                            mse_opaque_done},
	{CONFIG_DECODE_AS_MERAKI,CONFIG_MERAKI_SECRETS_KEY,meraki_decode,meraki_opaque_creator,
	                                         meraki_opaque_reload,meraki_opaque_destructor},
	{CONFIG_DECODE_AS_RBHTTP2K,CONFIG_RBHTTP2K_CONFIG,rb_decode,rb_opaque_creator,
	                                         rb_opaque_reload,rb_opaque_done}
};

static const struct registered_listener{
	const char *proto;
	listener_creator creator;
} registered_listeners[] = {
#ifdef HAVE_LIBMICROHTTPD
	{CONFIG_PROTO_HTTP, create_http_listener},
#endif
	{CONFIG_PROTO_TCP, create_tcp_listener},
	{CONFIG_PROTO_UDP, create_udp_listener},
};

void init_global_config(){
	memset(&global_config,0,sizeof(global_config));
	global_config.kafka_conf = rd_kafka_conf_new();
	global_config.kafka_topic_conf = rd_kafka_topic_conf_new();
	global_config.blacklist = in_addr_list_new();
	rd_log_set_severity(LOG_INFO);
	LIST_INIT(&global_config.listeners);

	init_mse_database(&global_config.mse.database);
}

static const char *assert_json_string(const char *key,const json_t *value){<commit_msg>New decoder flags that tells when it supports streaming decoding, insted of full buffer<commit_after> /*
** Copyright (C) 2015 Eneo Tecnologia S.L.
** Author: <NAME> <<EMAIL>>
**
** This program is free software; you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program 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
** GNU Affero General Public License for more details.
**
** You should have received a copy of the GNU General Public License
** along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "util.h"
#include "global_config.h"
#include "librd/rdfile.h"
#include "librd/rdsysqueue.h"

#ifdef HAVE_LIBMICROHTTPD
#include "http.h"
#endif
#include "socket.h"

#include <errno.h>
#include <librd/rdlog.h>
#include <string.h>
#include <jansson.h>
#include <arpa/inet.h>
#include <assert.h>

#ifndef LIST_FOREACH_SAFE
#define	LIST_FOREACH_SAFE(var, head, field, tvar)			\
	for ((var) = LIST_FIRST((head));				\
	    (var) && ((tvar) = LIST_NEXT((var), field), 1);		\
	    (var) = (tvar))
#endif

#define CONFIG_LISTENERS_ARRAY "listeners"
#define CONFIG_PROTO_KEY "proto"
#define CONFIG_THREADS_KEY "threads"
#define CONFIG_TOPIC_KEY "topic"
#define CONFIG_BROKERS_KEY "brokers"
#define CONFIG_PORT_KEY "port"
#define CONFIG_DEBUG_KEY "debug"
#define CONFIG_RESPONSE_KEY "response"
#define CONFIG_BLACKLIST_KEY "blacklist"
#define CONFIG_MSE_SENSORS_KEY "mse-sensors"
#define CONFIG_MERAKI_SECRETS_KEY "meraki-secrets"
#define CONFIG_RBHTTP2K_CONFIG "rb_http2k_config"
#define CONFIG_RDKAFKA_KEY "rdkafka."
#define CONFIG_TCP_KEEPALIVE "tcp_keepalive"

#define CONFIG_PROTO_TCP  "tcp"
#define CONFIG_PROTO_UDP  "udp"
#define CONFIG_PROTO_HTTP "http"

#define CONFIG_DECODE_AS_NULL           ""
#define CONFIG_DECODE_AS_MSE            "MSE"
#define CONFIG_DECODE_AS_MERAKI         "meraki"
#define CONFIG_DECODE_AS_RBHTTP2K       "rb_http2k"

struct n2kafka_config global_config;

static const struct registered_decoder{
	const char *decode_as;
	const char *config_parameters;
	listener_callback cb;
	listener_opaque_creator opaque_creator;
	listener_opaque_reload opaque_reload;
	listener_opaque_destructor opaque_destructor;

/**
    Tells if the listener support straming API, instead of wait to all data
    to be available
*/
#define DECODER_F_SUPPORT_STREAMING 0x01
    int flags;
} registered_decoders[] = {
	{CONFIG_DECODE_AS_NULL,NULL,dumb_decoder,NULL,NULL,NULL,0},
	// @TODO destructors
	{CONFIG_DECODE_AS_MSE,CONFIG_MSE_SENSORS_KEY,mse_decode,mse_opaque_creator,mse_opaque_reload,
	    mse_opaque_done,0},
	{CONFIG_DECODE_AS_MERAKI,CONFIG_MERAKI_SECRETS_KEY,meraki_decode,meraki_opaque_creator,
	    meraki_opaque_reload,meraki_opaque_destructor,0},
	{CONFIG_DECODE_AS_RBHTTP2K,CONFIG_RBHTTP2K_CONFIG,rb_decode,rb_opaque_creator,
	    rb_opaque_reload,rb_opaque_done,0}
};

static const struct registered_listener{
	const char *proto;
	listener_creator creator;
} registered_listeners[] = {
#ifdef HAVE_LIBMICROHTTPD
	{CONFIG_PROTO_HTTP, create_http_listener},
#endif
	{CONFIG_PROTO_TCP, create_tcp_listener},
	{CONFIG_PROTO_UDP, create_udp_listener},
};

void init_global_config(){
	memset(&global_config,0,sizeof(global_config));
	global_config.kafka_conf = rd_kafka_conf_new();
	global_config.kafka_topic_conf = rd_kafka_topic_conf_new();
	global_config.blacklist = in_addr_list_new();
	rd_log_set_severity(LOG_INFO);
	LIST_INIT(&global_config.listeners);

	init_mse_database(&global_config.mse.database);
}

static const char *assert_json_string(const char *key,const json_t *value){
{'fraction_non_alphanumeric': 0.09555730809674028, 'fraction_numerical': 0.004600420609884332, 'mean_word_length': 5.593587521663778, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '603918', 'n_tokens_mistral': 2730, 'n_tokens_neox': 2655, 'n_words': 609}