#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>

#include "dy_socket_session.h"
#include "lnxall_list.h"
#include "dy_common.h"

static ssize_t socket_udp_send(socket_session_t *session, const char *buffer, size_t buflen)
{
    ssize_t tmp;
    size_t total = buflen;
    const char *p = buffer;

    while (1)
    {
        tmp = sendto(session->sockfd, p, total, 0, (struct sockaddr *)&session->server_addr, session->socklen);
        if (tmp < 0)
        {
            // 当send收到信号时,可以继续写,但这里返回-1.
            if (errno == EINTR)
            {
                continue;
            }

            // 当socket是非阻塞时,如返回此错误,表示写缓冲队列已满,
            // 在这里做延时后再重试.
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                usleep(1000);
                continue;
            }

            return -1;
        }

        if ((size_t)tmp == total)
        {
            return buflen;
        }

        total -= tmp;
        p += tmp;
    }

    return tmp;
}

static int _do_rcv(socket_session_t *session, struct list_head *rcv_list)
{
    socket_data_node_t *node = NULL;
    socket_data_node_t *tmp = NULL;

    if (list_empty(rcv_list))
    {
        return 0;
    }
    list_for_each_entry_safe(node, tmp, rcv_list, list)
    {
        session->handle_recv_msg(session->user_data, node->buf, node->buf_len);
        list_del(&node->list);
        free(node);
    }
    return 0;
}

static int _do_snd(socket_session_t *session, struct list_head *snd_list)
{
    socket_data_node_t *node = NULL;
    socket_data_node_t *tmp = NULL;

    if (list_empty(snd_list))
    {
        return 0;
    }
    list_for_each_entry_safe(node, tmp, snd_list, list)
    {
        dy_syslog_hex(LOG_DEBUG, node->buf, node->buf_len, "send socket data len:%d", node->buf_len);
        if (session->type == SOCKET_TCP)
            socket_send(session->sockfd, node->buf, node->buf_len);
        else
            socket_udp_send(session, node->buf, node->buf_len);
        list_del(&node->list);
        free(node);
    }
    return 0;
}

static int socket_client_to_loop(socket_session_t *session)
{
    struct list_head tmp_rcv_list = LIST_HEAD_INIT(tmp_rcv_list);
    struct list_head tmp_snd_list = LIST_HEAD_INIT(tmp_snd_list);
    int need_snd = 0;
    int need_rcv = 0;

    pthread_mutex_lock(&session->sock_lock);
    while (session->op_code == 0)
    {
        lnxall_condvar_timedwait(&session->sock_cond, &session->sock_lock, 5000);
    }

    if (session->op_code & SOCKET_DO_RCV_MASK)
    {
        // real received packets
        if (!list_empty(&session->rcv_list))
        {
            list_move_tail_list(&session->rcv_list, &tmp_rcv_list);
            need_rcv = 1;
        }
    }
    if (session->op_code & SOCKET_DO_SND_MASK)
    {
        // real received packets
        if (!list_empty(&session->snd_list))
        {
            list_move_tail_list(&session->snd_list, &tmp_snd_list);
            need_snd = 1;
        }
        session->send_cnt = 0;
    }
    session->op_code = 0;
    pthread_mutex_unlock(&session->sock_lock);

    if (need_snd == 1)
    {
        if (session->state != SOCKET_CONNECTED)
        {
            while (session->state != SOCKET_CONNECTED)
            {
                sleep(1);
            }
        }
        _do_snd(session, &tmp_snd_list);
    }

    if (need_rcv == 1)
    {
        _do_rcv(session, &tmp_rcv_list);
    }

    return 0;
}

void *socket_client_loop(void *param)
{
    socket_session_t *session = (socket_session_t *)param;
    while (1)
    {
        socket_client_to_loop(session);
    }
}

int dy_socket_send_buf(socket_type_e type, int socket, unsigned char *buf, int len)
{
    if (type == SOCKET_TCP)
        socket_send(socket, (const char *) buf, len);
    // TODO UDP
    return 0;
}

int dy_socket_recv(socket_type_e type, int socket, unsigned char *buf, int max_len, int timeout)
{
    fd_set rfds;
    struct timeval tv;
    int retval, maxfd = -1;

    FD_ZERO(&rfds);
    maxfd = 0;
    FD_SET(socket, &rfds);
    if (socket > maxfd)
    {
        maxfd = socket;
    }

    tv.tv_sec = timeout;
    tv.tv_usec = 0;

    retval = select(maxfd + 1, &rfds, NULL, NULL, &tv);
    if (retval == -1)
    {
        dy_syslog(LOG_ERR, "select is error! %s", strerror(errno));
        return -1;
    }
    else if (retval > 0)
    {
        if (FD_ISSET(socket, &rfds))
        {
            int len = 0;

            if (type == SOCKET_TCP)
                len = recv(socket, buf, max_len, 0);

            // TODO UDP

            return len;
        }
    }
    return 0;
}