#include "stdafx.h"
#include "../include/qtl_mysql.hpp"
#include <vector>
#include <thread>
#include <system_error>
#include <time.h>
#include <limits.h>
#include "../include/qtl_mysql_pool.hpp"

#if MARIADB_VERSION_ID < 0100000
#error "The program need mariadb version > 10.0"
#endif 

class simple_event_loop
{
public:
	simple_event_loop()
	{
		m_expired = 0;
		m_maxfd = 0;
		m_stoped = false;
		FD_ZERO(&m_readset);
		FD_ZERO(&m_writeset);
		FD_ZERO(&m_exceptset);
	}

	void reset()
	{
		m_stoped = false;
	}

	void run()
	{
		while (!m_stoped)
			do_once(nullptr);
	}

	void run_for(long seconds)
	{
		time_t expired, now;
		time(&now);
		expired = now + seconds;
		timeval tv = { seconds };
		do
		{
			do_once(&tv);
			time(&now);
			tv.tv_sec = expired - now;
			tv.tv_usec = 0;
		} while (!m_stoped && now < expired);
	}

	void stop()
	{
		m_stoped = true;
	}

private:
	class event_item : public qtl::event
	{
	public:
		event_item(simple_event_loop* loop, qtl::socket_type fd)
			: m_loop(loop), m_fd(fd), m_expired(LONG_MAX)
		{
		}

		virtual void set_io_handler(int flags, long timeout, std::function<void(int)>&& handler) NOEXCEPT override
		{
			if (timeout > 0)
			{
				time_t now;
				time(&now);
				if (m_expired > now + timeout)
					m_expired = now + timeout;
				m_loop->set(this, flags, m_expired);
			}
			else
			{
				m_expired = LONG_MAX;
				m_loop->set(this, flags, 0);
			}
			m_io_handler = std::forward<std::function<void(int)>>(handler);
		}
		virtual void remove() override
		{
			m_loop->remove(this);
		}
		virtual bool is_busying() override
		{
			return m_io_handler!=nullptr;
		}

		simple_event_loop* m_loop;
		qtl::socket_type m_fd;
		time_t m_expired;
		std::function<void(int)> m_io_handler;
	};
	std::vector<std::unique_ptr<event_item>> m_events;

	//implement for QTL
public:
	template<typename Connection>
	event_item* add(Connection* connection)
	{
		qtl::socket_type fd = connection->socket();
		event_item* result = new event_item(this, fd);
		std::unique_ptr<event_item> item(result);
		m_events.emplace_back(std::move(item));
		return result;
	}

	template<typename Handler>
	event_item* set_timeout(const timeval& timeout, Handler&& handler)
	{
		event_item* result = new event_item(this, 0);
		std::unique_ptr<event_item> item(result);
		::time(&result->m_expired);
		result->m_expired+=timeout.tv_sec;
		if(m_expired>result->m_expired)
			m_expired=result->m_expired;
		m_events.emplace_back(std::move(item));
		return result;
	}

public:
	void set(event_item* item, int flags, time_t expired)
	{
		if (item->m_fd + 1 > m_maxfd)
			m_maxfd = item->m_fd + 1;
		if (flags&qtl::event::ef_read)
			FD_SET(item->m_fd, &m_readset);
		if (flags&qtl::event::ef_write)
			FD_SET(item->m_fd, &m_writeset);
		if (flags&qtl::event::ef_exception)
			FD_SET(item->m_fd, &m_exceptset);
		if (expired > 0 && m_expired > expired)
			m_expired = expired;
	}

	void remove(event_item* item)
	{
		auto it = m_events.begin();
		while (it != m_events.end())
		{
			if (it->get() == item)
			{
				m_events.erase(it);
				return;
			}
			++it;
		}
	}

private:
	fd_set m_readset, m_writeset, m_exceptset;
	qtl::socket_type m_maxfd;
	int do_once(timeval* timeout)
	{
		fd_set rs = m_readset, ws = m_writeset, es = m_exceptset;
		timeval tv = { INT_MAX, 0 };
		time_t now;
		time(&now);
		if (timeout)
		{
			tv = *timeout;
			timeout = &tv;
		}

		long d = m_expired - now;
		if (d > 0 && tv.tv_sec > d)
		{
			tv.tv_sec = d;
			timeout = &tv;
		}

		qtl::socket_type maxfd = m_maxfd;
		FD_ZERO(&m_readset);
		FD_ZERO(&m_writeset);
		FD_ZERO(&m_exceptset);
		m_maxfd = 0;

		if (maxfd == 0)
		{
			if (timeout)
			{
				std::this_thread::sleep_for(std::chrono::microseconds(timeout->tv_sec*std::micro::den + timeout->tv_usec));
				check_timeout();
			}
			else
			{
				std::this_thread::sleep_for(std::chrono::seconds(1));
			}
			return 0;
		}
		else
		{
			int ret = select(maxfd, &rs, &ws, &es, timeout);
			if (ret > 0)
			{
				for (auto& item : m_events)
				{
					int flags = 0;
					if (FD_ISSET(item->m_fd, &rs))
						flags |= qtl::event::ef_read;
					if (FD_ISSET(item->m_fd, &ws))
						flags |= qtl::event::ef_write;
					if (FD_ISSET(item->m_fd, &es))
						flags |= qtl::event::ef_exception;
					if (flags && item->m_io_handler)
					{
						auto handler = std::move(item->m_io_handler);
						handler(flags);
					}
				}
			}
			else if (ret == 0)
			{
				time(&now);
				for (auto& item : m_events)
				{
					if (item->m_expired > 0 && item->m_io_handler && item->m_expired < now)
					{
						item->m_io_handler(qtl::event::ef_timeout);
						item->m_expired = 0;
						item->m_io_handler = nullptr;
					}
				}
			}
			else if (ret < 0)
			{
				int errc;
#ifdef _WIN32
				errc = WSAGetLastError();
#else
				errc = errno;
#endif
				throw std::system_error(errc, std::system_category());
			}
			return ret;
		}
	}

