// influxdb2test.cpp: 定义应用程序的入口点。
//
#include <unistd.h>
#include <iostream>
#include <sstream>
#include <string>
#include <memory>
#include "iotdbtest.h"
#include "cpr/cpr.h"
#include "nlohmann/json.hpp"
#include "iotdbclient.h"
#include "ThreadPool.h"
#include <fmt/format.h>
#include <mutex>
#include <chrono>
#include <fmt/chrono.h>

// for convenience
//using json = nlohmann::json;

using namespace std;
using namespace chrono;

shared_ptr<ThreadPool> gthreadpools;
shared_ptr<ThreadPool> gthreadpoolm;

string g_address = "localhost:8086";

int g_start_no_mul = 1; //多点写，起始点号
int g_num_onetime_mul = 5000; //多点写，每包数据的点数
int g_write_times_mul = 3600 * 24 * 365; //多点写，写次数
int g_time_offset_mul = 0; //多点写，往前推的时间偏移(S)，即从当前时间往前推多少时间的时标开始写
int g_sleep_time_mul = 1000; //单点写，每次写的sleep时间

int g_start_no_single = 1; //单点写，起始点号
int g_num_onetime_single = 1; //单点写，每包数据的点(tag)数
int g_write_time_single = 3600; //单点写，每次每个点(tag)写的时间数(s)
int g_time_offset_single = 31536000; //单点写，往前推的时间偏移(S)，即从当前时间往前推多少时间的时标开始写
int g_sleep_time_single = 1000; //单点写，每次写的sleep时间

int g_create_tag_start_no = 1; //起始点号
int g_create_tag_onetime_num = 10000; // 每次创建多少点
int g_create_tag_num = 50000; //要创建的总点数

std::atomic<int> g_min_write_time(50000); //最小写时间
std::atomic<int> g_max_write_time(0); //最大写时间


time_t test_start_time = 0; //测试开始时间

//CRITICAL_SECTION g_criticalSection;
mutex g_mutex;
void workerthreads()
{
	//写数据
	//string strUrlw = "http://localhost:8086/write";
	IotdbClient client;

	string strUrlw = "http://";
	strUrlw += g_address;
	strUrlw += "/api/v2/write";

	int iRet = -1;
	string strResponse;

	int iStartNo = 0;
	int iEndNo = 0;
	int write_times = 0;

	std::unique_lock<std::mutex> lk(g_mutex);
	iStartNo = g_start_no_single;
	iEndNo = iStartNo + g_num_onetime_single;
	g_start_no_single = iEndNo;
	write_times = (g_time_offset_single - 1) / g_write_time_single + 1;

	lk.unlock();
	//int iBufLen = 1024*500;
	time_t tt_now = time(NULL);

	fmt::memory_buffer data_buffer;
	data_buffer.resize(1024*5);

	int64_t time1 = 0;
	int write_count = 0;
	double dValue = 0;
	int i_q = 1;
	srand(time(nullptr));

	for (int j = 0; j < write_times; j++)
	{
		auto time_1 = steady_clock::now();
		data_buffer.clear();
		for (int i = iStartNo; i < iEndNo; i++)
		{
			for (int k = 0; k < g_write_time_single; k++)
			{
				i_q = k % 4;
				fmt::format_to(data_buffer, "v,TagName=Tag_{}", i);
				dValue = (double)rand() / 101;
				fmt::format_to(data_buffer, " value={},q={} ", dValue, i_q);

				time1 = (tt_now - g_time_offset_single + g_write_time_single* j + k) * 1000000000; //时间转换为ns
				fmt::format_to(data_buffer, "{}\n", time1);
			}
		}
		auto time_2 = steady_clock::now();
		string data = fmt::to_string(data_buffer);
		bool bRet = client.WriteData(strUrlw, "org_test", "bucket_test", data);
		write_count++;
		auto time_3 = steady_clock::now();

		auto duration_write = duration_cast<milliseconds>(time_3 - time_2);
		auto duration_make_pkt = duration_cast<milliseconds>(time_2 - time_1);

		cout << "iStartNo=" << iStartNo << ",count=" << write_count << ",write time=" << duration_write.count() << ",make pkg time=" << duration_make_pkt.count() << ",iRet=" << iRet << ",ilen=" << data.size() << ",Response=" << strResponse << endl;

		if (g_sleep_time_single > 0)//sleep时间大于0
		{
			auto use_time = duration_cast<milliseconds>(time_3 - time_1);
			auto sleep_time = std::chrono::milliseconds(g_sleep_time_single) - use_time;
			if (sleep_time.count() > 0)
			{
				this_thread::sleep_for(sleep_time);
			}
		}
	}
}

