/*
 * =====================================================================================
 *
 *       Filename:  topic_node.h
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  2021年04月07日 09时05分26秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), lichao@aiotlink.com
 *   Organization:  
 *
 * =====================================================================================
 */
#ifndef TOPIC_NODE_YVKWA6TF
#define TOPIC_NODE_YVKWA6TF

#include "msg.h"
#include "shm_socket.h"
#include <atomic>
#include <memory>
#include <mutex>
#include <vector>

using namespace bhome_shm;
using namespace bhome_msg;

// a node is a client.
class TopicNode
{
	SharedMemory &shm_;
	ProcInfo info_;

	SharedMemory &shm() const { return shm_; }
	const MQInfo &CenterAddr() const { return BHTopicCenterAddress(shm()); }
	const MQInfo &BusAddr() const { return BHTopicBusAddress(shm()); }

public:
	TopicNode(SharedMemory &shm, MQId ssn_id = 0);
	~TopicNode();

	// topic node
	bool Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms) { return DoRegister(false, proc, reply_body, timeout_ms); }
	bool DoRegister(const bool internal, ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
	bool Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
	bool Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
	bool Heartbeat(const int timeout_ms);
	bool QueryTopicAddress(BHAddress &dest, MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms);
	bool QueryProcs(BHAddress &dest, MsgQueryProc &query, MsgQueryProcReply &reply_body, const int timeout_ms);

	// topic rpc server
	typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerSyncCB;
	typedef std::function<void(void *src_info, std::string &client_proc_id, MsgRequestTopic &request)> ServerAsyncCB;
	bool ServerStart(ServerSyncCB const &cb, const int nworker = 2);
	bool ServerStart(ServerAsyncCB const &cb, const int nworker = 2);
	bool ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply, const int timeout_ms) { return DoServerRegisterRPC(false, topics, reply, timeout_ms); }
	bool DoServerRegisterRPC(const bool internal, MsgTopicList &topics, MsgCommonReply &reply, const int timeout_ms);
	bool ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms);
	bool ServerSendReply(void *src_info, const MsgRequestTopicReply &reply);

	// topic client
	typedef std::function<void(const BHMsgHead &head, const MsgRequestTopicReply &reply)> RequestResultCB;
	bool ClientStartWorker(RequestResultCB const &cb, const int nworker = 2);
	bool ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &msg_id, const RequestResultCB &rrcb = RequestResultCB());
	bool ClientSyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &proc_id, MsgRequestTopicReply &reply, const int timeout_ms);

	// publish
	bool Publish(const MsgPublish &pub, const int timeout_ms);

	// subscribe
	typedef std::function<void(const std::string &proc_id, const MsgPublish &data)> SubDataCB;
	bool SubscribeStartWorker(const SubDataCB &tdcb, int nworker = 2);
	bool Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
	{
		return DoSubscribe(topics, false, reply_body, timeout_ms);
	}
	bool SubscribeNet(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
	{
		return DoSubscribe(topics, true, reply_body, timeout_ms);
	}
	bool DoSubscribe(MsgTopicList &topics, const bool net, MsgCommonReply &reply_body, const int timeout_ms);
	bool RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms);

	void Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker = 2);
	void Stop();

private:
	MQId ssn() { return SockNode().id(); }
	bool ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms);
	typedef MsgQueryTopicReply::BHNodeAddress NodeAddress;
	int QueryTopicServers(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms);
	const std::string &proc_id() { return info_.proc_id(); }

	typedef BHAddress Address;
	class TopicQueryCache
	{
		class Impl
		{
			struct TimedRec {
				Address addr_;
				int64_t timestamp_;
			};
			typedef std::unordered_map<Topic, TimedRec> Records;
			Records records_;

		public:
			bool Find(const Topic &topic, Address &addr)
			{
				auto pos = records_.find(topic);
				if (pos != records_.end()) {
					if (NowSec() - pos->second.timestamp_ < NodeTimeoutSec() / 2) {
						addr = pos->second.addr_;
						return true;
					} else {
						LOG_TRACE() << "topic dest cache timeout.";
					}
				}
				return false;
			}
			bool Store(const Topic &topic, const Address &addr)
			{
				records_[topic] = {addr, NowSec()};
				return true;
			}
		};
		Synced<Impl> impl_;

	public:
		bool Find(const Topic &topic, Address &addr) { return impl_->Find(topic, addr); }
		bool Store(const Topic &topic, const Address &addr) { return impl_->Store(topic, addr); }
	};

	// some sockets may be the same one, using functions make it easy to change.
	enum { eSockStart,
		   eSockNode = eSockStart,
		   eSockPub = eSockNode,
		   eSockServer,
		   eSockClient,
		   eSockSub,
		   eSockEnd,
	};
	std::vector<std::shared_ptr<ShmSocket>> sockets_;

	ShmSocket &SockNode() { return *sockets_[eSockNode]; }
	ShmSocket &SockPub() { return *sockets_[eSockPub]; }
	ShmSocket &SockSub() { return *sockets_[eSockSub]; }
	ShmSocket &SockClient() { return *sockets_[eSockClient]; }
	ShmSocket &SockServer() { return *sockets_[eSockServer]; }

	void SetProcIndex(int index)
	{
		proc_index_ = index;
		for (int i = eSockStart; i < eSockEnd; ++i) {
			sockets_[i]->SetNodeProc(index, i);
		}
	}

	enum State {
		eStateUninited,
		eStateUnregistered,
		eStateOnline,
		eStateOffline // heartbeat fail.
	};
	void state(const State st) { state_.store(st); }
	void state_cas(State expected, const State val) { state_.compare_exchange_strong(expected, val); }
	State state() const { return state_.load(); }
	bool IsOnline() { return state() == eStateOnline; }
	bool Init();
	bool Valid() const { return state() != eStateUninited; }
	std::mutex mutex_;
	std::atomic<State> state_;
	MQId ssn_id_ = 0;
	int proc_index_ = -1;

	TopicQueryCache topic_query_cache_;

	class RecvQ
	{
	public:
		void Write(BHMsgHead &&head, std::string &&body) { q_.push_back({std::move(head), std::move(body)}); }
		bool Read(BHMsgHead &head, std::string &body)
		{
			if (q_.empty()) {
				return false;
			} else {
				head = std::move(q_.front().head);
				body = std::move(q_.front().body);
				q_.pop_front();
				return true;
			}
		}

	private:
		struct MsgData {
			BHMsgHead head;
			std::string body;
		};
		std::deque<MsgData> q_;
	};
	Synced<RecvQ> server_buffer_;
	Synced<RecvQ> sub_buffer_;
};

#endif // end of include guard: TOPIC_NODE_YVKWA6TF