#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <time.h>
#include <netinet/tcp.h>

#include "common.h"

#define DEBUG 1
/*
 * Add socket received packete data into LIST
 * */
static int add_pkt_data(int socket, uint8_t *buf, int len)
{
    socket_data_node_t *pkt_data = NULL;
    int send_signal = 0;

    pkt_data = malloc(sizeof(socket_data_node_t) + len);
    pkt_data->socket = socket;
    pkt_data->payload_len = len;
    memcpy(pkt_data->payload, buf, len);

    // 将数据加入LIST
    pthread_mutex_lock(&socket_mqtt_var.socket_data_lock);
    list_add_tail(&pkt_data->list, &socket_mqtt_var.socket_rcv_head);
    socket_mqtt_var.socket_received++;

    // 累积到一定数量的packet，再统一唤醒
    if (socket_mqtt_var.socket_received > NAPI_QUEUE_LEN)
    {
        /*
         * send the signal after unlock
         * because the waitting thread need the lock too,
         * once received the signal, it will be waked up,
         * but it will sleep soon when calling the lock.
         * it will waste the CPU resource when system is busy.
         *
         * send the signal after the unlock can avoid the issue.
         */
        send_signal = 1;
    }
    pthread_mutex_unlock(&socket_mqtt_var.socket_data_lock);

    if (send_signal == 1)
    {
        pthread_cond_signal(&socket_mqtt_var.socket_recv_cond);
    }
    return 0;
}

/*
setnonblocking - 设置句柄为非阻塞方式
*/
static int setnonblocking(int sockfd)
{
    if (fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0) | O_NONBLOCK) == -1)
    {
        return -1;
    }
    return 0;
}

/*
handle_message - 处理每个 socket 上的消息收发
*/
static int handle_message(int socket)
{
    uint8_t buf[MAXBUF + 1];
    int len;

    /* 开始处理每个新连接上的数据收发 */
    bzero(buf, MAXBUF + 1);

    /* 接收客户端的消息 */
    while (1)
    {
        len = recv(socket, buf, MAXBUF, 0);
        if (len > 0)
        {
            add_pkt_data(socket, buf, len);
            if (len < MAXBUF)
            {
                break;
            }
        }
        else
        {
            if (len < 0)
            {
                dy_syslog(LOG_INFO, "receive failed, %d:'%s'", errno, strerror(errno));
                if ((errno != EAGAIN) && (errno != EWOULDBLOCK) && (errno != EINTR)) //TCP 错误
                {
                    dy_syslog(LOG_INFO, "TCP connection error");
                    return -2;
                }
            }
            else
            {
                dy_syslog(LOG_INFO, "client %d disconnect!", socket);
                return -2;
            }
            return -1;
        }
    }
    /* 处理每个新连接上的数据收发结束 */
    return len;
}

/* Set TCP keep alive option to detect dead peers. The interval option
 * is only used for Linux as we are using Linux-specific APIs to set
 * the probe send time, interval, and count. */
int set_keep_alive(int fd)
{
    int val = 1;
    int interval = 300;

    //开启keepalive机制
    if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val)) == -1)
    {
        dy_syslog(LOG_ERR, "setsockopt SO_KEEPALIVE: %s", strerror(errno));
        return -1;
    }

    /* Default settings are more or less garbage, with the keepalive time
     * set to 7200 by default on Linux. Modify settings to make the feature
     * actually useful. */

    /* Send first probe after interval. */
    val = interval;
    if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &val, sizeof(val)) < 0) {
        dy_syslog(LOG_ERR, "setsockopt TCP_KEEPIDLE: %s\n", strerror(errno));
        return -1;
    }

    /* Send next probes after the specified interval. Note that we set the
     * delay as interval / 3, as we send three probes before detecting
     * an error (see the next setsockopt call). */
    val = interval/3;
    if (val == 0) val = 1;
    if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &val, sizeof(val)) < 0) {
        dy_syslog(LOG_ERR, "setsockopt TCP_KEEPINTVL: %s\n", strerror(errno));
        return -1;
    }

    /* Consider the socket in error state after three we send three ACK
     * probes without getting a reply. */
    val = 3;
    if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &val, sizeof(val)) < 0) {
        dy_syslog(LOG_ERR, "setsockopt TCP_KEEPCNT: %s\n", strerror(errno));
        return -1;
    }

    dy_syslog(LOG_DEBUG, "set keep alive success");

    return 0;
}

/*
   假如发送端流量大于接收端的流量
   (意思是epoll所在的程序读比转发的socket要快),
   由于是非阻塞的socket,那么send()函数虽然返回,
   但实际缓冲区的数据并未真正发给接收端,
   这样不断的读和发，
   当缓冲区满后会产生EAGAIN错误(参考man send),同时,
   不理会这次请求发送的数据.所以,
   需要封装socket_send()的函数用来处理这种情况,
   该函数会尽量将数据写完再返回，返回-1表示出错。
   在socket_send()内部,当写缓冲已满(send()返回-1,且errno为EAGAIN),
   那么会等待后再重试.这种方式并不很完美,
   在理论上可能会长时间的阻塞在socket_send()内部,
   但暂没有更好的办法.
 */