void workerthreadaggregation()
{
	//写数据
	//string strUrlw = "http://localhost:8086/query?";
	string strUrlw = "http://172.21.31.51:8086/query?";
	IotdbClient client;
	int iRet = -1;
	string strResponse;
	int iTagNo = 0;
	int iTagNum = 0;
	std::unique_lock<std::mutex> lk(g_mutex);
	iTagNum = g_num_onetime_single;
	iTagNo = g_start_no_single;
	g_start_no_single += g_num_onetime_single;

	iTagNum = g_num_onetime_single;
	lk.unlock();
	//int iBufLen = 1024*500;
	time_t tt = time(NULL);

	fmt::memory_buffer data_buffer;
	data_buffer.resize(1024 * 5);

	int iLen = 0;
	int iCount = 0;
	long lValue = 0;
	for (int k = 0; k < iTagNum; k++)
	{
		data_buffer.clear();
		/*
		wPkt << "SELECT max(value),min(value),mean(value) INTO mydb.longyears.TagValue20m FROM TagValue where TagName='Tag_";
		wPkt << iTagNo;
		wPkt << "'";
		wPkt << " GROUP BY time(20m),TagName";

		wPkt.GetUsedBuf(&szData, iLen);
		cout << "aggregation begain" << endl;

		pClient->aggregation(strUrlw, "mydb", szData, strResponse, 300000);
		cout << "aggregation end " << strResponse << endl;

		iTagNo++;
		*/
	}

}

