/*
 * =====================================================================================
 *
 *       Filename:  shm_socket.cpp
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  2021年03月30日 15时48分58秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), 
 *   Organization:  
 *
 * =====================================================================================
 */

#include "shm_socket.h"
#include "bh_util.h"
#include "defs.h"
#include "msg.h"
#include <chrono>
using namespace std::chrono;
using namespace std::chrono_literals;

using namespace bhome_msg;
using namespace bhome_shm;

ShmSocket::ShmSocket(Shm &shm, const MQId id, const int len) :
    run_(false), mq_(shm, id, len), alloc_id_(0) { Start(); }
ShmSocket::ShmSocket(Shm &shm, const bool create_or_else_find, const MQId id, const int len) :
    run_(false), mq_(shm, create_or_else_find, id, len), alloc_id_(0) { Start(); }
ShmSocket::ShmSocket(int64_t abs_addr, Shm &shm, const MQId id) :
    run_(false), mq_(abs_addr, shm, id), alloc_id_(0) { Start(); }

ShmSocket::~ShmSocket() { Stop(); }

bool ShmSocket::Start(int nworker, const RecvCB &onData, const RawRecvCB &onRaw, const IdleCB &onIdle)
{
	auto ioProc = [this, onData, onRaw, onIdle]() {
		auto DoSend = [this]() { return send_buffer_.TrySend(mq()); };
		auto DoRecv = [=] {
			// do not recv if no cb is set.
			if (!onData && per_msg_cbs_->empty() && !onRaw && alloc_cbs_->empty()) { return false; }

			auto onMsgCB = [this, onData](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) {
				RecvCB cb;
				if (per_msg_cbs_->Pick(head.msg_id(), cb)) {
					cb(socket, imsg, head);
				} else if (onData) {
					onData(socket, imsg, head);
				}
			};
			auto onCmdCB = [this, onRaw](ShmSocket &socket, int64_t val) {
				int cmd = DecodeCmd(val);
				if (cmd == eCmdAllocReply0) {
					int id = (val >> 4) & MaskBits(28);
					RawRecvCB cb;
					if (alloc_cbs_->Pick(id, cb)) {
						cb(socket, val);
						return;
					}
				}
				if (onRaw) {
					onRaw(socket, val);
				}
			};

			auto onRecv = [&](auto &val) {
				if (IsCmd(val)) {
					onCmdCB(*this, val);
				} else {
					MsgI imsg(val);
					DEFER1(imsg.Release());
					BHMsgHead head;
					if (imsg.ParseHead(head)) {
						onMsgCB(*this, imsg, head);
					}
				}
			};
			ShmMsgQueue::RawData val = 0;
			for (int i = 0; i < 100; ++i) {
				if (mq().TryRecv(val)) {
					onRecv(val);
					return true;
				}
			}
			return false;
		};

		try {
			bool more_to_send = DoSend();
			bool more_to_recv = DoRecv();
			if (onIdle) { onIdle(*this); }
			if (!more_to_send && !more_to_recv) {
				robust::QuickSleep();
			}
		} catch (...) {
		}
	};

	std::lock_guard<std::mutex> lock(mutex_);
	StopNoLock();

	auto worker_proc = [this, ioProc]() {
		while (run_) { ioProc(); }
		// try send pending msgs.
		auto end_time = steady_clock::now() + 3s;
		while (send_buffer_.TrySend(mq()) && steady_clock::now() < end_time) {
			// LOG_DEBUG() << "try send pending msgs.";
		}
	};

	run_.store(true);
	for (int i = 0; i < nworker; ++i) {
		workers_.emplace_back(worker_proc);
	}
	return true;
}

bool ShmSocket::Stop()
{
	std::lock_guard<std::mutex> lock(mutex_);
	return StopNoLock();
}

bool ShmSocket::StopNoLock()
{
	if (run_.exchange(false)) {
		for (auto &w : workers_) {
			if (w.joinable()) {
				w.join();
			}
		}
		workers_.clear();
		return true;
	}
	return false;
}

bool ShmSocket::Send(const MQInfo &remote, std::string &&content, const std::string &msg_id, RecvCB &&cb)
{
	size_t size = content.size();
	auto OnResult = [content = std::move(content), msg_id, remote, cb = std::move(cb), this](MsgI &msg) mutable {
		if (!msg.Fill(content)) { return false; }

		try {
			if (!cb) {
				Send(remote, msg);
			} else {
				per_msg_cbs_->Store(msg_id, std::move(cb));
				auto onExpireRemoveCB = [this, msg_id](SendQ::Data const &msg) {
					RecvCB cb_no_use;
					per_msg_cbs_->Pick(msg_id, cb_no_use);
				};
				Send(remote, msg, onExpireRemoveCB);
			}
			return true;
		} catch (...) {
			SetLastError(eError, "Send internal error.");
			return false;
		}
	};
#if 0
	// self alloc
	MsgI msg;
	if (msg.Make(size)) {
		DEFER1(msg.Release());
		return OnResult(msg);
	}
#else
	// center alloc
	return RequestAlloc(size, OnResult);
#endif
}

bool ShmSocket::RequestAlloc(const int64_t size, std::function<void(MsgI &msg)> const &onResult)
{ // 8bit size, 4bit socket index, 16bit proc index, 28bit id, ,4bit cmd+flag
	// LOG_FUNCTION;
	if (node_proc_index_ == -1 || socket_index_ == -1) {
		return false;
	}
	int id = (++alloc_id_) & MaskBits(28);
	int64_t cmd = (CalcAllocIndex(size) << 52) |
	              ((socket_index_ & MaskBits(4)) << 48) |
	              ((node_proc_index_ & MaskBits(16)) << 32) |
	              (id << 4) |
	              EncodeCmd(eCmdAllocRequest0);
	auto rawCB = [onResult](ShmSocket &sock, int64_t &val) {
		MsgI msg((val >> 32) & MaskBits(31));
		DEFER1(msg.Release());
		onResult(msg);
		return true;
	};

	alloc_cbs_->Store(id, std::move(rawCB));
	auto onExpireRemoveCB = [this, id](SendQ::Data const &msg) {
		RawRecvCB cb_no_use;
		alloc_cbs_->Pick(id, cb_no_use);
	};

	return Send(BHTopicCenterAddress(), cmd, onExpireRemoveCB);
}