#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <queue>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <time.h>
#include <string.h>
#include <unordered_map>
#include <sys/un.h>
#include <arpa/inet.h>
#include <fcntl.h>

using namespace std;

struct server_data;

bool flag;
int socket_for_server;//socket_id -> socket_for_server
sockaddr_in address; //self_addr -> address
timespec tm;
unordered_map <int, server_data*> client_map;


struct server_data {
    int client_id;
    struct sockaddr_in address; //client_addr -> address
    char str_address[INET_ADDRSTRLEN]; // str_addr -> str_address
    queue< pair<char*, int> > q_process, q_send;
    bool flag;
    pthread_t thread_main, thread_reception, thread_process, thread_send;
    pthread_mutex_t mutex_q_process, mutex_q_send;
};

void *data_reception(void *args) { //функция приема данных из сокета; func_recv -> data_reception
    server_data *data = (server_data*)args;
    int bytes_read;
    printf("Reception thread started \n");
    fflush(stdout);
    while (data->flag) {
        char *buffer = (char*)malloc(64);
        while ((bytes_read = recv(data->client_id, buffer, 64, 0)) <= 0) {
            if (bytes_read == 0) {
                data->flag = false;
                pthread_exit(NULL);
            }
            if (!data->flag) {
                pthread_exit(NULL);
            }
            nanosleep(&tm, NULL);
        }
        pthread_mutex_lock(&(data->mutex_q_process));
        data->q_process.push(make_pair(buffer, bytes_read));
        pthread_mutex_unlock(&(data->mutex_q_process));
        printf("Client address:%s Received data: %s\n", data->str_address, buffer);
        fflush(stdout);
    }
}

void *data_process(void *args) {
    server_data *data = (server_data*)args;
    pair<char*, int> msg_pair;
    printf("Process thread started\n");
    fflush(stdout);
    while (data->flag) {
        while (true) {
            pthread_mutex_lock(&(data->mutex_q_process));
            if (!data->q_process.empty()) {
                pthread_mutex_unlock(&(data->mutex_q_process));
                break;
            }
            pthread_mutex_unlock(&(data->mutex_q_process));
            if (!data->flag) {
                pthread_exit(NULL);
            }
            nanosleep(&tm, NULL);
        }
        pthread_mutex_lock (&(data->mutex_q_process));
        msg_pair = data->q_process.front();
        data->q_process.pop();
        pthread_mutex_unlock(&(data->mutex_q_process));
        const time_t timer = time(NULL);
        char *new_buffer = (char*)malloc(64);
        sprintf(new_buffer, "(Send: %s Date: %s\n)", msg_pair.first, ctime(&timer));
        msg_pair.second = strlen(new_buffer);
        free(msg_pair.first);
        msg_pair.first = new_buffer;

        pthread_mutex_lock(&(data->mutex_q_send));
        data->q_send.push(msg_pair);
        pthread_mutex_unlock(&(data->mutex_q_send));

        printf("Client address:%s Send to client: %s\n", data->str_address, new_buffer);

        fflush(stdout);
    }
    printf("Process thread ended\n");
}

void *data_send(void *args) {
    server_data *data = (server_data*)args;
    pair <char*, int> msg_pair;
    int bytes_send;
    printf("Sending thread started\n");
    fflush(stdout);
    while (data->flag) {
        while (true) {
            pthread_mutex_lock(&(data->mutex_q_send));
            if (!data->q_send.empty()) {
                pthread_mutex_unlock(&(data->mutex_q_send));
                break;
            }
            pthread_mutex_unlock(&(data->mutex_q_send));
            if (!data->flag) {
                printf("Sending thread close\n");
                pthread_exit(NULL);
            }
            nanosleep(&tm, NULL);
        }
        pthread_mutex_lock(&(data->mutex_q_send));
        msg_pair = data->q_send.front();
        data->q_send.pop();
        pthread_mutex_unlock(&(data->mutex_q_send));
        bytes_send = send(data->client_id, msg_pair.first, msg_pair.second, 0);

        if (bytes_send <= 0) {
            data->flag = false;
            pthread_exit(NULL);
        }

        fflush(stdout);
    }
    printf("Sending thread close");
}

void *func_create_client_thread(void *args) {
    server_data *data = (server_data*)args;

    inet_ntop(AF_INET, &(data->address.sin_addr), data->str_address, INET_ADDRSTRLEN);
    data->flag = true;
    client_map[data->client_id] = data;

    printf("Client thread started\n");

    pthread_mutex_init(&(data->mutex_q_process), NULL);
    pthread_mutex_init(&(data->mutex_q_send), NULL);

    pthread_create(&(data->thread_reception), NULL, &data_reception, (void*)data);
    pthread_create(&(data->thread_process), NULL, &data_process, (void*)data);
    pthread_create(&(data->thread_send), NULL, &data_send, (void*)data);

    pthread_join(data->thread_reception, NULL);
    pthread_join(data->thread_process, NULL);
    pthread_join(data->thread_send, NULL);

    pthread_mutex_destroy(&(data->mutex_q_process));
    pthread_mutex_destroy(&(data->mutex_q_send));

    shutdown(data->client_id, SHUT_WR);
    close(data->client_id);

    printf("Client thread close\n");
}

void *data_listen(void *args) {
    printf("Listen thread started\n");
    int client_addr_sz = sizeof(struct sockaddr_in);
    while (flag) {
        server_data *data = new server_data;
        while ((data->client_id = accept(socket_for_server, (sockaddr*)&(data->address), (socklen_t*)&client_addr_sz)) == -1) {
            if (!flag) {
                printf("Listen thread close\n");
                pthread_exit(NULL);
            }
            nanosleep(&tm, NULL);
        }
        int flags = fcntl(data->client_id, F_GETFL);
        fcntl(data->client_id, F_SETFL, flags | O_NONBLOCK);

        pthread_create(&(data->thread_main), NULL, &func_create_client_thread, (void*)data);
    }
    printf("Listen thread close\n");
}

int main()
{
    flag = true;
    tm.tv_sec = 0;
    tm.tv_nsec = 10000000;

    address.sin_family = AF_INET;
    address.sin_port = htons(31337);
    address.sin_addr.s_addr = htonl(INADDR_ANY);
    socket_for_server = socket(AF_INET, SOCK_STREAM, 0);
    int flags = fcntl(socket_for_server, F_GETFL);
    fcntl(socket_for_server, F_SETFL, flags | O_NONBLOCK);

    int yes = 1;
    setsockopt(socket_for_server, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
    int res = bind(socket_for_server, (struct sockaddr *)&address, sizeof(struct sockaddr));
    res = listen(socket_for_server, 10);
    pthread_t thread_listen; //thr_listen -> thread_listen
    pthread_create(&thread_listen, NULL, &data_listen, NULL);
    getchar();
    flag = false;
    pthread_join(thread_listen, NULL);
    for( const auto &t_data : client_map ) {
        t_data.second->flag = false;
        pthread_join(t_data.second->thread_main, NULL);
    }
    shutdown(socket_for_server, SHUT_RDWR);
    close(socket_for_server);
    printf("Server shut down\n");
    return 0;
}