/**
 * @brief 多点单时刻写
*/
void workerthreadm()
{
	//写数据
	//string strUrlw = "http://172.21.31.51:8086/write";
	//string strUrlw = "http://localhost:8086/write";

	string strUrlw = "http://";
	strUrlw += g_address;
	strUrlw += "/api/v2/write";

	//string strUrlw = "http://localhost:8086/write";

	//string strUrlw = "http://localhost:8428/write";

	IotdbClient client;
	client.SessionOpen("127.0.0.1", 6667, "root", "root");
	int iRet = -1;
	string strResponse;
	int iStartNo = 0;
	int iEndNo = 0;
	std::unique_lock<std::mutex> lk(g_mutex);
	iStartNo = g_start_no_mul;
	iEndNo = iStartNo + g_num_onetime_mul;
	g_start_no_mul = iEndNo;
	lk.unlock();

	time_t tt = time(NULL);

	fmt::memory_buffer data_buffer;
	data_buffer.resize(1024 * 5);
	int iLen = 0;

	long interval = 31536 * 1000;
	long interval1 = interval;
	int64_t time1 = 0;
	int iCount = 0;
	long lValue = 0;
	double dValue = 1.1;
	double tag1_value = 0;
	int i_q = 1;

	string deviceId;
	vector<string> measurements;
	measurements.push_back("v");
	measurements.push_back("q");

	time_t start_tt = time(NULL);
	//有前推时间，造较早时间的数据
	if (g_time_offset_mul > 0)
	{
		tt = time(NULL);
		start_tt = tt;
		tt -= g_time_offset_mul;
	}
	srand(time(nullptr));

	for (int j = 0; j < g_write_times_mul; j++)
	{
		vector<string> deviceIds;
		vector<vector<string>> measurementsList;
		vector<std::vector<TSDataType::TSDataType>> typesList;
		vector<vector<string>> valuesList;
		vector<std::vector<char*>> sz_valuesList;
		vector<int64_t> timestamps;
		vector<string>  values_temp;
		string value;

		data_buffer.clear();
		i_q = j % 4;
		time_t now_tt = time(NULL);
		if (0 == g_time_offset_mul)
		{
			tt = time(NULL);
		}
		else
		{
			tt++;
		}

		auto time_1 = steady_clock::now();

		//fmt::format_to(data_buffer, "INSERT INTO v VALUES");

		for (int i = iStartNo; i < iEndNo; i++)
		{
			dValue = (double)rand() / 101;

			string deviceId = fmt::format("root.test2.tag_{}", i);
			deviceIds.push_back(deviceId);

			measurementsList.push_back(measurements);

			//vector<string> values;
			vector<char*> sz_values;
			value = fmt::format("{}", dValue);
			values_temp.push_back(value);
			//sz_values.push_back((char*)value.c_str());
			sz_values.push_back((char*)&dValue);

			value = fmt::format("{}", i_q);
			values_temp.push_back(value);
			//sz_values.push_back((char*)value.c_str());
			sz_values.push_back((char*)&i_q);

			sz_valuesList.push_back(sz_values);


			vector<TSDataType::TSDataType> dataTypes;
			dataTypes.push_back(TSDataType::DOUBLE);
			dataTypes.push_back(TSDataType::INT32);
			typesList.push_back(dataTypes);

			//time1 = tt * 1000000000;
			time1 = tt *   1000;

			timestamps.push_back(time1);

			if (i == iStartNo)
			{
				tag1_value = dValue;
			}

		}
		iCount++;
		if (iCount > 1000)
		{
			iCount = 0;
		}
		auto time_2 = steady_clock::now();
		interval1 = interval - j;
		/*
		for (int i = 0; i < 5000; i++)
		{
			ssDatas << "Tag_0000009";
			ssDatas << ",";
			ssDatas<<"TagName=Tag_0000009";
			ssDatas<<" value=";
			ssDatas<<GetTickCount();
			ssDatas<<" ";
			ssDatas<<(tt-interval)*1000000000;
			ssDatas<<endl;
			interval--;
		}
		*/
		//strDatasw = ssDatas.str();
		//ssDatas.str("");
		//ssDatas.clear();
		auto time_3 = steady_clock::now();
		string data = fmt::to_string(data_buffer);
		//bool bRet = client.WriteDataTcp(data);
		client.insertAlignedRecords(deviceIds, timestamps, measurementsList, typesList, sz_valuesList);
		//bool bRet = client.Exec("http://localhost:9000/exec",data);
		auto time_4 = steady_clock::now();

		auto duration_write = duration_cast<milliseconds>(time_4 - time_3);
		auto duration_make_pkt = duration_cast<milliseconds>(time_2 - time_1);

		/*
		cout << "iStartNo=" << iStartNo << ",count=" << j << ",write time=" << duration_write.count()
			<< ",make pkg time=" << duration_make_pkt.count() << ",iRet=" << iRet << ",ilen=" << iLen << ",Response=" << strResponse << endl;
		*/

		if (duration_write.count() < g_min_write_time)
		{
			g_min_write_time = duration_write.count();
		}

		if (duration_write.count() > g_max_write_time)
		{
			g_max_write_time = duration_write.count();
		}

		string str_time = fmt::format("{:%Y-%m-%d %H:%M:%S}", fmt::localtime(tt));
		//string str_start_time = fmt::format("{:%Y-%m-%d %H:%M:%S}", fmt::localtime(start_tt));
		string str_now_time = fmt::format("{:%Y-%m-%d %H:%M:%S}", fmt::localtime(now_tt));
		//fmt::print("The date is {:%Y-%m-%d}.", fmt::localtime(t));

		cout << "iStartNo=" << iStartNo << ",time=" << str_time << ",starttime=" << start_tt << ",now=" << str_now_time
			<< ",tag1=" << tag1_value << ",count=" << j << ",write time=" << duration_write.count() << ",make pkg time="
			<< duration_make_pkt.count() << ",iRet=" << iRet << ",ilen=" << iLen << ",Response=" << strResponse
			<<",min write time="<< g_min_write_time<<",max write time=" << g_max_write_time<<endl;

		if (g_sleep_time_mul > 0)//sleep时间大于0
		{
			auto use_time = duration_cast<milliseconds>(time_4 - time_1);
			auto sleep_time = std::chrono::milliseconds(g_sleep_time_mul) - use_time;
			if (sleep_time.count() > 0)
			{
				this_thread::sleep_for(sleep_time);
			}
		}

		//this_thread::sleep_for(std::chrono::milliseconds(1000)- duration_write);
		//Sleep(abs((int)(1000 - (dwTick2 - dwTick3))));
	}

	time_t test_end_time = time(NULL);
	string str_start_time = fmt::format("{:%Y-%m-%d %H:%M:%S}", fmt::localtime(test_start_time));
	string str_end_time = fmt::format("{:%Y-%m-%d %H:%M:%S}", fmt::localtime(test_end_time));
	cout << "test end,start time:" << str_start_time << ",end time:" << str_end_time<< endl;
}

