#include "stdafx.h"
#include "Terminal.h"
#include <cstring>
#include <algorithm>

namespace smackbio{
	namespace logserver{

		Terminal::Terminal()
		{
		}

		Terminal::~Terminal()
		{
			if (list_)
			{
				std::unique_lock<std::mutex> lock(list_->mutex);
				list_->terminals.erase(list_->terminals.iterator_to(*this));
				list_ = nullptr;
			}
		}

		std::shared_ptr<Terminal> Terminal::create(TerminalList* list, const std::shared_ptr<tcp::socket>& sock)
		{
			auto p = std::shared_ptr<Terminal>(new Terminal, [](Terminal* p) { delete p; });
			p->weak_self_ = p;

			p->socket_ = sock;
			p->begin_receive();

			{
				std::unique_lock<std::mutex> lock(list->mutex);
				list->terminals.push_back(*p);

				p->last_active_ = system_clock::now();
				p->list_ = list;
			}

			return p;
		}

		void Terminal::begin_receive()
		{
			if (recv_pos_ == MaxMessageSize)
			{
				// One message is too long.
				return;
			}

			auto self = hold();
			socket_->async_receive(asio::buffer(recv_buf_ + recv_pos_, MaxMessageSize - recv_pos_), [self](const error_code& ec, size_t len)
			{
				if (ec || len == 0)
					return;

				try
				{
					{
						std::unique_lock<std::mutex> lock(self->list_->mutex);
						self->last_active_ = system_clock::now();
					}

					self->recv_pos_ += len;
					while (self->process_messages())
						;
				}
				catch (std::exception&)
				{ }
			});
		}

		bool Terminal::process_messages()
		{
			auto null_pos = std::char_traits<unsigned char>::find(
				recv_buf_ + find_null_pos_,
				recv_pos_ - find_null_pos_,
				0);
			if (!null_pos)
			{
				find_null_pos_ = recv_pos_;
				begin_receive();
				return false;
			}

			if (null_pos == recv_buf_)
			{
				while (null_pos != recv_buf_ + recv_pos_ && *null_pos == 0)
					++null_pos;

				size_t consumed = null_pos - recv_buf_;
				std::copy(null_pos, (const unsigned char*)recv_buf_ + recv_pos_, recv_buf_);

				recv_pos_ -= consumed;
				find_null_pos_ = 0;

				return true;
			}

			auto reply_len = (*list_->on_message)(
				list_->on_message_context,
				(const char*)recv_buf_, null_pos - recv_buf_,
				(char*)send_buf_, MaxMessageSize - 1);

			// Move remainder to the first
			size_t consumed = null_pos + 1 - recv_buf_;
			std::copy(null_pos + 1, (const unsigned char*)recv_buf_ + recv_pos_, recv_buf_);
			
			recv_pos_ -= consumed;
			find_null_pos_ = 0;

			// Send reply
			if (!reply_len)
				return true;

			// Begin asynchronous send
			send_buf_[reply_len] = 0;

			auto self = hold();
			asio::async_write(*socket_, asio::buffer(send_buf_, reply_len + 1), [self, reply_len](const error_code& ec, size_t len)
			{
				try
				{
					if (ec || len != reply_len + 1)
						return;

					while (self->process_messages())
						;
				}
				catch (std::exception&)
				{ }
			});
			return true;
		}

		void Terminal::disconnect()
		{
			socket_->close();
		}

		system_clock::duration Terminal::inactive_time() const
		{
			std::unique_lock<std::mutex> lock(list_->mutex);
			return system_clock::now() - last_active_;
		}
	}
}
