| | |
| | | #include <vector> |
| | | |
| | | using namespace bhome_msg; |
| | | |
| | | class ShmSocket : private boost::noncopyable |
| | | { |
| | | template <class... T> |
| | | bool SendImpl(const void *valid_remote, T &&...rest) |
| | | { |
| | | send_buffer_.Append(*static_cast<const MQId *>(valid_remote), std::forward<decltype(rest)>(rest)...); |
| | | return true; |
| | | } |
| | | |
| | | protected: |
| | | typedef bhome_shm::ShmMsgQueue Queue; |
| | | |
| | | public: |
| | | typedef ShmMsgQueue::MQId MQId; |
| | | typedef bhome_shm::SharedMemory Shm; |
| | | typedef std::function<void(ShmSocket &sock, MsgI &imsg, BHMsgHead &head)> RecvCB; |
| | | typedef std::function<bool(ShmSocket &sock, MsgI &imsg, BHMsgHead &head)> PartialRecvCB; |
| | | typedef std::function<void(ShmSocket &sock)> IdleCB; |
| | | |
| | | ShmSocket(Shm &shm, const MQId &id, const int len); |
| | | ShmSocket(Shm &shm, const MQId id, const int len); |
| | | ShmSocket(Shm &shm, const int len = 12); |
| | | ~ShmSocket(); |
| | | static bool Remove(SharedMemory &shm, const MQId &id) { return Queue::Remove(shm, id); } |
| | | static bool Remove(SharedMemory &shm, const MQId id) { return Queue::Remove(shm, id); } |
| | | bool Remove() { return Remove(shm(), id()); } |
| | | const MQId &id() const { return mq().Id(); } |
| | | MQId id() const { return mq().Id(); } |
| | | // start recv. |
| | | bool Start(int nworker = 1, const RecvCB &onData = RecvCB(), const IdleCB &onIdle = IdleCB()); |
| | | bool Start(const RecvCB &onData, const IdleCB &onIdle, int nworker = 1) { return Start(nworker, onData, onIdle); } |
| | | bool Start(const RecvCB &onData, int nworker = 1) { return Start(nworker, onData); } |
| | | bool Stop(); |
| | | size_t Pending() const { return mq().Pending(); } |
| | | |
| | | template <class Body> |
| | | bool Send(const void *valid_remote, BHMsgHead &head, Body &body, RecvCB &&cb = RecvCB()) |
| | | bool Send(const MQId remote, BHMsgHead &head, Body &body, RecvCB &&cb = RecvCB()) |
| | | { |
| | | try { |
| | | if (!cb) { |
| | | return SendImpl(valid_remote, MsgI::Serialize(head, body)); |
| | | return SendImpl(remote, MsgI::Serialize(head, body)); |
| | | } else { |
| | | std::string msg_id(head.msg_id()); |
| | | per_msg_cbs_->Store(msg_id, std::move(cb)); |
| | |
| | | RecvCB cb_no_use; |
| | | per_msg_cbs_->Pick(msg_id, cb_no_use); |
| | | }; |
| | | return SendImpl(valid_remote, MsgI::Serialize(head, body), onExpireRemoveCB); |
| | | return SendImpl(remote, MsgI::Serialize(head, body), onExpireRemoveCB); |
| | | } |
| | | } catch (...) { |
| | | SetLastError(eError, "Send internal error."); |
| | |
| | | } |
| | | } |
| | | |
| | | bool Send(const void *valid_remote, const MsgI &imsg) |
| | | bool Send(const MQId remote, const MsgI &imsg) |
| | | { |
| | | return SendImpl(valid_remote, imsg); |
| | | return SendImpl(remote, imsg); |
| | | } |
| | | |
| | | bool SyncRecv(MsgI &msg, bhome_msg::BHMsgHead &head, const int timeout_ms); |
| | | |
| | | template <class Body> |
| | | bool SendAndRecv(const void *remote, BHMsgHead &head, Body &body, MsgI &reply, BHMsgHead &reply_head, const int timeout_ms) |
| | | bool SendAndRecv(const MQId remote, BHMsgHead &head, Body &body, MsgI &reply, BHMsgHead &reply_head, const int timeout_ms) |
| | | { |
| | | struct State { |
| | | std::mutex mutex; |
| | |
| | | bool StopNoLock(); |
| | | bool RunningNoLock() { return !workers_.empty(); } |
| | | |
| | | template <class... Rest> |
| | | bool SendImpl(const MQId remote, Rest &&...rest) |
| | | { |
| | | send_buffer_.Append(remote, std::forward<decltype(rest)>(rest)...); |
| | | return true; |
| | | } |
| | | |
| | | std::vector<std::thread> workers_; |
| | | std::mutex mutex_; |
| | | std::atomic<bool> run_; |