void createMultiTimeseries(IotdbClient& client)
{
	/*
	vector<string> paths;
	vector<TSDataType::TSDataType> tsDataTypes;
	vector<TSEncoding::TSEncoding> tsEncodings;
	vector<CompressionType::CompressionType> compressionTypes;
	*/

	/*
	int g_create_tag_start_no = 1; //起始点号
	int g_create_tag_onetime_num = 10000; // 每次创建多少点
	int g_create_tag_num = 50000; //要创建的总点数
	*/

	int j_count = g_create_tag_num / g_create_tag_onetime_num;
	if (g_create_tag_num % g_create_tag_onetime_num == 0)
	{
		j_count = g_create_tag_num / g_create_tag_onetime_num;
	}
	else
	{
		j_count = g_create_tag_num / g_create_tag_onetime_num + 1;
	}
	int tag_count = 0; //要创建的总点数计数
	int tag_no = g_create_tag_start_no; //点号

	for (int j = 0; j < j_count; j++)
	{
		/*
		paths.clear();
		tsDataTypes.clear();
		tsEncodings.clear();
		compressionTypes.clear();
		*/
		vector<string> paths;
		vector<TSDataType::TSDataType> tsDataTypes;
		vector<TSEncoding::TSEncoding> tsEncodings;
		vector<CompressionType::CompressionType> compressionTypes;
		string path;
		for (int i = 1; i <= g_create_tag_onetime_num && tag_count < g_create_tag_num; i++)
		{
			path = fmt::format("root.test2.tag_{}.v", tag_no);
			paths.push_back(path);
			tsDataTypes.push_back(TSDataType::DOUBLE);
			tsEncodings.push_back(TSEncoding::GORILLA);
			compressionTypes.push_back(CompressionType::LZ4);

			path = fmt::format("root.test2.tag_{}.q", tag_no);
			paths.push_back(path);
			tsDataTypes.push_back(TSDataType::INT64);
			tsEncodings.push_back(TSEncoding::RLE);
			compressionTypes.push_back(CompressionType::LZ4);

			tag_no++;
			tag_count++;
		}

		client.createMultiTimeseries(paths, tsDataTypes, tsEncodings, compressionTypes);
	}

	std::cout << "createMultiTimeseries end"<<endl;
}

void insertRecords(IotdbClient& client)
{
	string deviceId;
	vector<string> measurements;
	measurements.push_back("v");
	measurements.push_back("q");

	for (int k = 0; k < 1000; k++)
	{
		time_t now_tt = time(NULL);
		for (int j = 0; j < 5; j++)
		{
			vector<string> deviceIds;
			vector<vector<string>> measurementsList;
			vector<vector<string>> valuesList;
			vector<int64_t> timestamps;

			for (int i = 1; i <= 10000; i++)
			{
				string deviceId = fmt::format("root.test2.tag_{}", j * 10000 + i);
				deviceIds.push_back(deviceId);

				measurementsList.push_back(measurements);

				vector<string> values;
				values.push_back("100");
				values.push_back("2");
				valuesList.push_back(values);

				int64_t time1 = now_tt * 1000000000;

				timestamps.push_back(time1);
			}

			client.insertRecords(deviceIds, timestamps, measurementsList, valuesList);
		}
		cout << "insertRecords count:" << k<<endl;
		sleep(1000);
	}

	cout << "insertRecords end" << endl;
}

void QueryTest(std::string& sql)
{
	if (sql.empty())
	{
		return;
	}

	IotdbClient client;
	auto time_1 = steady_clock::now();
	client.SessionOpen("127.0.0.1", 6667, "root", "root");
	//client.SetStorageGroup("root.test2");
	bool query = client.Query(sql);
	client.SessionClose();
	auto time_2 = steady_clock::now();

	auto query_write = duration_cast<milliseconds>(time_2 - time_1).count();

	cout << "QueryTest:sql=" << sql <<",res:"<< query<<",use time(ms):"<< query_write<< endl;
}


