/*
 * =====================================================================================
 *
 *       Filename:  topic_request.cpp
 *
 *    Description:  topic request sockets
 *
 *        Version:  1.0
 *        Created:  2021年04月01日 09时35分35秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), 
 *   Organization:  
 *
 * =====================================================================================
 */
#include "topic_request.h"
#include "bh_util.h"
#include "msg.h"
#include <chrono>
#include <condition_variable>

using namespace bhome_msg;

bool SocketRequest::StartWorker(const RequestResultCB &rrcb, int nworker)
{
	auto AsyncRecvProc = [this, rrcb](BHMsg &msg) {
		auto Find = [&](RecvBHMsgCB &cb) {
			std::lock_guard<std::mutex> lock(mutex());
			const std::string &msgid = msg.msg_id();
			auto pos = async_cbs_.find(msgid);
			if (pos != async_cbs_.end()) {
				cb.swap(pos->second);
				async_cbs_.erase(pos);
				return true;
			} else {
				return false;
			}
		};

		RecvBHMsgCB cb;
		if (Find(cb) && cb) {
			cb(msg);
		} else if (msg.type() == kMsgTypeRequestTopicReply && rrcb) {
			MsgRequestTopicReply reply;
			if (reply.ParseFromString(msg.body())) {
				rrcb(reply.data());
			}
		} else {
			// ignored, or dropped
		}
	};

	return Start(AsyncRecvProc, nworker);
}

bool SocketRequest::AsyncRequest(const Topic &topic, const void *data, const size_t size, const int timeout_ms)
{
	try {
		BHAddress addr;
		if (QueryRPCTopic(topic, addr, timeout_ms)) {
			const BHMsg &msg(MakeRequest(mq().Id(), topic, data, size));
			return AsyncSend(addr.mq_id().data(), &msg, timeout_ms);
		} else {
			return false;
		}
	} catch (...) {
		return false;
	}
}
bool SocketRequest::AsyncRequest(const Topic &topic, const void *data, const size_t size, const int timeout_ms, const RequestResultCB &cb)
{
	auto Call = [&](const void *remote) {
		const BHMsg &msg(MakeRequest(mq().Id(), topic, data, size));
		auto onRecv = [cb](BHMsg &msg) {
			if (msg.type() == kMsgTypeRequestTopicReply) {
				MsgRequestTopicReply reply;
				if (reply.ParseFromString(msg.body())) {
					cb(reply.data());
				}
			}
		};
		return AsyncSend(remote, &msg, timeout_ms, onRecv);
	};

	try {
		BHAddress addr;
		if (QueryRPCTopic(topic, addr, timeout_ms)) {
			return Call(addr.mq_id().data());
		} else {
			return false;
		}
	} catch (...) {
		return false;
	}
}

bool SocketRequest::SyncRequest(const Topic &topic, const void *data, const size_t size, std::string &out, const int timeout_ms)
{
	try {
		BHAddress addr;
		if (QueryRPCTopic(topic, addr, timeout_ms)) {
			const BHMsg &req(MakeRequest(mq().Id(), topic, data, size));
			BHMsg reply;
			if (SyncSendAndRecv(addr.mq_id().data(), &req, &reply, timeout_ms) && reply.type() == kMsgTypeRequestTopicReply) {
				MsgRequestTopicReply dr;
				if (dr.ParseFromString(reply.body())) {
					dr.mutable_data()->swap(out);
					return true;
				} else {
					printf("error parse reply.\n");
				}
			} else {
				printf("error recv data. line: %d\n", __LINE__);
			}
		} else {
			printf("error recv data. line: %d\n", __LINE__);
		}
	} catch (...) {
		printf("error recv data. line: %d\n", __LINE__);
	}
	return false;
}

bool SocketRequest::AsyncSend(const void *remote, const void *pmsg, const int timeout_ms)
{
	assert(remote && pmsg);
	try {
		const BHMsg &msg = *static_cast<const BHMsg *>(pmsg);
		return mq().Send(*static_cast<const MQId *>(remote), msg, timeout_ms);
	} catch (...) {
		return false;
	}
}
bool SocketRequest::AsyncSend(const void *remote, const void *pmsg, const int timeout_ms, const RecvBHMsgCB &cb)
{
	assert(remote && pmsg);
	try {
		const BHMsg &msg = *static_cast<const BHMsg *>(pmsg);
		auto RegisterCB = [&]() {
			std::lock_guard<std::mutex> lock(mutex());
			async_cbs_.emplace(msg.msg_id(), cb);
		};

		return mq().Send(*static_cast<const MQId *>(remote), msg, timeout_ms, RegisterCB);
	} catch (...) {
		return false;
	}
}

bool SocketRequest::SyncSendAndRecv(const void *remote, const void *msg, void *result, const int timeout_ms)
{
	struct State {
		std::mutex mutex;
		std::condition_variable cv;
		bool canceled = false;
	};

	try {
		std::shared_ptr<State> st(new State);
		auto endtime = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);

		auto OnRecv = [=](BHMsg &msg) {
			std::unique_lock<std::mutex> lk(st->mutex);
			if (!st->canceled) {
				static_cast<BHMsg *>(result)->Swap(&msg);
				st->cv.notify_one();
			} else {
			}
		};

		std::unique_lock<std::mutex> lk(st->mutex);
		bool sendok = AsyncSend(remote, msg, timeout_ms, OnRecv);
		if (sendok && st->cv.wait_until(lk, endtime) == std::cv_status::no_timeout) {
			return true;
		} else {
			st->canceled = true;
			return false;
		}
	} catch (...) {
		return false;
	}
}

bool SocketRequest::QueryRPCTopic(const Topic &topic, bhome::msg::BHAddress &addr, const int timeout_ms)
{
	if (topic_cache_.Find(topic, addr)) {
		return true;
	}

	BHMsg result;
	const BHMsg &msg = MakeQueryTopic(mq().Id(), topic);
	if (SyncSendAndRecv(&BHTopicCenterAddress(), &msg, &result, timeout_ms)) {
		if (result.type() == kMsgTypeQueryTopicReply) {
			MsgQueryTopicReply reply;
			if (reply.ParseFromString(result.body())) {
				addr = reply.address();
				if (addr.mq_id().empty()) {
					return false;
				} else {
					topic_cache_.Update(topic, addr);
					return true;
				}
			}
		}
	} else {
	}
	return false;
}