| | |
| | | */ |
| | | |
| | | #include "socket.h" |
| | | #include <chrono> |
| | | #include "msg.h" |
| | | #include "defs.h" |
| | | #include "bh_util.h" |
| | | #include "defs.h" |
| | | #include "msg.h" |
| | | #include <chrono> |
| | | |
| | | using namespace bhome_msg; |
| | | using namespace bhome_shm; |
| | | using namespace std::chrono_literals; |
| | | |
| | | namespace { |
| | | |
| | | int GetSocketDefaultLen(ShmSocket::Type type) |
| | | namespace |
| | | { |
| | | switch (type) { |
| | | case ShmSocket::eSockRequest : return 12; |
| | | case ShmSocket::eSockReply : return 64; |
| | | case ShmSocket::eSockPublish : return 0; |
| | | case ShmSocket::eSockSubscribe : return 64; |
| | | default: return 0; |
| | | } |
| | | } |
| | | |
| | | |
| | | } |
| | | |
| | | ShmSocket::ShmSocket(Type type, bhome_shm::SharedMemory &shm) |
| | | : shm_(shm), |
| | | type_(type), |
| | | run_(false) |
| | | int GetSocketDefaultLen(ShmSocket::Type type) |
| | | { |
| | | int len = GetSocketDefaultLen(type); |
| | | if (len != 0) { |
| | | mq_.reset(new Queue(shm_, len)); |
| | | |
| | | auto RecvProc = [this](){ |
| | | while (run_) { |
| | | try { |
| | | std::unique_lock<std::mutex> lk(mutex_); |
| | | if (cv_recv_cb_.wait_for(lk, 100ms, [this](){ return HasRecvCB(); })) { |
| | | BHMsg msg; |
| | | if (mq_->Recv(msg, 100)) { |
| | | this->onRecv_(msg); |
| | | } |
| | | } |
| | | } catch (...) {} |
| | | } |
| | | }; |
| | | run_.store(true); |
| | | workers_.emplace_back(RecvProc); |
| | | } |
| | | switch (type) { |
| | | case ShmSocket::eSockRequest: return 12; |
| | | case ShmSocket::eSockReply: return 64; |
| | | case ShmSocket::eSockPublish: return 0; |
| | | case ShmSocket::eSockSubscribe: return 64; |
| | | default: return 0; |
| | | } |
| | | } |
| | | ShmSocket::ShmSocket(Type type) |
| | | : ShmSocket(type, BHomeShm()) |
| | | |
| | | } // namespace |
| | | |
| | | ShmSocket::ShmSocket(Type type, bhome_shm::SharedMemory &shm) : |
| | | shm_(shm), type_(type), run_(false) |
| | | { |
| | | int len = GetSocketDefaultLen(type); |
| | | if (len != 0) { |
| | | mq_.reset(new Queue(shm_, len)); |
| | | |
| | | auto RecvProc = [this]() { |
| | | while (run_) { |
| | | try { |
| | | std::unique_lock<std::mutex> lk(mutex_); |
| | | if (cv_recv_cb_.wait_for(lk, 100ms, [this]() { return HasRecvCB(); })) { |
| | | BHMsg msg; |
| | | if (mq_->Recv(msg, 100)) { |
| | | this->onRecv_(msg); |
| | | } |
| | | } |
| | | } catch (...) { |
| | | } |
| | | } |
| | | }; |
| | | run_.store(true); |
| | | workers_.emplace_back(RecvProc); |
| | | } |
| | | } |
| | | |
| | | ShmSocket::ShmSocket(Type type) : |
| | | ShmSocket(type, BHomeShm()) |
| | | { |
| | | } |
| | | |
| | | ShmSocket::~ShmSocket() |
| | | { |
| | | Stop(); |
| | | Stop(); |
| | | } |
| | | |
| | | bool ShmSocket::Publish(const std::string &topic, const void *data, const size_t size, const int timeout_ms) |
| | | { |
| | | if (type_ != eSockPublish) { |
| | | return false; |
| | | } |
| | | 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; |
| | | } |
| | | if (type_ != eSockPublish) { |
| | | return false; |
| | | } |
| | | 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; |
| | | } |
| | | if (type_ != eSockSubscribe) { |
| | | return false; |
| | | } |
| | | assert(mq_); |
| | | try { |
| | | return mq_->Send(kBHBusQueueId, MakeSub(mq_->Id(), topics), timeout_ms); |
| | | } catch (...) { |
| | | return false; |
| | | } |
| | | } |
| | | |
| | | bool ShmSocket::SetRecvCallback(const RecvCB &onRecv) |
| | | { |
| | | std::lock_guard<std::mutex> lock(mutex_); |
| | | onRecv_ = onRecv; |
| | | cv_recv_cb_.notify_one(); |
| | | return true; |
| | | std::lock_guard<std::mutex> lock(mutex_); |
| | | onRecv_ = onRecv; |
| | | cv_recv_cb_.notify_one(); |
| | | return true; |
| | | } |
| | | |
| | | bool ShmSocket::HasRecvCB() |
| | | { |
| | | return static_cast<bool>(onRecv_); |
| | | return static_cast<bool>(onRecv_); |
| | | } |
| | | |
| | | void ShmSocket::Stop() |
| | | { |
| | | run_ = false; |
| | | for (auto &t : workers_) { |
| | | if (t.joinable()) { |
| | | t.join(); |
| | | } |
| | | } |
| | | run_ = false; |
| | | for (auto &t : workers_) { |
| | | if (t.joinable()) { |
| | | t.join(); |
| | | } |
| | | } |
| | | } |