#include <sys/epoll.h>
#include <curl/curl.h>
#include <prtpool.h>
#include <prlock.h>
#include <prcvar.h>

#define EASY_COUNT 100

PRThreadPool *gPool;
CURLM *gMulti;
int gEpollFd;
PRLock *gIoQueueLock;
PRCondVar *gIoQueueWait;
PRLock *gMultiLock;
int gIoQueueCount;

struct ConnectionInfo {
  CURL *easy;
  curl_socket_t socket;
  int action;
  PRJob *job;
} gEasy[EASY_COUNT];

size_t curl_writer(void *ptr, size_t size, size_t nmemb, void *param) {
  return size*nmemb;
}

void setup_easy(CURL *easy) {
  curl_easy_setopt(easy, CURLOPT_TIMEOUT, 300L);
  curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT, 30L);
  curl_easy_setopt(easy, CURLOPT_FOLLOWLOCATION, 0L);
  curl_easy_setopt(easy, CURLOPT_VERBOSE, 1L);
  curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, curl_writer); //Ignore output from server
  curl_easy_setopt(easy, CURLOPT_SSL_VERIFYPEER, 0L);
  curl_easy_setopt(easy, CURLOPT_URL, "https://www.google.com/ncr");
  curl_easy_setopt(easy, CURLOPT_SSL_SESSIONID_CACHE, 0L);
}

void curl_calls(void *arg) {
  ConnectionInfo *info = (ConnectionInfo*) arg;
  int still_running;
  
  PR_Lock(gMultiLock);
  if (info->socket == CURL_SOCKET_BAD) { //Job is running for the first time.
    //Do some operation which takes a couple of seconds.
    PR_Sleep(PR_SecondsToInterval(1));
    curl_multi_add_handle(gMulti, info->easy);
  }
  curl_multi_socket_action(gMulti, info->socket, info->action, &still_running);
  PR_Unlock(gMultiLock);

  if (info->action & CURL_POLL_INOUT) {
    epoll_event event;
    event.events = EPOLLRDHUP | EPOLLONESHOT
      | (info->action & CURL_POLL_IN ? EPOLLIN : 0)
      | (info->action & CURL_POLL_OUT ? EPOLLOUT : 0);
    event.data.ptr = info;
    epoll_ctl(gEpollFd, EPOLL_CTL_ADD, info->socket, &event);
    PR_Lock(gIoQueueLock);
    gIoQueueCount++;
    info->job = PR_QueueJob_Timer(gPool, PR_SecondsToInterval(300), curl_calls, info, PR_FALSE);
    PR_NotifyCondVar(gIoQueueWait);
    PR_Unlock(gIoQueueLock);
  }
}

int socket_callback(CURL *easy, curl_socket_t socket, int action, void *multi_data, void *socket_data) {
  ConnectionInfo *info;
  curl_easy_getinfo(easy, CURLINFO_PRIVATE, &info);
  info->action = action;
  info->socket = socket;
  return 0;
}

void epoll_calls(void *arg) {
  PR_Lock(gIoQueueLock);
  while(true) {
    PR_WaitCondVar(gIoQueueWait, PR_INTERVAL_NO_TIMEOUT);
    epoll_event events[10];
    int n;
    do {
      PR_Unlock(gIoQueueLock);
      if ((n = epoll_wait(gEpollFd, events, 10, 3000)) > 0) {
        PR_Lock(gIoQueueLock);
        for (int i = 0; i < n; ++i) {
          ConnectionInfo *info = (ConnectionInfo*) events[i].data.ptr;
          PR_CancelJob(info->job);
          epoll_ctl(gEpollFd, EPOLL_CTL_DEL, info->socket, NULL);
          info->job = PR_QueueJob(gPool, curl_calls, info, PR_FALSE);
        }
        gIoQueueCount -= n;
        PR_Unlock(gIoQueueLock);
      }
      PR_Lock(gIoQueueLock);
    } while (gIoQueueCount > 0);
  }
}

int main() {
  curl_global_init(CURL_GLOBAL_ALL);
  gEpollFd = epoll_create(1);
  gIoQueueLock = PR_NewLock();
  gIoQueueWait = PR_NewCondVar(gIoQueueLock);
  gMultiLock = PR_NewLock();
  gPool = PR_CreateThreadPool(5, 5, 1024*1024);
  gMulti = curl_multi_init();
  curl_multi_setopt(gMulti, CURLMOPT_SOCKETFUNCTION, socket_callback);

  PR_QueueJob(gPool, epoll_calls, NULL, PR_FALSE);

  for (int i = 0; i < EASY_COUNT; ++i) {
    gEasy[i].socket = CURL_SOCKET_BAD;
    gEasy[i].easy = curl_easy_init();
    curl_easy_setopt(gEasy[i].easy, CURLOPT_PRIVATE, gEasy + i);
    setup_easy(gEasy[i].easy);

    gEasy[i].job = PR_QueueJob(gPool, curl_calls, gEasy + i, PR_FALSE);
  }
  PR_JoinThreadPool(gPool);
  return 0;
}