#include "common.h"
#include "dy_utils/dy_common.h"

static int send_data_to_socket(int socket, char *payload, int payload_len)
{
    int len;

    len = socket_send(socket, payload, payload_len);
    if (len < 0)
    {
        dy_syslog(LOG_INFO, "send error: errno:%d, reason:%s", errno, strerror(errno));
    }
    else
    {
        dy_syslog_hex(LOG_DEBUG, payload, payload_len, "send data to socket:%d", socket);
    }

    return 0;
}

static int mqtt_handle_data(mqtt_message_t *mqtt_msg)
{
    char *buf = NULL;
    cJSON *root = cJSON_Parse(mqtt_msg->payload);
    int len;
    int socket = -1;

    if (root)
    {
        char *payload = NULL;
        GET_JSON_VALUE_DY_STRING(root, "payload", payload);
        buf = (char *)malloc(MAXBUF);
        len = b64_decode(payload, buf, MAXBUF);
        free(payload);
    }
    else
    {
        dy_syslog(LOG_ERR, "json parse error");
        return -1;
    }
    GET_JSON_VALUE_INT(root, "socket", socket);
    cJSON_Delete(root);
    if (socket != -1)
    {
        send_data_to_socket(socket, buf, len);
    }
    free(buf);

    return 0;
}

static int recv_mqtt_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    char topic[TOPIC_MAX_LEN] = {0};
    unsigned char matched;
    int j = 0, i = 0;
    int ret;

    mqtt_session_t *session = (mqtt_session_t *)obj;

    dy_syslog(LOG_DEBUG, "MQTT client received MQTT topic:%s payload:%s",
              mqtt_msg->topic, mqtt_msg->payload);

    if (strstr(mqtt_msg->topic, "transparent"))
    {
        // data
        mqtt_handle_data(mqtt_msg);
    }

out:
    return 0;
}

int mqtt_init(socket_mqtt_var_t *var)
{
    int i = 0;
    char topic[TOPIC_MAX_LEN] = {0};

    for (i = 0; i < var->mqtt_cnt; i++)
    {
        mqtt_session_t *session = var->mqtt[i];
        if (session == NULL)
        {
            dy_syslog(LOG_ERR, "malloc error");
            continue;
        }

        mqtt_session_set_opts(session, DEFAULT_QOS, KEEPALIVEINTERVAL);
        mqtt_session_set_callbacks(session, recv_mqtt_msg, NULL);
        mqtt_session_start(session);

        sprintf(topic, "GW/%s/+/transparent", session->client.clientId);
        mqtt_session_subscribe(session, topic);
    }
}