| | |
| | | |
| | | ShmSocket::~ShmSocket() |
| | | { |
| | | Stop(); |
| | | Stop(); //TODO should stop in sub class, incase thread access sub class data. |
| | | } |
| | | |
| | | bool ShmSocket::StartRaw(const RecvRawCB &onData, int nworker) |
| | | bool ShmSocket::Start(const RecvCB &onData, const IdleCB &onIdle, int nworker) |
| | | { |
| | | if (!mq_) { |
| | | return false; |
| | | if (!mq_ || !onData) { |
| | | return false; // TODO error code. |
| | | } |
| | | |
| | | std::lock_guard<std::mutex> lock(mutex_); |
| | | StopNoLock(); |
| | | auto RecvProc = [this, onData]() { |
| | | auto RecvProc = [this, onData, onIdle]() { |
| | | while (run_) { |
| | | try { |
| | | MsgI imsg; |
| | | DEFER1(imsg.Release(shm_)); |
| | | if (mq_->Recv(imsg, 100)) { onData(imsg); } |
| | | if (mq_->Recv(imsg, 100)) { |
| | | BHMsg msg; |
| | | if (imsg.Unpack(msg)) { |
| | | onData(*this, imsg, msg); |
| | | } |
| | | } else if (onIdle) { |
| | | onIdle(*this); |
| | | } |
| | | } catch (...) { |
| | | } |
| | | } |
| | |
| | | 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() |
| | |
| | | |
| | | bool ShmSocket::SyncSend(const void *id, const bhome_msg::BHMsg &msg, const int timeout_ms) |
| | | { |
| | | std::lock_guard<std::mutex> lock(mutex_); |
| | | if (!mq_ || RunningNoLock()) { |
| | | return false; |
| | | } else { |
| | | return mq_->Send(*static_cast<const MQId *>(id), msg, timeout_ms); |
| | | } |
| | | return mq_->Send(*static_cast<const MQId *>(id), msg, timeout_ms); |
| | | } |
| | | |
| | | bool ShmSocket::SyncRecv(bhome_msg::BHMsg &msg, const int timeout_ms) |