#include "cloud_mqtt.h"
#include "../common.h"
#include "pub_fun.h"

static mqtt_platform *platform[] = {
    // &lnxall_platform,
};

static int cloud_mqtt_recv_handle(void *obj, ipc_msg_t *mqtt_msg)
{
    char tx_buff[4096] = {0};
    int len = 0;
    cloud_mqtt *cloud = (cloud_mqtt *)obj;
    const cloud_mqtt_its *its = cloud->its;

    ems_syslog(LOG_DEBUG, "[%s:%d]received MQTT topic:%s payload length:%d", cloud->server_info.ip, cloud->server_info.port, mqtt_msg->topic, mqtt_msg->payloadLen);
    ems_syslog(LOG_DEBUG, "payload:%s", mqtt_msg->payload);

    for (int i = 0; i < its->rx_it_sum; i++)
    {
        if (strcmp(mqtt_msg->topic, its->rx_it[i].rx_topic) == 0)
        {
            len = its->rx_it[i].parse_message(its, mqtt_msg->payload, mqtt_msg->payloadLen, tx_buff, sizeof(tx_buff));
            if (len < 0)
            {
                ems_syslog(LOG_ERR, "parse message error:%d", len);
            }
            else
            {
                len = mqtt_session_publish(cloud->mqtt, (char *)its->rx_it[i].tx_topic, tx_buff, len);
                if (len != 0)
                {
                    ems_syslog(LOG_ERR, "publish fail, topic: %s, ret = %d", its->rx_it[i].tx_topic, len);
                }
                else
                {
                    ems_syslog(LOG_INFO, "publish message success, topic: %s", its->rx_it[i].tx_topic);
                }
            }
        }
    }

    return 0;
}

static int cloud_mqtt_state_change(void *obj, int state)
{
    return 0;
}

static cloud_mqtt_its *new_mqtt_its(const char *platform_name)
{
	int plat_siz = sizeof(platform) / sizeof(platform[0]);

    for (int i = 0; i < plat_siz; i++)
    {
        if (strcmp(platform_name, platform[i]->name) == 0)
        {
            return platform[i]->new_its();
        }
    }
    return NULL;
}

static int delete_mqtt_its(cloud_mqtt_its *its)
{
	int plat_siz = sizeof(platform) / sizeof(platform[0]);

    for (int i = 0; i < plat_siz; i++)
    {
        if (strcmp(its->name, platform[i]->name) == 0)
        {
            if (platform[i]->delete_its != NULL)
            {
                return platform[i]->delete_its(its);
            }
        }
    }

    ems_syslog(LOG_INFO, "delete mqtt its by default");
    int tmp = its->rx_it_sum;
    its->rx_it_sum = 0;
    for (int i = 0; i < tmp; i++)
    {
        free(its->rx_it[i].rx_topic);
        free(its->rx_it[i].tx_topic);
    }
    free(its->rx_it);

    tmp = its->tx_it_sum;
    its->tx_it_sum = 0;
    for (int i = 0; i < tmp; i++)
    {
        free(its->tx_it[i].topic);
    }
    free(its->tx_it);

    free(its->name);
    free(its->station_id);
    return 0;
}

/************************************************************************************************/
static int cloud_mqtt_get_server_info(mqtt_server_info *server_info)
{
    int re = 0;
    const char *cfg_f = "/app/config/mqtt_server.json";
    char *f_data = read_file_data(cfg_f);
    if (f_data == NULL)
    {
        proto_syslog(LOG_ERR, "load device config file error, file:%s", cfg_f);
        return -1;
    }

    cJSON *root = json_parse_string_with_comments(f_data);
    if (!root)
    {
        proto_syslog(LOG_ERR, "parse file to json obj error, file:%s", cfg_f);
        re = -2;
        goto cloud_mqtt_get_server_info_exit;
    }

    cJSON *obj = cJSON_GetObjectItemCaseSensitive(root, "host");
    server_info->ip = strdup(obj->valuestring);
    obj = cJSON_GetObjectItemCaseSensitive(root, "port");
    server_info->port = obj->valueint;
    obj = cJSON_GetObjectItemCaseSensitive(root, "user");
    server_info->user = strdup(obj->valuestring);
    obj = cJSON_GetObjectItemCaseSensitive(root, "pass");
    server_info->pass = strdup(obj->valuestring);

    cJSON_Delete(root);
cloud_mqtt_get_server_info_exit:

    free(f_data);
    return re;
}

