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

#include "socket.h"
#include "bh_util.h"
#include "defs.h"
#include "msg.h"

using namespace bhome_msg;
using namespace bhome_shm;

namespace
{

} // namespace

ShmSocket::ShmSocket(Type type, bhome_shm::SharedMemory &shm) :
    shm_(shm), type_(type), run_(false)
{
	switch (type) {
	case eSockBus: mq_.reset(new Queue(kBHBusQueueId, shm_, 1000)); break;
	case eSockRequest: mq_.reset(new Queue(shm_, 12)); break;
	case eSockReply: mq_.reset(new Queue(shm_, 64)); break;
	case eSockSubscribe: mq_.reset(new Queue(shm_, 64)); break;
	case eSockPublish: break; // no recv mq needed
	default: break;
	}
}

ShmSocket::ShmSocket(Type type) :
    ShmSocket(type, BHomeShm()) {}

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

bool ShmSocket::Publish(const std::string &topic, const void *data, const size_t size, const int timeout_ms)
{
	if (type_ != eSockPublish) {
		return false;
	}
	assert(!mq_);
	try {
		MsgI imsg;
		if (!imsg.MakeRC(shm_, MakePub(topic, data, size))) {
			return false;
		}
		DEFER1(imsg.Release(shm_));
		return Queue::Send(shm_, kBHBusQueueId, imsg, timeout_ms);

	} catch (...) {
		return false;
	}
}

bool ShmSocket::Subscribe(const std::vector<std::string> &topics, const int timeout_ms)
{
	if (type_ != eSockSubscribe) {
		return false;
	}
	assert(mq_);
	try {
		return mq_->Send(kBHBusQueueId, MakeSub(mq_->Id(), topics), timeout_ms);
	} catch (...) {
		return false;
	}
}

bool ShmSocket::StartRaw(const RecvRawCB &onData, int nworker)
{
	auto CanRecv = [this]() {
		switch (type_) {
		case eSockRequest:
		case eSockReply:
		case eSockBus:
		case eSockSubscribe:
			return true;
		default:
			return false;
		}
	};
	if (!CanRecv()) {
		return false;
	}
	std::lock_guard<std::mutex> lock(mutex_);

	StopNoLock();
	auto RecvProc = [this, onData]() {
		while (run_) {
			try {
				MsgI imsg;
				DEFER1(imsg.Release(shm_));
				if (mq_->Recv(imsg, 100)) { onData(imsg); }
			} catch (...) {
			}
		}
	};

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

bool ShmSocket::Start(const RecvCB &onData, int nworker)
{
	return StartRaw([this, onData](MsgI &imsg) { BHMsg m; if (imsg.Unpack(m)) { onData(m); } }, nworker);
}

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();
			}
		}
		return true;
	}
	return false;
}