<commit_before>/*
 * This file is part of the KNOT Project
 *
 * Copyright (c) 2019, CESAR. All rights reserved.
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library 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
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include <stdbool.h>
#include <stdint.h>
#include <ell/ell.h>
#include <json-c/json.h>
#include <hal/linux_log.h>
#include <amqp.h>

#include <knot/knot_protocol.h>

#include "settings.h"
#include "amqp.h"
#include "parser.h"
#include "cloud.h"

#define AMQP_QUEUE_FOG "fog-messages"

/* Exchanges */
#define AMQP_EXCHANGE_FOG "fog"
#define AMQP_EXCHANGE_CLOUD "cloud"

 /* Southbound traffic (commands) */
#define AMQP_EVENT_DATA_UPDATE "data.update"
#define AMQP_EVENT_DATA_REQUEST "data.request"
#define AMQP_EVENT_DEVICE_UNREGISTERED "device.unregistered"

 /* Northbound traffic (control, measurements) */
#define AMQP_CMD_DATA_PUBLISH "data.publish"
#define AMQP_CMD_DEVICE_UNREGISTER "device.unregister"

/* Hashmap to user callback from southbound traffic */
static struct l_hashmap *map_event_to_cb;

/* Union with all cloud callbacks */
union cloud_cb_t {
	cloud_downstream_cb_t update_cb;
	cloud_downstream_cb_t request_cb;
	cloud_device_removed_cb_t removed_cb;
};

/* Struct that has a callback #cb that handle event in #event_id */
struct event_handler {
	void *cb;
	enum {
		UPDATE_EVT,
		REQUEST_EVT,
		REMOVED_EVT
	} event_id;
};

static const char *get_id_from_json_obj(const json_object *jso)
{
	json_object *id_json;

	if (!json_object_object_get_ex(jso, "id", &id_json))
		return NULL;

	if (json_object_get_type(id_json) != json_type_string)
		return NULL;

	return json_object_get_string(id_json);
}

/**
 * Callback function to consume and parse the received message from AMQP queue
 * and call the respective handling callback function. In case of a error on
 * parse, the message is consumed, but not used.
 *
 * Returns true if the message envelope was consumed or returns false otherwise.
 */
static bool on_cloud_receive_message(const char *exchange,
				     const char *routing_key,
				     const char *body, void *user_data)
{
	json_object *jso;
	struct l_queue *list;
	const char *id;
	bool consumed = true;
	const struct event_handler *evt_handler;
	union cloud_cb_t cb_handler;

	evt_handler = l_hashmap_lookup(map_event_to_cb, routing_key);
	if (!evt_handler) {
		hal_log_error("Error cb handler not set");
		return false;
	}

	jso = json_tokener_parse(body);
	if (!jso) {
		hal_log_error("Error on parse JSON object");
		return consumed;
	}

	switch (evt_handler->event_id) {
	case UPDATE_EVT:
		cb_handler.update_cb = evt_handler->cb;
		id = get_id_from_json_obj(jso);
		if (!id) {
			hal_log_error("Malformed JSON message");
			break;
		}

		list = parser_update_to_list(jso);
		if (list == NULL)
			break;

		consumed = cb_handler.update_cb(id, list, user_data);<commit_msg>Log malformed json on update list in cloud source<commit_after>/*
 * This file is part of the KNOT Project
 *
 * Copyright (c) 2019, CESAR. All rights reserved.
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library 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
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include <stdbool.h>
#include <stdint.h>
#include <ell/ell.h>
#include <json-c/json.h>
#include <hal/linux_log.h>
#include <amqp.h>

#include <knot/knot_protocol.h>

#include "settings.h"
#include "amqp.h"
#include "parser.h"
#include "cloud.h"

#define AMQP_QUEUE_FOG "fog-messages"

/* Exchanges */
#define AMQP_EXCHANGE_FOG "fog"
#define AMQP_EXCHANGE_CLOUD "cloud"

 /* Southbound traffic (commands) */
#define AMQP_EVENT_DATA_UPDATE "data.update"
#define AMQP_EVENT_DATA_REQUEST "data.request"
#define AMQP_EVENT_DEVICE_UNREGISTERED "device.unregistered"

 /* Northbound traffic (control, measurements) */
#define AMQP_CMD_DATA_PUBLISH "data.publish"
#define AMQP_CMD_DEVICE_UNREGISTER "device.unregister"

/* Hashmap to user callback from southbound traffic */
static struct l_hashmap *map_event_to_cb;

/* Union with all cloud callbacks */
union cloud_cb_t {
	cloud_downstream_cb_t update_cb;
	cloud_downstream_cb_t request_cb;
	cloud_device_removed_cb_t removed_cb;
};

/* Struct that has a callback #cb that handle event in #event_id */
struct event_handler {
	void *cb;
	enum {
		UPDATE_EVT,
		REQUEST_EVT,
		REMOVED_EVT
	} event_id;
};

static const char *get_id_from_json_obj(const json_object *jso)
{
	json_object *id_json;

	if (!json_object_object_get_ex(jso, "id", &id_json))
		return NULL;

	if (json_object_get_type(id_json) != json_type_string)
		return NULL;

	return json_object_get_string(id_json);
}

/**
 * Callback function to consume and parse the received message from AMQP queue
 * and call the respective handling callback function. In case of a error on
 * parse, the message is consumed, but not used.
 *
 * Returns true if the message envelope was consumed or returns false otherwise.
 */
static bool on_cloud_receive_message(const char *exchange,
				     const char *routing_key,
				     const char *body, void *user_data)
{
	json_object *jso;
	struct l_queue *list;
	const char *id;
	bool consumed = true;
	const struct event_handler *evt_handler;
	union cloud_cb_t cb_handler;

	evt_handler = l_hashmap_lookup(map_event_to_cb, routing_key);
	if (!evt_handler) {
		hal_log_error("Error cb handler not set");
		return false;
	}

	jso = json_tokener_parse(body);
	if (!jso) {
		hal_log_error("Error on parse JSON object");
		return consumed;
	}

	switch (evt_handler->event_id) {
	case UPDATE_EVT:
		cb_handler.update_cb = evt_handler->cb;
		id = get_id_from_json_obj(jso);
		list = parser_update_to_list(jso);
		if (!id || !list) {
			hal_log_error("Malformed JSON message");
			l_queue_destroy(list, l_free);
			break;
		}

		consumed = cb_handler.update_cb(id, list, user_data);
{'fraction_non_alphanumeric': 0.0906319290465632, 'fraction_numerical': 0.0047117516629711755, 'mean_word_length': 4.5558121632024635, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 23, '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': '1006780', 'n_tokens_mistral': 2545, 'n_tokens_neox': 2432, 'n_words': 774}