	void check_timeout()
	{
		time_t now;
		time(&now);
		for (auto& item : m_events)
		{
			if (item->m_expired > 0 && item->m_io_handler && item->m_expired < now)
			{
				item->m_io_handler(qtl::event::ef_timeout);
				item->m_expired = 0;
				item->m_io_handler = nullptr;
			}
		}
	}

	time_t m_expired;
	bool m_stoped;
};

simple_event_loop ev;

using namespace qtl::mysql;

void LogError(const error& e)
{
	fprintf(stderr, "MySQL Error(%d): %s\n", e.code(), e.what());
}

const char mysql_server[]="localhost";
const char mysql_user[]="root";
const char mysql_password[]="";
const char mysql_database[]="test";

void SimpleTest()
{
	async_connection connection;
	ev.reset();
	connection.open(ev, [&connection](const error& e) {
		if (e)
		{
			LogError(e);
			ev.stop();
		}
		else
		{
			printf("Connect to mysql ok.\n");
			connection.simple_query("select * from test", 0, [](MYSQL_ROW row, int field_count) {
				for (int i = 0; i != field_count; i++)
					printf("%s\t", row[i]);
				printf("\n");
				return true;
			}, [&connection](const error& e, size_t row_count) {
				if (e)
					LogError(e);
				else
					printf("Total %lu rows.\n", row_count);

				connection.close([]() {
					printf("Close connection ok.\n");
					ev.stop();
				});
			});
		}
	}, mysql_server, mysql_user, mysql_password, mysql_database);

	ev.run();
}

void ExecuteTest()
{
	async_connection connection;
	ev.reset();
	connection.open(ev, [&connection](const error& e) {
		if (e)
		{
			LogError(e);
			ev.stop();
		}
		else
		{
			printf("Connect to mysql ok.\n");
			connection.execute([&connection](const error& e, uint64_t affected) mutable {
					if (e)
						LogError(e);
					else
						printf("Insert %llu records ok.\n", affected);
					connection.close([]() {
						printf("Close connection ok.\n");
						ev.stop();
					});
			}, "insert into test(name, createtime, company) values(?, now(), ?)", 0, std::make_tuple("test name", "test company"));
		}
	}, mysql_server, mysql_user, mysql_password, mysql_database);

	ev.run();
};

void QueryTest()
{
	async_connection connection;
	ev.reset();
	connection.open(ev, [&connection](const error& e) {
		if (e)
		{
			LogError(e);
			ev.stop();
		}
		else
		{
			printf("Connect to mysql ok.\n");
			connection.query("select id, name, CreateTime, Company from test", 0, 
				[](int64_t id, const std::string& name, const qtl::mysql::time& create_time, const std::string& company) {
				char szTime[128] = { 0 };
				if (create_time.year != 0)
				{
					struct tm tm;
					create_time.as_tm(tm);
					strftime(szTime, 128, "%c", &tm);
				}
				printf("%lld\t%s\t%s\t%s\n", id, name.data(), szTime, company.data());
			}, [&connection](const error& e) {
				printf("query has completed.\n");
				if (e)
					LogError(e);

				connection.close([]() {
					printf("Close connection ok.\n");
					ev.stop();
				});
			});
		}
	}, mysql_server, mysql_user, mysql_password, mysql_database);

	ev.run();
}

void MultiQueryTest()
{
	async_connection connection;
	ev.reset();
	connection.open(ev, [&connection](const error& e) {
		if (e)
		{
			LogError(e);
			ev.stop();
		}
		else
		{
			printf("Connect to mysql ok.\n");
			connection.query_multi("call test_proc",
				[&connection](const error& e) {
				if (e)
					LogError(e);

				connection.close([]() {
					printf("Close connection ok.\n");
					ev.stop();
				});
			}, [](uint32_t i, const std::string& str) {
				printf("0=\"%d\", 'hello world'=\"%s\"\n", i, str.data());
			}, [](const qtl::mysql::time& time) {
				struct tm tm;
				time.as_tm(tm);
				printf("current time is: %s\n", asctime(&tm));
			});
		}
	}, mysql_server, mysql_user, mysql_password, mysql_database);

	ev.run();
}

int main(int argc, char* argv[])
{
	ExecuteTest();
	SimpleTest();
	QueryTest();
	MultiQueryTest();
	return 0;
}