/*
 * =====================================================================================
 *
 *       Filename:  reqrep_center.cpp
 *
 *    Description:  topic request/reply center
 *
 *        Version:  1.0
 *        Created:  2021年04月01日 14时08分50秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), 
 *   Organization:  
 *
 * =====================================================================================
 */
#include "reqrep_center.h"
#include "bh_util.h"
using namespace bhome_shm;

struct A {
	void F(int){};
};

namespace
{
inline uint64_t Now()
{
	time_t t;
	return time(&t);
}

} // namespace
bool ReqRepCenter::Start(const int nworker)
{
	auto onRecv = [&](BHMsg &msg) {
#ifndef NDEBUG
		static std::atomic<time_t> last(0);
		time_t now = 0;
		time(&now);
		if (last.exchange(now) < now) {
			printf("bus queue size: %ld\n", socket_.Pending());
		}
#endif
		if (msg.route_size() == 0) {
			return;
		}
		auto &src_mq = msg.route(0).mq_id();

		auto OnRegister = [&]() {
			DataProcRegister reg;
			if (!reg.ParseFromString(msg.body())) {
				return;
			}
			ProcInfo pi;
			pi.server_mqid_ = src_mq;
			pi.proc_id_ = reg.proc().name();
			pi.ext_info_ = reg.proc().info();
			pi.timestamp_ = Now();

			std::lock_guard<std::mutex> lock(mutex_);
			for (auto &t : reg.topics()) {
				topic_mq_[t] = pi.server_mqid_;
			}
			procs_[pi.proc_id_] = pi;
		};

		auto OnHeartbeat = [&]() {
			DataProcHeartbeat hb;
			if (!hb.ParseFromString(msg.body())) {
				return;
			}

			std::lock_guard<std::mutex> lock(mutex_);
			auto pos = procs_.find(hb.proc().name());
			if (pos != procs_.end() && pos->second.server_mqid_ == src_mq) { // both name and mq should be the same.
				pos->second.timestamp_ = Now();
				pos->second.ext_info_ = hb.proc().info();
			}
		};

		auto OnQueryTopic = [&]() {
			DataProcQueryTopic query;
			if (!query.ParseFromString(msg.body())) {
				return;
			}

			std::string dest;
			auto FindDest = [&]() {
				std::lock_guard<std::mutex> lock(mutex_);
				auto pos = topic_mq_.find(query.topic());
				if (pos != topic_mq_.end()) {
					dest = pos->second;
					return true;
				} else {
					return false;
				}
			};
			if (FindDest()) {
				MQId remote;
				memcpy(&remote, msg.route().rbegin()->mq_id().data(), sizeof(remote));
				MsgI imsg;
				if (!imsg.Make(shm(), MakeQueryTopicReply(dest, msg.msg_id()))) { return; }
				if (!ShmMsgQueue::Send(shm(), remote, imsg, 100)) {
					imsg.Release(shm());
				}
			}
		};

		switch (msg.type()) {
		case kMsgTypeProcRegisterTopics: OnRegister(); break;
		case kMsgTypeProcHeartbeat: OnHeartbeat(); break;
		case kMsgTypeProcQueryTopic: OnQueryTopic(); break;
		default: break;
		}
	};

	const int kMaxWorker = 16;
	return socket_.Start(onRecv, std::min((nworker > 0 ? nworker : 2), kMaxWorker));
}