int main(int argc, char* argv[])
{
	IotdbClient client;
	string url = "http://localhost:9000/exec";
	/*
	string query;
	query = "CREATE TABLE IF NOT EXISTS tagvalue1(name STRING, value double,ts timestamp) TIMESTAMP(ts) PARTITION BY MONTH";

	client.Exec(url, query);
	*/

/*
	//cout << "Hello CMake." << endl;
	url = "http://";
	url += g_address;
	url += "/api/v2/setup";

	UserInfor user_infor;
	user_infor.user_name = "admin";
	user_infor.password = "12345678";
	user_infor.org = "org_test";
	user_infor.bucket = "bucket_test";
	user_infor.token = "token_string";
	//client.InitUser_Org_Bucket(url,user_infor);

	user_infor.user_name = "admin";
	user_infor.password = "12345678";
	user_infor.org = "org_test2";
	user_infor.bucket = "bucket_test2";
	user_infor.token = "token_string2";
	url = "http://";
	url += g_address;
	url += "/api/v2/setup/user";

	//client.AddUser_Org_Bucket(url,user_infor);

	//std::unique_lock<std::mutex> lk(g_mutex);
*/
	int ithreadnum0 = 1;
	int imode = 1;
	int ithreadnum1 = 5;
	if (argc > 1)
	{
		g_address = argv[1];
		if (argc > 2)
		{
			imode = atoi(argv[2]);
		}

		if (0 == imode)//写单标签模式
		{
			if (argc > 3)
			{
				ithreadnum0 = atoi(argv[3]);
			}

			if (argc > 4)
			{
				g_start_no_single = atoi(argv[4]);
			}

			if (argc > 5)
			{
				g_num_onetime_single = atoi(argv[5]);
			}

			if (argc > 6)
			{
				g_write_time_single = atoi(argv[6]);
			}

			if (argc > 7)
			{
				g_time_offset_single = atoi(argv[7]);
			}

			if (argc > 8)
			{
				g_sleep_time_single = atoi(argv[8]);
			}
		}
		else if (1 == imode)//批量标签模式
		{
			if (argc > 3)
			{
				ithreadnum1 = atoi(argv[3]);
			}
			if (argc > 4)
			{
				g_write_times_mul = atoi(argv[4]);
			}

			if (argc > 5)
			{
				g_start_no_mul = atoi(argv[5]);
			}

			if (argc > 6)
			{
				g_num_onetime_mul = atoi(argv[6]);
			}

			if (argc > 7)
			{
				g_time_offset_mul = atoi(argv[7]);
			}

			if (argc > 8)
			{
				g_sleep_time_mul = atoi(argv[8]);
			}
		}
		else if (2 == imode)//写单标签模式和批量标签模式
		{
			if (argc > 10)
			{
				ithreadnum0 = atoi(argv[3]);
				g_start_no_single = atoi(argv[4]);
				g_num_onetime_single = atoi(argv[5]);
				g_write_time_single = atoi(argv[6]);
				g_time_offset_single = atoi(argv[7]);
				g_sleep_time_single = atoi(argv[8]);

				ithreadnum1 = atoi(argv[9]);
				g_write_times_mul = atoi(argv[10]);
				g_start_no_mul = atoi(argv[11]);
				g_num_onetime_mul = atoi(argv[12]);
				g_time_offset_mul = atoi(argv[13]);
				g_sleep_time_mul = atoi(argv[14]);
			}
		}
		else if (3 == imode) //聚合
		{
			if (argc > 3)
			{
				ithreadnum0 = atoi(argv[3]);
			}

			if (argc > 4)
			{
				g_start_no_single = atoi(argv[5]);
			}

			if (argc > 6)
			{
				g_num_onetime_single = atoi(argv[7]);
			}
		}
		else if (3 == imode) //聚合
		{
			if (argc > 2)
			{
				ithreadnum0 = atoi(argv[2]);
			}

			if (argc > 3)
			{
				g_start_no_single = atoi(argv[3]);
			}

			if (argc > 4)
			{
				g_num_onetime_single = atoi(argv[4]);
			}
		}
		else if (4 == imode)
		{
			if (argc > 3)
			{
				g_create_tag_num = atoi(argv[3]);
			}

			if (argc > 4)
			{
				g_create_tag_start_no = atoi(argv[4]);
			}

			if (argc > 5)
			{
				g_create_tag_onetime_num = atoi(argv[5 ]);
			}
		}


	}
	cout <<"param num:           " << argc<<endl;
	//<< "    " << *argv << endl;
	cout<< "mode=                " << imode << endl
		<< "g_address=           " << g_address << endl
		<< endl
		<< "threadnum0=          " << ithreadnum0 << endl
		<< "g_start_no_single=   " << g_start_no_single << endl
		<< "g_num_onetime_single=" << g_num_onetime_single << endl
		<< "g_write_time_single= " << g_write_time_single << endl
		<< "g_time_offset_single=" << g_time_offset_single << endl
		<< "g_sleep_time_single= " << g_sleep_time_single << endl
		<< endl
		<< "threadnum1=          " << ithreadnum1 << endl
		<< "g_write_times_mul=   " << g_write_times_mul << endl
		<< "g_start_no_mul=      " << g_start_no_mul << endl
		<< "g_num_onetime_mul=   "<< g_num_onetime_mul << endl
		<< "g_time_offset_mul=   " << g_time_offset_mul << endl
		<< "g_sleep_time_mul=    "<< g_sleep_time_mul<< endl
	    << "g_create_tag_num=    " << g_create_tag_num << endl
		<< "g_create_tag_start_no=    " << g_create_tag_start_no << endl
		<< "g_create_tag_onetime_num=    " << g_create_tag_onetime_num << endl;
	char c1;
	cin >> c1;
	//return 0;

	//写单标签模式
	if (0 == imode)
	{
		gthreadpools = make_shared<ThreadPool>(ithreadnum0);
		int par = 1;
		for (int i = 0; i < ithreadnum0; i++)
		{
			gthreadpools->enqueue(std::bind(&workerthreads));
		}
	}
	else if (1 == imode) //批量标签模式
	{
		test_start_time = time(NULL);

		gthreadpoolm = make_shared<ThreadPool>(ithreadnum1);
		int par = 1;
		for (int i = 0; i < ithreadnum1; i++)
		{
			gthreadpoolm->enqueue(std::bind(&workerthreadm));
		}
	}
	else if (2 == imode)//写单标签模式和批量标签模式
	{
		gthreadpools = make_shared<ThreadPool>(ithreadnum0);
		int par = 1;
		for (int i = 0; i < ithreadnum0; i++)
		{
			gthreadpools->enqueue(std::bind(&workerthreads));
		}

		gthreadpoolm = make_shared<ThreadPool>(ithreadnum1);
		for (int i = 0; i < ithreadnum1; i++)
		{
			gthreadpoolm->enqueue(std::bind(&workerthreadm));
		}


	}
	else if (3 == imode)//聚合
	{
		gthreadpools = make_shared<ThreadPool>(ithreadnum0);
		int par = 1;
		for (int i = 0; i < ithreadnum0; i++)
		{
			gthreadpools->enqueue(std::bind(&workerthreadaggregation));
		}
	}
	else if (4 == imode) //创建时间序列
	{
		client.SessionOpen("127.0.0.1", 6667, "root", "root");
		client.SetStorageGroup("root.test2");
		createMultiTimeseries(client);
		//insertRecords(client);
		client.SessionClose();
	}
	else if (5 == imode) //查询
	{
		while (true)
		{
			std::string s_input;
			std::getline(std::cin, s_input);

			if (s_input == "q" || s_input == "Q")
			{
				break;
			}

			QueryTest(s_input);
		}
	}
	//lk.unlock();
	char c;
	cin >> c;

	stringstream ss;
	url = "http://localhost:8086/write";
	ss << "mem,host=host1 used_percent=23.43234543 1556896326" << endl;
	ss << "mem,host=host2 used_percent=26.81522361 1556896326" << endl;
	ss << "mem,host=host1 used_percent=22.52984738 1556896336" << endl;
	ss << "mem,host=host2 used_percent=27.18294630 1556896336" << endl;
	string sData = ss.str();
	//client.WriteData(url,"org_test","bucket_test", sData);


	return 0;
}