#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "common.h"
#include "dy_utils/cJSON.h"
#include "libubox/utils.h"

/*
 * forward the received socket data to MQTT
 */

// we don't know the SN, send the broadcast topic
static int _broadcast_to_mqtt(socket_data_node_t *pkt)
{
    char topic[TOPIC_MAX_LEN] = {0};
    cJSON *broadcast = cJSON_CreateObject();
    struct sockaddr_in peer_addr;
    int peer_len;
    char ip[INET_ADDRSTRLEN];
    char *data_tmp = NULL;
    char *encode_str = NULL;
    int i = 0;

    peer_len = sizeof(peer_addr);
    getpeername(pkt->socket, (struct sockaddr *)&peer_addr, &peer_len);
    inet_ntop(AF_INET, &peer_addr.sin_addr, ip, sizeof(ip));

    cJSON_AddStringToObject(broadcast, "ip", ip);
    cJSON_AddNumberToObject(broadcast, "port", ntohs(peer_addr.sin_port));
    cJSON_AddStringToObject(broadcast, "type", "TCP");
    cJSON_AddNumberToObject(broadcast, "socket", pkt->socket);

    // 将payload进行base64 encode
    encode_str = (char *)malloc(MAXBUF);
    b64_encode(pkt->payload, pkt->payload_len, encode_str, MAXBUF);
    cJSON_AddStringToObject(broadcast, "payload", encode_str);

    // 以字符串的方式发送数据
    sprintf(topic, "F/SearchSN");
    for (i = 0; i < socket_mqtt_var.mqtt_cnt; i++)
    {
        cJSON_AddStringToObject(broadcast, "clientId", socket_mqtt_var.mqtt[i]->client.clientId);
        data_tmp = cJSON_Print(broadcast);
        cJSON_DeleteItemFromObject(broadcast, "clientId");
        mqtt_session_publish(socket_mqtt_var.mqtt[i], topic, data_tmp, strlen(data_tmp));
        free(data_tmp);
    }

    // 回收资源
    free(encode_str);
    cJSON_Delete(broadcast);

    return 0;
}

static int _unicast_to_mqtt(sock_sess_node_t *sess, uint8_t *payload, int payload_len)
{
    char topic[TOPIC_MAX_LEN] = {0};
    int socket = sess->socket;
    cJSON *unicast = cJSON_CreateObject();
    struct sockaddr_in peerAddr;
    int peerLen;
    char ipAddr[INET_ADDRSTRLEN];
    char *data_tmp = NULL;
    char *encode_str = NULL;

    peerLen = sizeof(peerAddr);
    getpeername(socket, (struct sockaddr *)&peerAddr, &peerLen);
    inet_ntop(AF_INET, &peerAddr.sin_addr, ipAddr, sizeof(ipAddr));
    cJSON_AddStringToObject(unicast, "ip", ipAddr);
    cJSON_AddNumberToObject(unicast, "port", ntohs(peerAddr.sin_port));
    cJSON_AddStringToObject(unicast, "type", "TCP");
    cJSON_AddNumberToObject(unicast, "socket", socket);
    cJSON_AddStringToObject(unicast, "clientId", sess->mqtt->client.clientId);
    encode_str = (char *)malloc(MAXBUF);
    b64_encode(payload, payload_len, encode_str, MAXBUF);
    cJSON_AddStringToObject(unicast, "payload", encode_str);
    data_tmp = cJSON_Print(unicast);

    sprintf(topic, "F/%s/transprent", sess->sn);
    mqtt_session_publish(sess->mqtt, topic, data_tmp, strlen(data_tmp));

    free(encode_str);
    free(data_tmp);
    cJSON_Delete(unicast);

    return 0;
}

// 真正的处理转发socket那边的packet
static int _do_forward(struct list_head *fwd_list)
{
    socket_data_node_t *tmp = NULL;
    socket_data_node_t *pkt = NULL;
    sock_sess_node_t *sess = NULL;
    const char *sn = NULL;

    if (list_empty(fwd_list))
    {
        return 0;
    }

    list_for_each_entry_safe(pkt, tmp, fwd_list, list)
    {
        sess = find_session_by_socket(pkt->socket);
        if (sess == NULL)
        {
            //broadcast
            _broadcast_to_mqtt(pkt);
        }
        else
        {
            //unicast to sn
            _unicast_to_mqtt(sess, pkt->payload, pkt->payload_len);
        }
        list_del(&pkt->list);
        free(pkt);
    }

    return 0;
}

static int do_forward()
{
    struct list_head tmp_fwd_list = LIST_HEAD_INIT(tmp_fwd_list);
    struct timeval tv;
    struct timespec ts;

    pthread_mutex_lock(&socket_mqtt_var.socket_data_lock);

    while (socket_mqtt_var.socket_received == 0)
    {
        int ret = 0;

        gettimeofday(&tv, NULL);
        ts.tv_sec  = tv.tv_sec;
        ts.tv_nsec = tv.tv_usec * 1000;
        ts.tv_sec += WAIT_TIME_SECONDS;
        ret = pthread_cond_timedwait(&socket_mqtt_var.socket_recv_cond, &socket_mqtt_var.socket_data_lock, &ts);
    }
    // 将socket的接收list移到tmp_fwd_list,
    // 然后处理tmp_fwd_list时可以不需要锁
    if (!list_empty(&socket_mqtt_var.socket_rcv_head))
    {
        list_move_tail_list(&socket_mqtt_var.socket_rcv_head, &tmp_fwd_list);
    }
    else
    {
        dy_syslog(LOG_ERR, "no packet received, should not run here");
    }
    socket_mqtt_var.socket_received = 0;
    pthread_mutex_unlock(&socket_mqtt_var.socket_data_lock);

    _do_forward(&tmp_fwd_list);

    return 0;
}

void *socket_forward_loop(void *param)
{
    while (1)
    {
        do_forward();
    }
}