/*
 * =====================================================================================
 *
 *       Filename:  api_test.cpp
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  2021年04月13日 14时31分46秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), lichao@aiotlink.com
 *   Organization:  
 *
 * =====================================================================================
 */
#include "bh_api.h"
#include "robust.h"
#include "util.h"
#include <atomic>
#include <boost/lockfree/queue.hpp>

using namespace bhome_msg;

namespace
{
typedef std::atomic<uint64_t> Number;

void GetApiError(int &ec, std::string &msg)
{
	void *pmsg = 0;
	int msg_len = 0;
	ec = BHGetLastError(&pmsg, &msg_len);
	msg.assign((char *) pmsg, msg_len);
	BHFree(pmsg, msg_len);
}

void Assign(Number &a, const Number &b) { a.store(b.load()); }
struct MsgStatus {

	Number nrequest_;
	Number nfailed_;
	Number nreply_;
	Number nserved_;
	MsgStatus() :
	    nrequest_(0), nreply_(0), nserved_(0) {}
	MsgStatus &operator=(const MsgStatus &a)
	{
		Assign(nrequest_, a.nrequest_);
		Assign(nserved_, a.nserved_);
		Assign(nreply_, a.nreply_);
		Assign(nfailed_, a.nfailed_);
		return *this;
	}
};

MsgStatus &Status()
{
	static MsgStatus st;
	return st;
}

void SubRecvProc(const void *proc_id,
                 const int proc_id_len,
                 const void *data,
                 const int data_len)
{
	std::string proc((const char *) proc_id, proc_id_len);
	MsgPublish pub;
	pub.ParseFromArray(data, data_len);
	printf("****************************************** Sub data, %s : %s\n", pub.topic().c_str(), pub.data().c_str());
}

void ServerProc(const void *proc_id,
                const int proc_id_len,
                const void *data,
                const int data_len,
                void *src)
{
	// printf("ServerProc: ");
	// DEFER1(printf("\n"););
	MsgRequestTopic request;
	if (request.ParseFromArray(data, data_len)) {
		MsgRequestTopicReply reply;
		reply.set_data(" reply: " + request.data());
		std::string s(reply.SerializeAsString());
		// printf("%s", reply.data().c_str());
		BHSendReply(src, s.data(), s.size());
		++Status().nserved_;
	}
}

void ClientProc(const void *proc_id,
                const int proc_id_len,
                const void *msg_id,
                const int msg_id_len,
                const void *data,
                const int data_len)
{
	std::string proc((const char *) proc_id, proc_id_len);
	MsgRequestTopicReply reply;
	if (reply.ParseFromArray(data, data_len)) {
		++Status().nreply_;
	}
	// printf("client Recv reply : %s\n", reply.data().c_str());
}

} // namespace
#include <chrono>
using namespace std::chrono;
// using namespace std::chrono_literals;

bool Register(const std::string &proc_id)
{
	bool reg = false;
	for (int i = 0; i < 3 && !reg; ++i) {
		ProcInfo proc;
		proc.set_proc_id(proc_id);
		proc.set_public_info("public info of demo_client. etc...");
		std::string proc_buf(proc.SerializeAsString());
		void *reply = 0;
		int reply_len = 0;
		reg = BHRegister(proc_buf.data(), proc_buf.size(), &reply, &reply_len, 2000);
		if (reg) {
			printf("register ok\n");
			// bool r = BHUnregister(proc_buf.data(), proc_buf.size(), &reply, &reply_len, 2000);
			// printf("unregister %s\n", r ? "ok" : "failed");
			// reg = BHRegister(proc_buf.data(), proc_buf.size(), &reply, &reply_len, 2000);
			// if (!reg) {
			// 	int ec = 0;
			// 	std::string msg;
			// 	GetApiError(ec, msg);
			// 	printf("reg error: %s\n", msg.c_str());
			// }
		} else {
			int ec = 0;
			std::string msg;
			GetApiError(ec, msg);
			printf("register failed, %d, %s\n", ec, msg.c_str());
		}

		BHFree(reply, reply_len);
		// Sleep(1s);
	}
	return reg;
}

