//
// Created by Administrator on 2015/5/24.
//

#ifdef __ESVR_LINUX__
#include <sys/epoll.h>
#endif

#include "es_log.h"
#include "es_socket.h"
#include <sys/epoll.h>

namespace esvr {
void IOManager::main_loop() {
  m_io_type = "epoll";
  const int MAX_FDS = 100000;
  int epoll_fd = epoll_create(MAX_FDS);
  struct epoll_event event;
  struct epoll_event *events = NULL;
  event.data.fd = m_serverfd;
  event.events = EPOLLIN | EPOLLET;
  epoll_ctl(epoll_fd, EPOLL_CTL_ADD, m_serverfd, &event);
  events = new epoll_event[MAX_FDS];
  while (m_start) {
    auto queue = get_to_close_queue();
    socket_t *pfd;
    while ((pfd = queue->pop()) != NULL) {
      event.events = EPOLLIN;
      event.data.fd = *pfd;
      epoll_ctl(epoll_fd, EPOLL_CTL_DEL, *pfd, &event);
      es_close(*pfd);
      delete_fd(*pfd);
      // println("force close fd: function:", __FUNCTION__, ", line:", __LINE__,
      // ", fd:", *pfd, ", errno:", errno, ", connections:",
      // m_active_clients.size());
      // event dispatch
      ClosedEvent event{*pfd};
      EventBus::get_instance().fire_event(&event);
    }
    before_loop_once();
    int ret = epoll_wait(epoll_fd, events, MAX_FDS, m_wait_millisecond);
    after_loop_once();
    if (ret < 0) {
      // print("epoll_wait  error : function:", __FUNCTION__, ", line:",
      // __LINE__, ", errno:", errno, ", ret:", ret, "\n");
      m_start = false;
      break;
    } else if (ret == 0) {
      on_idle();
      continue;
    }
    for (int i = 0; i < ret; ++i) {
      if (events[i].data.fd == m_serverfd) {
        while (1) {
          socket_t cli_fd = accept(m_serverfd, NULL, NULL);
          if (cli_fd > 0) {
            event.events = EPOLLIN;
            event.data.fd = cli_fd;
            epoll_ctl(epoll_fd, EPOLL_CTL_ADD, cli_fd, &event);
            new_fd(cli_fd);
            ConnectedEvent event{cli_fd};
            EventBus::get_instance().fire_event(&event);
          } else {
            // print("accept error : function:", __FUNCTION__, ", line:",
            // __LINE__, ", errno:", errno, ", ret:", ret, "\n");
            break;
          }
        }
      } else {
        char *buffer = NULL;
        size_t buffer_size = 0;
        writable_buffer(events[i].data.fd, buffer, buffer_size);
        LOG_DEBUG("writable buffer size:", buffer_size);
        if (buffer == NULL || buffer_size <= 0) {
          LOG_ERROR("writable buffer for socket ", events[i].data.fd,
                    " is null!");
          continue;
        }
        int len;
        if ((len = es_recv(events[i].data.fd, buffer, buffer_size)) <= 0) {
          delete_fd(events[i].data.fd);
          // println("recv error : function:", __FUNCTION__, ", line:",
          // __LINE__, ", errno:", errno, ", ret:", ret,", connections:",
          // m_active_clients.size());
          es_close(events[i].data.fd);
          epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, &events[i]);
          ClosedEvent event{events[i].data.fd};
          EventBus::get_instance().fire_event(&event);
          events[i].data.fd = -1;
        } else {
          increase_readable(events[i].data.fd, static_cast<size_t>(len));
          ReadEvent event{events[i].data.fd, buffer, static_cast<size_t>(len)};
          EventBus::get_instance().fire_event(&event);
        }
      }
    }
  }
  delete[] events;
}
};