/*
|
* =====================================================================================
|
*
|
* 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 "defs.h"
|
#include "msg.h"
|
#include "shm_queue.h"
|
|
using namespace bhome_shm;
|
using namespace bhome_msg;
|
|
#define BH_USE_ATOMIC_Q
|
|
class 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;
|
|
ShmMsgQueue(ShmType &segment, const MQId id, const int len);
|
ShmMsgQueue(ShmType &segment, const bool create_or_else_find, const MQId id, const int len);
|
ShmMsgQueue(const int64_t abs_addr, ShmType &segment, const MQId id);
|
~ShmMsgQueue();
|
static bool Remove(ShmType &shm, const MQId id);
|
MQId Id() const { return id_; }
|
ShmType &shm() const { return queue_.shm(); }
|
int64_t AbsAddr() const { return queue_.offset(); }
|
|
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);
|
static bool TrySend(ShmType &shm, const MQInfo &remote, const RawData val);
|
bool TrySend(const MQInfo &remote, const RawData val) { return TrySend(shm(), remote, 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
|