bool SyncRequest(const std::string &topic, const std::string &data)
{ // SyncRequest
	MsgRequestTopic req;
	req.set_topic(topic);
	req.set_data(data);
	std::string s(req.SerializeAsString());
	// Sleep(10ms, false);
	std::string dest(BHAddress().SerializeAsString());
	void *proc_id = 0;
	int proc_id_len = 0;
	DEFER1(BHFree(proc_id, proc_id_len););
	void *reply = 0;
	int reply_len = 0;
	DEFER1(BHFree(reply, reply_len));
	bool r = BHRequest(dest.data(), dest.size(), s.data(), s.size(), &proc_id, &proc_id_len, &reply, &reply_len, 100);
	if (!r) {
		int ec = 0;
		std::string msg;
		GetApiError(ec, msg);
		printf("request error: %d, %s\n", ec, msg.c_str());
	} else {
		MsgRequestTopicReply ret;
		ret.ParseFromArray(reply, reply_len);
		printf("request result: %s\n", ret.data().c_str());
	}
	return r;
}

BOOST_AUTO_TEST_CASE(RestartTest)
{
	const std::string proc_id = "demo_node";
	for (int i = 0; i < 5; ++i) {
		printf("loop: %d --------------------------\n", i);
		if (!Register(proc_id)) { continue; }

		const std::string topic_ = "topic_";

		{ // Server Register Topics
			MsgTopicList topics;
			for (int i = 0; i < 10; ++i) {
				topics.add_topic_list(topic_ + std::to_string(i));
			}
			std::string s = topics.SerializeAsString();
			void *reply = 0;
			int reply_len = 0;
			bool r = BHRegisterTopics(s.data(), s.size(), &reply, &reply_len, 1000);
			DEFER1(BHFree(reply, reply_len));
		}

		BHStartWorker(&ServerProc, &SubRecvProc, &ClientProc);

		for (int i = 0; i < 60; ++i) {
			bool r = SyncRequest(topic_ + std::to_string(0), "request_data_" + std::to_string(i));
			if (!r) {
				void *msg = 0;
				int msg_len = 0;
				DEFER1(BHFree(msg, msg_len));
				int ec = BHGetLastError(&msg, &msg_len);
				if (ec == eNotRegistered) {
					printf("need re-register\n");
					break;
				}
			}
			Sleep(1s);
		}
	}
	BHCleanup();
}