cloud_mqtt *new_cloud_mqtt(mqtt_server_info *server_info, const char *platform_name)
{
    mqtt_server_info ser_info = {0};
    if (server_info == NULL)
    {
        if (0 == cloud_mqtt_get_server_info(&ser_info))
        {
            server_info = &ser_info;
        }
    }

    char client_id[48] = {0};
    snprintf(client_id, sizeof(client_id), "%s_client_id_%lld", "gw_sn_buff", clock_get_ms());
    proto_syslog(LOG_NOTICE, "cloud_mqtt client_id:%s", client_id);
    cloud_mqtt_its *its = new_mqtt_its(platform_name);
    if (its == NULL)
    {
        ems_syslog(LOG_ERR, "get its error, platform name:%s", platform_name);
        return NULL;
    }

    cloud_mqtt *cloud = calloc(1, sizeof(cloud_mqtt));
    mqtt_session_t *session = mqtt_session_new(client_id, cloud);
    if (session == NULL)
    {
        free(cloud);
        ems_syslog(LOG_ERR, "mqtt_session_new fail,ip:%s,port:%d", server_info->ip, server_info->port);
        return NULL;
    }

    cloud->mqtt = session;
    cloud->server_info = *server_info;
    cloud->its = its;
    mqtt_session_set_address(cloud->mqtt, cloud->server_info.ip, cloud->server_info.port, cloud->server_info.user, cloud->server_info.pass);
    mqtt_session_set_opts(cloud->mqtt, cloud->server_info.qos, cloud->server_info.keepalive);
    mqtt_session_set_callbacks(cloud->mqtt, cloud_mqtt_recv_handle, cloud_mqtt_state_change);

    return cloud;
}

int delete_cloud_mqtt(cloud_mqtt *cloud)
{
    mqtt_session_disconn(cloud->mqtt);
    mqtt_session_destroy_and_free(cloud->mqtt);
    delete_mqtt_its(cloud->its);
    free(cloud);
    return 0;
}

int cloud_mqtt_start(cloud_mqtt *cloud)
{
    if (cloud == NULL)
    {
        return -1000;
    }
    // 订阅
    for (int i = 0; i < cloud->its->rx_it_sum; i++)
    {
        if (0 == mqtt_session_subscribe(cloud->mqtt, cloud->its->rx_it[i].rx_topic))
        {
            ems_syslog(LOG_NOTICE, "subscribe topic: %s success", cloud->its->rx_it[i].rx_topic);
        }
        else
        {
            ems_syslog(LOG_ERR, "subscribe topic: %s fail", cloud->its->rx_it[i].rx_topic);
        }
    }

    return mqtt_session_start(cloud->mqtt);
}

int cloud_mqtt_update(cloud_mqtt *cloud)
{
    char tx_buff[2048] = {0};
    int ret = 0;
    int tx_len = 0;
    double currue_power = dev_get_dev_tag_float(DEV_NO_EMS, SET_ACT_POWER);
    if (IS_IDLE(currue_power)) // 空闲上报慢一些
    {
        if (time(NULL) < (cloud->last_update_time + (10 * 60)))
        {
            return ret;
        }
    }

    cloud->last_update_time = time(NULL);
    const cloud_mqtt_its *its = cloud->its;
    for (int i = 0; i < its->tx_it_sum; i++)
    {
        tx_len = its->tx_it[i].create_message(its, tx_buff, sizeof(tx_buff));
        if (tx_len > 0)
        {
            tx_len = mqtt_session_publish(cloud->mqtt, (char *)its->tx_it[i].topic, tx_buff, tx_len);
            if (tx_len != 0)
            {
                ems_syslog(LOG_ERR, "publish fail, topic: %s", its->tx_it[i].topic);
                ret--;
            }
            else
            {
                ems_syslog(LOG_INFO, "publish message success, topic: %s", its->tx_it[i].topic);
            }
        }
        else
        {
            ems_syslog(LOG_ERR, "create mqtt message fail, topic: %s", its->tx_it[i].topic);
            ret--;
        }
    }

    return ret;
}