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

#include "dy_socket_server_session.h"
#include "lnxall_list.h"
#include "dy_utils/dy_common.h"

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

    while (1)
    {
        tmp = sendto(client_fd, 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 socket_connect_check(socket_server_session_t *session, int socket_fd)
{
	socket_connect_node_t *connect_node = NULL;
	list_for_each_entry(connect_node, &session->connect_list, list)
	{
		if(socket_fd == connect_node->socket_fd)
		{
			return 1;
		}
	}

	return 0;
}

static ssize_t sound_send(socket_server_session_t *session, int sockfd, const char *buffer, size_t buflen)
{
    ssize_t tmp = 0;
    size_t total = buflen;
    const char *p = buffer;

    while (1)
    {
    	if(socket_connect_check(session, sockfd) == 0)
			return 0;
		
		if(total > 40960)
			tmp = send(sockfd, p, 40960, 0);
		else
			tmp = send(sockfd, p, total, 0);
		
        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_server_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, node->client_fd, node->buf, node->buf_len);
        list_del(&node->list);
        free(node);
    }

}

static int _do_snd(socket_server_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(LOG_DEBUG, "send socket data client_fd %d len:%d", node->client_fd,node->buf_len);
        if (session->type == SOCKET_TCP)
        {
            sound_send(session, node->client_fd, node->buf, node->buf_len);
			session->handle_data_send_status(session, node->client_fd, 0, node->user_param);
        }
        else
            socket_udp_send(session, node->client_fd, node->buf, node->buf_len);
        list_del(&node->list);
        free(node);
    }

}

static int socket_server_to_loop(socket_server_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->op_code = 0;
    pthread_mutex_unlock(&session->sock_lock);

    if (need_snd == 1)
    {
        _do_snd(session, &tmp_snd_list);
    }

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

    return 0;

}


void *socket_server_loop(void *param)
{
    socket_server_session_t *session = (socket_server_session_t *)param;
    while (1)
    {
        socket_server_to_loop(session);
    }
}