#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>

#include "can_mqtt_proxy_common.h"

static int can_rcv_frame(void *obj, can_frame_node_t *node)
{
    can_mqtt_proxy_var_t *var = (can_mqtt_proxy_var_t *)obj;
    can_frame_packed_t can_packed;
    char *data_tmp = NULL;
    static char *buf = NULL;
    cJSON *rt_data = cJSON_CreateObject();
    char topic[TOPIC_MAX_LEN] = {0};
    static int mi = 0;

    can_packed.can_id = node->frame.can_id;
    can_packed.can_dlc = node->frame.can_dlc;
    memcpy(can_packed.data, node->frame.data, 8);

    // 以字符串的方式发送数据
    data_tmp = cJSON_Print(rt_data);

    mqtt_session_publish(var->session, var->pub_topic, (char *)&can_packed, sizeof(can_packed));

    // 回收资源
    free(data_tmp);
    cJSON_Delete(rt_data);

    return 0;
}

static void can_mqtt_proxy_loop(can_mqtt_proxy_var_t *var)
{
    while (1)
    {
        sleep(20);
    }
}

static void can_mqtt_proxy_subscribe_all(can_mqtt_proxy_var_t *var)
{
    mqtt_session_t *mqtt_session = var->session;

    mqtt_session_subscribe(mqtt_session, var->sub_topic);
}

static int can_mqtt_proxy_handle_rsp(can_mqtt_proxy_var_t *var, mqtt_message_t *mqtt_msg)
{
    int ret = 0;
    can_frame_packed_t *can_packed = NULL;
    struct can_frame can;
    int port_idx;

    if (mqtt_msg->payloadLen != sizeof(can_frame_packed_t))
    {
        dy_syslog(LOG_WARNING, "can frame length not right");
        return -1;
    }

    can_packed = (can_frame_packed_t *)mqtt_msg->payload;

    if (can_packed == NULL)
    {
        dy_syslog(LOG_ERR, "can frame is empty");
        goto out;
    }

    can.can_id = can_packed->can_id;
    can.can_dlc = can_packed->can_dlc;
    memcpy(can.data, can_packed->data, 8);

    port_idx = can_port_map(var->port);
    ret = can_send(var->can_cfg->can_dev[port_idx].fd, &can);
    if (ret < 0)
    {
        dy_syslog(LOG_ERR, "ret:%d", ret);
    }

out:
    return ret;
}

static int can_mqtt_proxy_mqtt_handle_recv_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    can_mqtt_proxy_var_t *var = (can_mqtt_proxy_var_t *)obj;
    bool matched;
    int ret;

    ret = mosquitto_topic_matches_sub(var->sub_topic, mqtt_msg->topic, &matched);
    if (ret == 0 && matched)
    {
        //发送下行数据
        can_mqtt_proxy_handle_rsp(var, mqtt_msg);
    }
    else
    {
    }
}

// 建立与broker之间的MQTT连接
static int can_mqtt_proxy_mqtt_client_init(can_mqtt_proxy_var_t *var, char *cfg_file)
{

    char clientId[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "%s_can", var->sn_str);
    var->session = mqtt_session_new(clientId, (void *)var);
    if (var->session == NULL)
        return -1;

    if (load_server_config(var->session, cfg_file) != 0)
    {
        dy_syslog(LOG_ERR, "load MQTT server cfg file failed");
        return -1;
    }
    mqtt_session_set_opts(var->session, DEFAULT_QOS, KEEPALIVEINTERVAL);
    mqtt_session_set_callbacks(var->session, can_mqtt_proxy_mqtt_handle_recv_msg, NULL);
    can_mqtt_proxy_subscribe_all(var);
    mqtt_session_start(var->session);

    return 0;
}

static int can_mqtt_proxy_load_cfg(can_mqtt_proxy_var_t *var, char *cfg_file)
{
    char *data = NULL;
    cJSON *root = NULL;
    int ret = -1;
    char buf[16] = {0};
    char topic[TOPIC_MAX_LEN] = {0};
    char topic_new[TOPIC_MAX_LEN] = {0};

    if (cfg_file == NULL)
    {
        dy_syslog(LOG_ERR, "cfg_file:%s not exist", cfg_file);
        goto out;
    }

    data = read_file_data(cfg_file);
    if (data == NULL)
    {
        dy_syslog(LOG_ERR, "read cfg file %s error", cfg_file);
        goto out;
    }

    root = cJSON_Parse(data);
    if (!root)
    {
        dy_syslog(LOG_ERR, "parse cfg file %s error", cfg_file);
        goto out;
    }

    GET_JSON_VALUE_STRING(root, "sub", topic);
    if (strlen(topic) == 0)
    {
        dy_syslog(LOG_ERR, "sub topic is empty");
        goto out;
    }
    replace_sub_str(topic, "[GW_SN]", var->sn_str, topic_new);
    var->sub_topic = strdup(topic_new);

    GET_JSON_VALUE_STRING(root, "pub", topic);
    if (strlen(topic) == 0)
    {
        dy_syslog(LOG_ERR, "pub topic is empty");
        goto out;
    }
    replace_sub_str(topic, "[GW_SN]", var->sn_str, topic_new);
    var->pub_topic = strdup(topic_new);

    GET_JSON_VALUE_STRING(root, "port", buf);
    if (strlen(buf) == 0)
    {
        strcpy(buf, "CANBUS_0");
        dy_syslog(LOG_WARNING, "port is empty, use default can 0");
    }
    var->port = port_char2enum(buf);

    cJSON_Delete(root);

    dy_syslog(LOG_DEBUG, "pub:%s sub:%s", var->pub_topic, var->sub_topic);
    ret = 0;

out:
    if (data)
    {
        free(data);
    }
    return ret;
}

static int can_mqtt_proxy_init(can_mqtt_proxy_var_t *var)
{
    get_board_sn(var->sn_str);
    if (can_mqtt_proxy_load_cfg(var, CAN_MQTT_PROXY_CFG) < 0)
    {
        dy_syslog(LOG_ERR, "load cfg failed");
        return -1;
    }

    if (can_mqtt_proxy_mqtt_client_init(var, MQTT_SERVER_FILE) < 0)
    {
        dy_syslog(LOG_ERR, "load mqtt cfg failed");
        return -1;
    }

    var->can_cfg = dy_can_cfg_init((void *)var, NULL, can_rcv_frame);
    if (load_can_cfg_and_init(var->can_cfg, CAN_MODULE_CFG) < 0)
    {
        dy_syslog(LOG_ERR, "load can cfg failed");
        return -1;
    }

    return 0;
}

int main(int argc, char *argv[])
{
    can_mqtt_proxy_var_t var = {0};

    memset(&var, 0, sizeof(var));

    can_mqtt_proxy_init(&var);
    can_mqtt_proxy_loop(&var);

    return 0;
}