#include "stdafx.h"
#include "LogServer.h"
#include <memory>

namespace smackbio{
	namespace logserver{

		const auto KeepAliveTimeout = std::chrono::seconds(30);

		LogServer::LogServer(OnMessageCallback cb, void* cb_context, int port, bool enable_v6)
			: listener_v4_(io_serv_)
			, listener_v6_(io_serv_)
			, keepalive_timer_(io_serv_)
		{
			TerminalList::on_message = cb;
			TerminalList::on_message_context = cb_context;

			// Create TCP listener
			listener_v4_.open(tcp::v4());
			listener_v4_.bind(tcp::endpoint(tcp::v4(), port));
			listener_v4_.listen();
			begin_accept(listener_v4_);

			if (enable_v6)
			{
				listener_v6_.open(tcp::v6());
				listener_v6_.bind(tcp::endpoint(tcp::v6(), port));
				listener_v6_.listen();
				begin_accept(listener_v6_);
			}

			// Begin keepalive timer
			begin_keepalive();

			// Create worker threads
			SYSTEM_INFO sys_info;
			GetSystemInfo(&sys_info);

			workers_.resize(sys_info.dwNumberOfProcessors);
			for (auto i = workers_.begin(); i != workers_.end(); ++i)
				*i = std::thread([this]() { io_serv_.run(); });
		}

		LogServer::~LogServer()
		{
			{
				std::unique_lock<std::mutex> lock(TerminalList::mutex);
				stopping_ = true;
			}

			// Stop listeners
			listener_v4_.close();
			listener_v6_.close();

			// Stop keepalive timer
			keepalive_timer_.cancel();

			// Close all terminals
			for_each_terminals([](const std::shared_ptr<Terminal>& p) { p->disconnect(); });

			// Join all worker threads
			for (auto i = workers_.begin(); i != workers_.end(); ++i)
				i->join();
		}

		template<typename F> void LogServer::for_each_terminals(F callback)
		{
			std::shared_ptr<Terminal> terminal;

			for (;;)
			{
				// Find next terminal
				std::shared_ptr<Terminal> next_terminal = nullptr;

				{
					std::unique_lock<std::mutex> lock(TerminalList::mutex);

					typename decltype(TerminalList::terminals)::iterator iter;
					if (terminal)
					{
						iter = TerminalList::terminals.iterator_to(*terminal);
						++iter;
					}
					else
						iter = TerminalList::terminals.begin();

					while (iter != TerminalList::terminals.end())
					{
						next_terminal = iter->try_hold();
						if (next_terminal)
							break;
					}
				}

				if (!next_terminal)
					break;

				// Deal with the terminal
				terminal = std::move(next_terminal);
				callback(terminal);
			}
		}

		void LogServer::begin_accept(tcp::acceptor& acceptor)
		{
			auto client = std::make_shared<tcp::socket>(io_serv_);

			std::unique_lock<std::mutex> lock(TerminalList::mutex);
			if (stopping_)
				return;

			acceptor.async_accept(*client, [this, client, &acceptor](const error_code& ec)
			{
				if (ec)
					return;

				try
				{
					add_client(client);
				}
				catch (std::exception&)
				{
				}

				try
				{
					begin_accept(acceptor);
				}
				catch (std::exception&)
				{
				}
			});
		}

		void LogServer::add_client(const std::shared_ptr<tcp::socket>& client)
		{
			Terminal::create(this, client);
		}

		void LogServer::begin_keepalive()
		{
			std::unique_lock<std::mutex> lock(TerminalList::mutex);
			if (stopping_)
				return;

			keepalive_timer_.expires_from_now(KeepAliveTimeout / 3);
			keepalive_timer_.async_wait([this](const error_code& ec)
			{
				if (ec)
					return;

				try
				{
					for_each_terminals([&](const std::shared_ptr<Terminal>& t)
					{
						disconnect_inactive(t);
					});

					begin_keepalive();
				}
				catch (std::exception&)
				{
				}
			});
		}

		void LogServer::disconnect_inactive(const std::shared_ptr<Terminal>& terminal)
		{
			if (terminal->inactive_time() > KeepAliveTimeout)
				terminal->disconnect();
		}
	}
}
