| | |
| | | |
| | | class ShmSocket : private boost::noncopyable |
| | | { |
| | | bool SendImpl(const void *valid_remote, MsgI const &imsg, SendQ::OnMsgEvent onExpire = SendQ::OnMsgEvent()) |
| | | template <class... T> |
| | | bool SendImpl(const void *valid_remote, T &&...rest) |
| | | { |
| | | // if (!mq().TrySend(*(MQId *) valid_remote, imsg)) { |
| | | send_buffer_.Append(*static_cast<const MQId *>(valid_remote), imsg, onExpire); |
| | | // } |
| | | send_buffer_.Append(*static_cast<const MQId *>(valid_remote), std::forward<decltype(rest)>(rest)...); |
| | | return true; |
| | | } |
| | | |
| | |
| | | template <class Body> |
| | | bool Send(const void *valid_remote, const BHMsgHead &head, const Body &body, const RecvCB &cb = RecvCB()) |
| | | { |
| | | MsgI msg; |
| | | if (msg.Make(shm(), head, body)) { |
| | | DEFER1(if (msg.IsCounted()) { msg.Release(shm()); }); |
| | | std::string msg_id(head.msg_id()); |
| | | try { |
| | | if (!cb) { |
| | | return SendImpl(valid_remote, msg); |
| | | return SendImpl(valid_remote, MsgI::Serialize(head, body)); |
| | | } else { |
| | | std::string msg_id(head.msg_id()); |
| | | per_msg_cbs_->Add(msg_id, cb); |
| | | auto onExpireRemoveCB = [this, msg_id](MsgI const &msg) { |
| | | auto onExpireRemoveCB = [this, msg_id](SendQ::Data const &msg) { |
| | | RecvCB cb_no_use; |
| | | per_msg_cbs_->Find(msg_id, cb_no_use); |
| | | }; |
| | | return SendImpl(valid_remote, msg, onExpireRemoveCB); |
| | | return SendImpl(valid_remote, MsgI::Serialize(head, body), onExpireRemoveCB); |
| | | } |
| | | } else { |
| | | SetLastError(ENOMEM, "Out of mem"); |
| | | } catch (...) { |
| | | SetLastError(eError, "Send internal error."); |
| | | return false; |
| | | } |
| | | return false; |
| | | } |
| | | |
| | | bool Send(const void *valid_remote, const MsgI &imsg) |