| | |
| | | } |
| | | }; |
| | | ShmMsgQueue::RawData val = 0; |
| | | auto TryRecvMore = [&]() { |
| | | for (int i = 0; i < 100; ++i) { |
| | | if (mq().TryRecv(val)) { |
| | | return true; |
| | | } |
| | | for (int i = 0; i < 100; ++i) { |
| | | if (mq().TryRecv(val)) { |
| | | onRecv(val); |
| | | return true; |
| | | } |
| | | return false; |
| | | }; |
| | | return TryRecvMore() ? (onRecv(val), true) : false; |
| | | } |
| | | return false; |
| | | }; |
| | | |
| | | try { |
| | |
| | | return false; |
| | | } |
| | | |
| | | bool ShmSocket::Send(const MQInfo &remote, std::string &&content, const std::string &msg_id, RecvCB &&cb) |
| | | { |
| | | size_t size = content.size(); |
| | | auto OnResult = [content = std::move(content), msg_id, remote, cb = std::move(cb), this](MsgI &msg) mutable { |
| | | if (!msg.Fill(content)) { return; } |
| | | |
| | | try { |
| | | if (!cb) { |
| | | Send(remote, msg); |
| | | } else { |
| | | per_msg_cbs_->Store(msg_id, std::move(cb)); |
| | | auto onExpireRemoveCB = [this, msg_id](SendQ::Data const &msg) { |
| | | RecvCB cb_no_use; |
| | | per_msg_cbs_->Pick(msg_id, cb_no_use); |
| | | }; |
| | | Send(remote, msg, onExpireRemoveCB); |
| | | } |
| | | } catch (...) { |
| | | SetLastError(eError, "Send internal error."); |
| | | } |
| | | }; |
| | | |
| | | return RequestAlloc(size, OnResult); |
| | | } |
| | | |
| | | bool ShmSocket::RequestAlloc(const int64_t size, std::function<void(MsgI &msg)> const &onResult) |
| | | { // 8bit size, 4bit socket index, 16bit proc index, 28bit id, ,4bit cmd+flag |
| | | // LOG_FUNCTION; |
| | |
| | | RawRecvCB cb_no_use; |
| | | alloc_cbs_->Pick(id, cb_no_use); |
| | | }; |
| | | |
| | | return Send(BHTopicCenterAddress(), cmd, onExpireRemoveCB); |
| | | } |