/*
 * =====================================================================================
 *
 *       Filename:  reqrep.cpp
 *
 *    Description:  topic request/reply sockets
 *
 *        Version:  1.0
 *        Created:  2021年04月01日 09时35分35秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), 
 *   Organization:  
 *
 * =====================================================================================
 */
#include "reqrep.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 = [&](RecvCB &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;
			}
		};

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

	return Start(AsyncRecvProc, nworker);
}

bool SocketRequest::AsyncRequest(const std::string &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() == kMsgTypeReply) {
				DataReply 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());
		}
	} catch (...) {
		return false;
	}
}

bool SocketRequest::SyncRequest(const std::string &topic, const void *data, const size_t size, const int timeout_ms, std::string &out)
{
	try {
		BHAddress addr;
		if (QueryRPCTopic(topic, addr, timeout_ms)) {
			const BHMsg &msg(MakeRequest(mq().Id(), topic, data, size));
			BHMsg reply;
			if (SyncSendAndRecv(addr.mq_id().data(), &msg, &reply, timeout_ms) && reply.type() == kMsgTypeReply) {
				DataReply dr;
				if (dr.ParseFromString(msg.body())) {
					dr.mutable_data()->swap(out);
					return true;
				}
			}
		}
	} catch (...) {
		return false;
	}
}

bool SocketRequest::AsyncSend(const void *remote, const void *pmsg, const int timeout_ms, const RecvCB &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 result is no longer valid.
		};

		std::unique_lock<std::mutex> lk(st->mutex);
		if (AsyncSend(remote, msg, timeout_ms, OnRecv) && 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 std::string &topic, bhome::msg::BHAddress &addr, const int timeout_ms)
{
	BHMsg result;
	const BHMsg &msg = MakeQueryTopic(topic);
	if (SyncSendAndRecv(&kBHTopicRPCId, &msg, &result, timeout_ms)) {
		if (result.type() == kMsgTypeQueryTopicReply) {
			DataQueryTopicReply reply;
			if (reply.ParseFromString(result.body())) {
				addr = reply.address();
				return !addr.mq_id().empty();
			}
		}
	}
	return false;
}