| | |
| | | |
| | | } // namespace |
| | | |
| | | ShmSocket::ShmSocket(Type type, bhome_shm::SharedMemory &shm) : |
| | | shm_(shm), type_(type), run_(false) |
| | | ShmSocket::ShmSocket(Shm &shm, const void *id, const int len) : |
| | | shm_(shm), 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; |
| | | if (id && len > 0) { |
| | | mq_.reset(new Queue(*static_cast<const MQId *>(id), shm, len)); |
| | | } |
| | | } |
| | | |
| | | ShmSocket::ShmSocket(Type type) : |
| | | ShmSocket(type, BHomeShm()) {} |
| | | ShmSocket::ShmSocket(bhome_shm::SharedMemory &shm, const int len) : |
| | | shm_(shm), run_(false) |
| | | { |
| | | if (len > 0) { |
| | | mq_.reset(new Queue(shm_, len)); |
| | | } |
| | | } |
| | | |
| | | ShmSocket::~ShmSocket() |
| | | { |
| | | Stop(); |
| | | Stop(); //TODO should stop in sub class, incase thread access sub class data. |
| | | } |
| | | |
| | | bool ShmSocket::Publish(const std::string &topic, const void *data, const size_t size, const int timeout_ms) |
| | | bool ShmSocket::Start(const RecvCB &onData, const IdleCB &onIdle, int nworker) |
| | | { |
| | | if (type_ != eSockPublish) { |
| | | return false; |
| | | if (!mq_ || !onData) { |
| | | return false; // TODO error code. |
| | | } |
| | | 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]() { |
| | | 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() |
| | |
| | | w.join(); |
| | | } |
| | | } |
| | | workers_.clear(); |
| | | return true; |
| | | } |
| | | return false; |
| | | } |
| | | |
| | | bool ShmSocket::SyncSend(const void *id, const bhome_msg::BHMsg &msg, const int timeout_ms) |
| | | { |
| | | return mq_->Send(*static_cast<const MQId *>(id), msg, timeout_ms); |
| | | } |
| | | |
| | | bool ShmSocket::SyncRecv(bhome_msg::BHMsg &msg, const int timeout_ms) |
| | | { |
| | | std::lock_guard<std::mutex> lock(mutex_); |
| | | if (!mq_ || RunningNoLock()) { |
| | | return false; |
| | | } else { |
| | | return mq_->Recv(msg, timeout_ms); |
| | | } |
| | | } |