/*
 * =====================================================================================
 *
 *       Filename:  shm_msg_queue.h
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  2021年04月26日 16时25分21秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), lichao@aiotlink.com
 *   Organization:  
 *
 * =====================================================================================
 */
#ifndef SHM_MSG_QUEUE_D847TQXH
#define SHM_MSG_QUEUE_D847TQXH

#include "msg.h"
#include "shm_queue.h"

using namespace bhome_shm;
using namespace bhome_msg;

#define BH_USE_ATOMIC_Q

class ShmMsgQueue : public StaticDataRef<std::atomic<uint64_t>, ShmMsgQueue>
{
public:
	typedef int64_t RawData;

#ifdef BH_USE_ATOMIC_Q
	typedef ShmObject<SharedQ63<0>> Shmq;
#else
	typedef ShmObject<SharedQueue<RawData>> Shmq;
	// typedef robust::FMutex Mutex;
	// typedef robust::SemMutex Mutex;
	typedef robust::NullMutex Mutex;
	typedef robust::Guard<Mutex> Guard;
#endif

	typedef Shmq::Data Queue;
	typedef Shmq::ShmType ShmType;
	typedef uint64_t MQId;

	static MQId NewId();

	ShmMsgQueue(const MQId id, ShmType &segment, const int len);
	ShmMsgQueue(const MQId id, const bool create_or_else_find, ShmType &segment, const int len);
	ShmMsgQueue(ShmType &segment, const int len);
	~ShmMsgQueue();
	static bool Remove(ShmType &shm, const MQId id);
	MQId Id() const { return id_; }
	ShmType &shm() const { return queue_.shm(); }

	bool Recv(RawData &val, const int timeout_ms)
	{
#ifndef BH_USE_ATOMIC_Q
		Guard lock(GetMutex(Id()));
#endif
		return queue().Read(val, timeout_ms);
	}

	bool TryRecv(RawData &val)
	{
#ifndef BH_USE_ATOMIC_Q
		Guard lock(GetMutex(Id()));
#endif
		return queue().TryRead(val);
	}

	bool Recv(MsgI &msg, const int timeout_ms) { return Recv(msg.OffsetRef(), timeout_ms); }
	bool TryRecv(MsgI &msg) { return TryRecv(msg.OffsetRef()); }
	static Queue *Find(ShmType &shm, const MQId remote_id);
	static bool TrySend(ShmType &shm, const MQId remote_id, const RawData val);
	static bool TrySend(ShmType &shm, const MQId remote_id, MsgI msg)
	{
		bool r = false;
		msg.AddRef(); // TODO check if we could avoid addref here.
		DEFER1(if (!r) { msg.Release(); });
		r = TrySend(shm, remote_id, msg.Offset());
		return r;
	}
	bool TrySend(const MQId remote_id, const MsgI &msg) { return TrySend(shm(), remote_id, msg); }
	bool TrySend(const MQId remote_id, const RawData val) { return TrySend(shm(), remote_id, val); }

private:
#ifndef BH_USE_ATOMIC_Q
	static Mutex &GetMutex(const MQId id);
#endif
	MQId id_;
	Queue &queue() { return *queue_.data(); }
	Shmq queue_;
};

#endif // end of include guard: SHM_MSG_QUEUE_D847TQXH