#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
 */
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, "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);

    // 以字符串的方式发送数据
    for (i = 0; i < socket_mqtt_var.mqtt_cnt; i++)
    {
        char *client_id = mqtt_session_get_client_id(socket_mqtt_var.mqtt[i]);
        if (client_id != NULL)
        {
            cJSON_AddStringToObject(broadcast, "clientId", client_id);
            data_tmp = cJSON_Print(broadcast);
            cJSON_DeleteItemFromObject(broadcast, "clientId");
            sprintf(topic, "F/%s/%d/transparent", client_id, pkt->socket);
            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;
}

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

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

    list_for_each_entry_safe(pkt, tmp, fwd_list, list)
    {
        _broadcast_to_mqtt(pkt);
        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();
    }
}