#include <atomic>
#include <chrono>
#include <ctime>
#include <iomanip>
#include <iostream>
#include <memory>
#include <sstream>
#include <string>
#include <thread>

#include <boost/asio.hpp>
#include <boost/asio/system_timer.hpp>

#include <botan/botan.h>

namespace asio = boost::asio;
using boost::asio::ip::tcp;

#define PORT 9090

std::atomic<uint64_t> stat_conn;
std::atomic<uint64_t> stat_pkt;

class client_connection : public std::enable_shared_from_this<client_connection> {
public:
  static auto create(asio::io_service &service) {
    return std::make_shared<client_connection>(service);
  }

  tcp::socket &socket() { return socket_; }

  client_connection(asio::io_service &service)
      : strand_(service),
        socket_(service),
        pipe_(new Botan::Hash_Filter("SHA-256"), new Botan::Hex_Encoder,
              new Botan::DataSink_Stream(sha256_)) {}

  void run() {
    asio::async_read(
        socket_, asio::buffer(data_),
        strand_.wrap([self = shared_from_this()](const boost::system::error_code &err,
                                                 size_t n) { self->read_cb(err, n); }));
  }

private:
  void read_cb(const boost::system::error_code &error, size_t) {
    if (error)
      return;

    stat_pkt++;

#if 1
    Botan::DataSource_Memory in(data_.data(), data_.size());
    pipe_.process_msg(in);
#else
    sha256_ << "0123456789012345678901234567890123456789012345678901234567890123";
#endif
    asio::async_write(
        socket_, asio::buffer(sha256_.str()),
        strand_.wrap([self = shared_from_this()](const boost::system::error_code &err,
                                                 size_t n) { self->write_cb(err, n); }));
    run();
  }
  void write_cb(const boost::system::error_code &, size_t) { sha256_.str(""); }

  asio::io_service::strand strand_;
  tcp::socket socket_;
  Botan::Pipe pipe_;
  std::array<unsigned char, 512> data_;
  std::stringstream sha256_;
};

class server {
public:
  server(asio::io_service &service, unsigned short port, unsigned nthr)
      : service_{service},
        acceptor_{service, tcp::endpoint(tcp::v4(), port)},
        timer_{service},
        nthr_{nthr},
        pkt_{0} {
    acceptor_.set_option(tcp::acceptor::reuse_address(true));
    start_accept();
    start_time_ = std::chrono::system_clock::now();
    start_timer();
  }

  void run() {
    std::vector<std::thread> thr;
    for (unsigned i = 0; i < nthr_; ++i) {
      auto t = std::thread([this] { service_.run(); });
      thr.push_back(std::move(t));
    }

    for (auto &t : thr) {
      t.join();
    }
  }

  asio::io_service &service() { return service_; }

private:
  void start_accept() {
    auto conn = client_connection::create(acceptor_.get_io_service());

    acceptor_.async_accept(conn->socket(), [=](const boost::system::error_code &err) {
      accept_cb(conn, err);
    });
  }

  void start_timer() {
    timer_.expires_from_now(std::chrono::seconds(5));
    timer_.async_wait([this](const boost::system::error_code &) { tick(); });
  }

  void accept_cb(std::shared_ptr<client_connection> conn,
                 const boost::system::error_code &error) {
    if (!error) {
      stat_conn++;
      conn->run();
    }
    start_accept();
  }

  void tick() {
    auto delta = std::chrono::duration_cast<std::chrono::seconds>(
                     std::chrono::system_clock::now() - start_time_)
                     .count();
    std::cout << "delta: " << delta << "\n";
    std::cout << "conn: " << stat_conn << ", pkt: " << stat_pkt << ", time: " << delta
              << "\n";
    std::cout << stat_pkt / delta << " pkt/s" << std::endl;

    start_timer();
  }

  asio::io_service &service_;
  tcp::acceptor acceptor_;
  asio::system_timer timer_;
  unsigned nthr_;
  uint64_t pkt_;
  std::chrono::system_clock::time_point start_time_;
};

#define NTHR 6

int main() {
  try {
    asio::io_service service;
    server(service, PORT, NTHR).run();
  } catch (std::exception &e) {
    std::cerr << e.what() << std::endl;
  }
  return 0;
}