BOOST_AUTO_TEST_CASE(ApiTest)
{
	auto max_time = std::chrono::steady_clock::time_point::max();
	auto dur = max_time.time_since_epoch();
	auto nsec = std::chrono::duration_cast<std::chrono::seconds>(dur).count();
	auto nmin = nsec / 60;
	auto nhour = nmin / 60;
	auto nday = nhour / 24;
	auto years = nday / 365;
	printf("seconds: %ld, hours: %ld , days:%ld, years: %ld\n",
	       nsec, nhour, nday, years);
	std::chrono::steady_clock::duration a(123456);
	printf("nowsec: %ld\n", NowSec());

	printf("maxsec: %ld\n", CountSeconds(max_time));

	const std::string proc_id = "demo_client";
	if (!Register(proc_id)) {
		return;
	}

	const std::string topic_ = proc_id + "_topic_";

	{ // Server Register Topics
		MsgTopicList topics;
		for (int i = 0; i < 10; ++i) {
			topics.add_topic_list(topic_ + std::to_string(i));
		}
		std::string s = topics.SerializeAsString();
		void *reply = 0;
		int reply_len = 0;
		bool r = BHRegisterTopics(s.data(), s.size(), &reply, &reply_len, 1000);
		DEFER1(BHFree(reply, reply_len));
	}
	auto PrintProcs = [](MsgQueryProcReply const &result) {
		printf("query proc result: %d\n", result.proc_list().size());
		for (int i = 0; i < result.proc_list().size(); ++i) {
			auto &info = result.proc_list(i);
			printf("proc [%d] %s, %s, %s\n\ttopics\n", i,
			       (info.online() ? "online" : "offline"),
			       info.proc().proc_id().c_str(), info.proc().name().c_str());
			for (auto &t : info.topics().topic_list()) {
				printf("\t\t %s\n", t.c_str());
			}
			printf("\n");
		}
		printf("\n");
	};
	if (0) {
		// query procs
		std::string dest(BHAddress().SerializeAsString());
		MsgQueryProc query;
		std::string s = query.SerializeAsString();
		void *reply = 0;
		int reply_len = 0;
		bool r = BHQueryProcs(dest.data(), dest.size(), s.data(), s.size(), &reply, &reply_len, 1000);
		DEFER1(BHFree(reply, reply_len));
		MsgQueryProcReply result;
		if (result.ParseFromArray(reply, reply_len) && IsSuccess(result.errmsg().errcode())) {
			PrintProcs(result);
		} else {
			printf("query proc error\n");
		}
		// printf("register topic : %s\n", r ? "ok" : "failed");
		// Sleep(1s);
	}
	for (int i = 0; i < 3; ++i) {
		// query procs with normal topic request
		MsgRequestTopic req;
		req.set_topic("#center_query_procs");
		// req.set_data("{\"proc_id\":\"#center.node\"}");
		std::string s(req.SerializeAsString());
		// Sleep(10ms, false);
		BHAddress host;
		printf("query with ip set\n");
		host.set_ip("127.0.0.1");
		host.set_port(kBHCenterPort);
		// center topic node address.
		host.set_mq_id(201);
		host.set_abs_addr(10072);

		std::string dest(host.SerializeAsString());
		void *proc_id = 0;
		int proc_id_len = 0;
		DEFER1(BHFree(proc_id, proc_id_len););
		void *reply = 0;
		int reply_len = 0;
		DEFER1(BHFree(reply, reply_len));
		bool r = BHRequest(dest.data(), dest.size(), s.data(), s.size(), &proc_id, &proc_id_len, &reply, &reply_len, 1000);
		if (!r) {
			int ec = 0;
			std::string msg;
			GetApiError(ec, msg);
			printf("topic query proc error: %s\n", msg.c_str());
		} else {
			MsgRequestTopicReply ret;
			ret.ParseFromArray(reply, reply_len);
			printf("\ntopic query proc : %s\n", ret.data().c_str());
			// MsgQueryProcReply result;
			// if (result.ParseFromArray(ret.data().data(), ret.data().size()) && IsSuccess(result.errmsg().errcode())) {
			// 	PrintProcs(result);
			// } else {
			// 	printf("topic query proc error\n");
			// }
		}
	}

	{ // Subscribe
		MsgTopicList topics;
		topics.add_topic_list("#center.node");
		for (int i = 0; i < 10; ++i) {
			topics.add_topic_list(topic_ + std::to_string(i * 2));
		}
		std::string s = topics.SerializeAsString();
		void *reply = 0;
		int reply_len = 0;
		bool r = BHSubscribeTopics(s.data(), s.size(), &reply, &reply_len, 1000);
		BHFree(reply, reply_len);
		printf("subscribe topic : %s\n", r ? "ok" : "failed");
	}

	auto ServerLoop = [&](std::atomic<bool> *run) {
		while (*run) {
			void *proc_id = 0;
			int proc_id_len = 0;
			DEFER1(BHFree(proc_id, proc_id_len););
			void *input = 0;
			int input_len = 0;
			DEFER1(BHFree(input, input_len));
			void *src = 0;
			if (BHReadRequest(&proc_id, &proc_id_len, &input, &input_len, &src, 10) && src) {

				MsgRequestTopic request;
				if (request.ParseFromArray(input, input_len)) {
					MsgRequestTopicReply reply;
					reply.set_data(" reply: " + request.data());
					std::string s(reply.SerializeAsString());
					// printf("%s", reply.data().c_str());
					BHSendReply(src, s.data(), s.size());
					++Status().nserved_;
				}
				src = 0;
			}
		}
	};

	{
		for (int i = 0; i < 1; ++i) {
			MsgPublish pub;
			pub.set_topic(topic_ + std::to_string(i));
			pub.set_data("pub_data_" + std::string(104 * 1, 'a'));
			std::string s(pub.SerializeAsString());
			BHPublish(s.data(), s.size(), 0);
			// Sleep(1s);
		}
	}

	auto asyncRequest = [&](uint64_t nreq) {
		for (uint64_t i = 0; i < nreq; ++i) {
			MsgRequestTopic req;
			req.set_topic(topic_ + std::to_string(0));
			req.set_data("request_data_" + std::to_string(i));
			std::string s(req.SerializeAsString());
			void *msg_id = 0;
			int len = 0;
			DEFER1(BHFree(msg_id, len););
			// Sleep(10ms, false);
			std::string dest(BHAddress().SerializeAsString());

			bool r = BHAsyncRequest(dest.data(), dest.size(), s.data(), s.size(), 0, 0);
			if (r) {
				++Status().nrequest_;
			} else {
				++Status().nfailed_;
				static std::atomic<int64_t> last(0);
				auto now = NowSec();
				if (last.exchange(now) < now) {
					int ec = 0;
					std::string msg;
					GetApiError(ec, msg);
					printf("request topic error --------- : %s\n", msg.c_str());
				}
			}
		}
	};
	auto showStatus = [](std::atomic<bool> *run) {
		MsgStatus last;
		while (*run) {
			auto &st = Status();
			Sleep(1s, false);
			printf("nreq: %8ld, spd %8ld | failed: %8ld | nsrv: %8ld, spd %8ld | nreply: %8ld, spd %8ld\n",
			       st.nrequest_.load(), st.nrequest_ - last.nrequest_,
			       st.nfailed_.load(),
			       st.nserved_.load(), st.nserved_ - last.nserved_,
			       st.nreply_.load(), st.nreply_ - last.nreply_);
			last = st;
		}
	};
	auto hb = [](std::atomic<bool> *run) {
		while (*run) {
			Sleep(1s, false);
			bool r = BHHeartbeatEasy(1000);
			printf("heartbeat: %s\n", r ? "ok" : "failed");
		}
	};

	std::atomic<bool> run(true);

	ThreadManager threads;

#if 1
	BHStartWorker(&ServerProc, &SubRecvProc, &ClientProc);
#else
	BHStartWorker(FServerCallback(), &SubRecvProc, &ClientProc);
	threads.Launch(ServerLoop, &run);
#endif

	boost::timer::auto_cpu_timer timer;
	threads.Launch(hb, &run);
	threads.Launch(showStatus, &run);
	int ncli = 10;
	const int64_t nreq = 1000 * 100;

	for (int i = 0; i < 10; ++i) {
		SyncRequest(topic_ + std::to_string(0), "request_data_" + std::to_string(i));
	}

	for (int i = 0; i < ncli; ++i) {
		threads.Launch(asyncRequest, nreq);
	}

	int same = 0;
	uint64_t last = 0;
	while (last < nreq * ncli && same < 3) {
		Sleep(1s, false);
		auto cur = Status().nreply_.load();
		if (last == cur) {
			++same;
			printf("same %d\n", same);
		} else {
			last = cur;
			same = 0;
		}
	}

	Sleep(1s);

	run = false;
	threads.WaitAll();
	auto &st = Status();
	printf("nreq: %8ld, nsrv: %8ld, nreply: %8ld\n", st.nrequest_.load(), st.nserved_.load(), st.nreply_.load());
	BHCleanup();
	printf("after cleanup\n");
}