ssize_t socket_send(int sockfd, const char *buffer, size_t buflen)
{
    ssize_t tmp;
    size_t total = buflen;
    const char *p = buffer;

    while (1)
    {
        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;
}

void *start_tcp_server(void *param)
{
    int listener, new_fd, nfds, n, ret;
    struct epoll_event ev;
    int epfd, curfds;
    socklen_t len;
    struct sockaddr_in my_addr, their_addr;
    struct epoll_event *events = NULL;
    struct rlimit rt;
    int port = *(int *)param;
    int opt = 1;
    if (port <= 0 || port > 65535)
    {
        dy_syslog(LOG_INFO, "port error:%d", port);
        return (void *) - 1;
    }
    /* 设置每个进程允许打开的最大文件数 */
    events = malloc(MAXEPOLLSIZE * sizeof(struct epoll_event));
    if (events == NULL)
    {
        dy_syslog(LOG_INFO, "malloc failed");
        return (void *) - 1;
    }
    /* 开启 socket 监听 */
    if ((listener = socket(PF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0)) == -1)
    {
        dy_syslog(LOG_INFO, "socket");
        free(events);
        return (void *) - 1;
    }
    else
    {
        dy_syslog(LOG_INFO, "socket created");
    }
    /*设置socket属性，端口可以重用*/
    if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1)
    {
        dy_syslog(LOG_ERR, "set socket option SO_REUSEADDR fail, errno:%d:%s", errno, strerror(errno));
        free(events);
        close(listener);
        return (void *) - 1;
    }
    else
    {
        dy_syslog(LOG_INFO, "set socket option SO_REUSEADDR success");
    }
    #ifdef SO_REUSEPORT
    if (setsockopt(listener, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) == -1)
    {
        dy_syslog(LOG_ERR, "set socket option SO_REUSEPORT fail, errno:%d:%s", errno, strerror(errno));
        free(events);
        close(listener);
        return (void *) - 1;
    }
    else
    {
        dy_syslog(LOG_INFO, "set socket option SO_REUSEPORT success");
    }
    #endif
    /*设置socket为非阻塞模式*/
    setnonblocking(listener);
    bzero(&my_addr, sizeof(my_addr));
    my_addr.sin_family = PF_INET;
    my_addr.sin_port = htons(port);
    my_addr.sin_addr.s_addr = INADDR_ANY;
    if (bind(listener, (struct sockaddr *) &my_addr, sizeof(struct sockaddr)) == -1)
    {
        dy_syslog(LOG_INFO, "bind");
        free(events);
        close(listener);
        return (void *) - 1;
    }
    else
    {
        dy_syslog(LOG_INFO, "IP and %d bind success", port);
    }
    if (listen(listener, LISTEN_BACKLOG) == -1)
    {
        dy_syslog(LOG_INFO, "listen");
        free(events);
        close(listener);
        return (void *) - 1;
    }
    else
    {
        dy_syslog(LOG_INFO, "start listenning");
    }
    /* 创建 epoll 句柄，把监听 socket 加入到 epoll 集合里 */
    epfd = epoll_create(MAXEPOLLSIZE);
    len = sizeof(struct sockaddr_in);
    ev.events = EPOLLIN;
    ev.data.fd = listener;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listener, &ev) < 0)
    {
        dy_syslog(LOG_INFO, "epoll set insertion error: fd=%d", listener);
        free(events);
        close(listener);
        close(epfd);
        return (void *) - 1;
    }
    else
    {
        dy_syslog(LOG_INFO, "listen socket add into epoll");
    }
    curfds = 1;
    while (1)
    {
        /* 等待有事件发生 */
        nfds = epoll_wait(epfd, events, MAXEPOLLSIZE, 10 * 1000);
        if (nfds == -1)
        {
            dy_syslog(LOG_INFO, "epoll_wait");
            break;
        }

        /* 处理所有事件 */
        for (n = 0; n < nfds; ++n)
        {
            if (events[n].data.fd == listener)
            {
                dy_syslog(LOG_DEBUG, "listener receive event");
                new_fd = accept4(listener, (struct sockaddr *) &their_addr, &len, SOCK_CLOEXEC);
                if (new_fd < 0)
                {
                    dy_syslog(LOG_INFO, "accept");
                    continue;
                }
                else
                {
                    dy_syslog(LOG_INFO, "connection request from %s:%d, allocated socket is:%d",
                              inet_ntoa(their_addr.sin_addr), ntohs(their_addr.sin_port), new_fd);
                }
                setnonblocking(new_fd);
                set_keep_alive(new_fd);
                ev.events = EPOLLIN | EPOLLRDHUP | EPOLLERR | EPOLLHUP;
                ev.data.fd = new_fd;
                if (epoll_ctl(epfd, EPOLL_CTL_ADD, new_fd, &ev) < 0)
                {
                    dy_syslog(LOG_INFO, "add socket '%d' into epoll failed%s",
                              new_fd, strerror(errno));
                    free(events);
                    close(listener);
                    close(epfd);
                    return (void *) - 1;
                }
                curfds++;
            }
            else
            {
                if (events[n].events & (EPOLLERR | EPOLLRDHUP | EPOLLHUP))
                {
                    /* error on the fd, close the socket */
                    dy_syslog(LOG_INFO, "socket:%d leave", events[n].data.fd);
                    ret = -2;
                }
                else
                {
                    ret = handle_message(events[n].data.fd);
                }

                if (ret == -2)
                {
                    if (epoll_ctl(epfd, EPOLL_CTL_DEL, events[n].data.fd,
                                  &ev) < 0)
                    {
                        dy_syslog(LOG_INFO, "remove socket '%d' from epoll failed%s",
                                  events[n].data.fd, strerror(errno));
                    }
                    curfds--;
                    close(events[n].data.fd);
                    delete_session_by_socket(events[n].data.fd);
                }
            }
        }
    }
    free(events);
    close(listener);
    close(epfd);
    return (void *)0;
}