/*
|
* =====================================================================================
|
*
|
* Filename: reqrep.cpp
|
*
|
* Description: topic request/reply sockets
|
*
|
* Version: 1.0
|
* Created: 2021年04月01日 09时35分35秒
|
* Revision: none
|
* Compiler: gcc
|
*
|
* Author: Li Chao (),
|
* Organization:
|
*
|
* =====================================================================================
|
*/
|
#include "reqrep.h"
|
#include "bh_util.h"
|
#include "msg.h"
|
#include <chrono>
|
#include <condition_variable>
|
|
using namespace bhome_msg;
|
|
bool SocketRequest::StartWorker(const RequestResultCB &rrcb, int nworker)
|
{
|
auto AsyncRecvProc = [this, rrcb](BHMsg &msg) {
|
auto Find = [&](RecvCB &cb) {
|
std::lock_guard<std::mutex> lock(mutex());
|
const std::string &msgid = msg.msg_id();
|
auto pos = async_cbs_.find(msgid);
|
if (pos != async_cbs_.end()) {
|
cb.swap(pos->second);
|
async_cbs_.erase(pos);
|
return true;
|
} else {
|
return false;
|
}
|
};
|
|
RecvCB cb;
|
if (Find(cb) && cb) {
|
cb(msg);
|
} else if (rrcb && msg.type() == kMsgTypeReply) {
|
DataReply reply;
|
if (reply.ParseFromString(msg.body())) {
|
rrcb(reply.data());
|
}
|
} else {
|
// ignored, or dropped
|
}
|
};
|
|
return Start(AsyncRecvProc, nworker);
|
}
|
|
bool SocketRequest::AsyncRequest(const std::string &topic, const void *data, const size_t size, const int timeout_ms, const RequestResultCB &cb)
|
{
|
auto Call = [&](const void *remote) {
|
const BHMsg &msg(MakeRequest(mq().Id(), topic, data, size));
|
auto onRecv = [cb](BHMsg &msg) {
|
if (msg.type() == kMsgTypeReply) {
|
DataReply reply;
|
if (reply.ParseFromString(msg.body())) {
|
cb(reply.data());
|
}
|
}
|
};
|
return AsyncSend(remote, &msg, timeout_ms, onRecv);
|
};
|
|
try {
|
BHAddress addr;
|
if (QueryRPCTopic(topic, addr, timeout_ms)) {
|
return Call(addr.mq_id().data());
|
} else {
|
return false;
|
}
|
} catch (...) {
|
return false;
|
}
|
}
|
|
bool SocketRequest::SyncRequest(const std::string &topic, const void *data, const size_t size, std::string &out, const int timeout_ms)
|
{
|
try {
|
BHAddress addr;
|
if (QueryRPCTopic(topic, addr, timeout_ms)) {
|
const BHMsg &req(MakeRequest(mq().Id(), topic, data, size));
|
BHMsg reply;
|
if (SyncSendAndRecv(addr.mq_id().data(), &req, &reply, timeout_ms) && reply.type() == kMsgTypeReply) {
|
DataReply dr;
|
if (dr.ParseFromString(reply.body())) {
|
dr.mutable_data()->swap(out);
|
return true;
|
}
|
}
|
} else {
|
}
|
} catch (...) {
|
}
|
return false;
|
}
|
|
bool SocketRequest::AsyncSend(const void *remote, const void *pmsg, const int timeout_ms, const RecvCB &cb)
|
{
|
assert(remote && pmsg);
|
try {
|
const BHMsg &msg = *static_cast<const BHMsg *>(pmsg);
|
auto RegisterCB = [&]() {
|
std::lock_guard<std::mutex> lock(mutex());
|
async_cbs_.emplace(msg.msg_id(), cb);
|
};
|
|
return mq().Send(*static_cast<const MQId *>(remote), msg, timeout_ms, RegisterCB);
|
} catch (...) {
|
return false;
|
}
|
}
|
|
bool SocketRequest::SyncSendAndRecv(const void *remote, const void *msg, void *result, const int timeout_ms)
|
{
|
struct State {
|
std::mutex mutex;
|
std::condition_variable cv;
|
bool canceled = false;
|
};
|
|
try {
|
std::shared_ptr<State> st(new State);
|
auto endtime = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);
|
|
auto OnRecv = [=](BHMsg &msg) {
|
std::unique_lock<std::mutex> lk(st->mutex);
|
if (!st->canceled) {
|
static_cast<BHMsg *>(result)->Swap(&msg);
|
st->cv.notify_one();
|
} else {
|
}
|
};
|
|
std::unique_lock<std::mutex> lk(st->mutex);
|
bool sendok = AsyncSend(remote, msg, timeout_ms, OnRecv);
|
if (sendok && st->cv.wait_until(lk, endtime) == std::cv_status::no_timeout) {
|
return true;
|
} else {
|
st->canceled = true;
|
return false;
|
}
|
} catch (...) {
|
return false;
|
}
|
}
|
|
bool SocketRequest::QueryRPCTopic(const std::string &topic, bhome::msg::BHAddress &addr, const int timeout_ms)
|
{
|
if (tmp_cache_.first == topic) {
|
addr = tmp_cache_.second;
|
return true;
|
}
|
|
BHMsg result;
|
const BHMsg &msg = MakeQueryTopic(mq().Id(), topic);
|
if (SyncSendAndRecv(&kBHTopicReqRepCenter, &msg, &result, timeout_ms)) {
|
if (result.type() == kMsgTypeProcQueryTopicReply) {
|
DataProcQueryTopicReply reply;
|
if (reply.ParseFromString(result.body())) {
|
addr = reply.address();
|
tmp_cache_.first = topic;
|
tmp_cache_.second = addr;
|
return !addr.mq_id().empty();
|
}
|
}
|
} else {
|
}
|
return false;
|
}
|
|
// reply socket
|
namespace
|
{
|
struct SrcInfo {
|
std::vector<BHAddress> route;
|
std::string msg_id;
|
};
|
|
} // namespace
|
|
bool SocketReply::Register(const ProcInfo &proc_info, const std::vector<std::string> &topics, const int timeout_ms)
|
{
|
//TODO check reply?
|
return SyncSend(&kBHTopicReqRepCenter, MakeRegister(mq().Id(), proc_info, topics), timeout_ms);
|
}
|
bool SocketReply::Heartbeat(const ProcInfo &proc_info, const int timeout_ms)
|
{
|
return SyncSend(&kBHTopicReqRepCenter, MakeHeartbeat(mq().Id(), proc_info), timeout_ms);
|
}
|
bool SocketReply::StartWorker(const OnRequest &rcb, int nworker)
|
{
|
auto onRecv = [this, rcb](BHMsg &msg) {
|
if (msg.type() == kMsgTypeRequest && msg.route_size() > 0) {
|
DataRequest req;
|
if (req.ParseFromString(msg.body())) {
|
std::string out;
|
if (rcb(req.topic(), req.data(), out)) {
|
BHMsg msg_reply(MakeReply(msg.msg_id(), out.data(), out.size()));
|
for (int i = 0; i < msg.route_size() - 1; ++i) {
|
msg.add_route()->Swap(msg.mutable_route(i));
|
}
|
SyncSend(msg.route().rbegin()->mq_id().data(), msg_reply, 100);
|
}
|
}
|
} else {
|
// ignored, or dropped
|
}
|
};
|
|
return rcb && Start(onRecv, nworker);
|
}
|
|
bool SocketReply::RecvRequest(void *&src_info, std::string &topic, std::string &data, const int timeout_ms)
|
{
|
BHMsg msg;
|
if (SyncRecv(msg, timeout_ms) && msg.type() == kMsgTypeRequest) {
|
DataRequest request;
|
if (request.ParseFromString(msg.body())) {
|
request.mutable_topic()->swap(topic);
|
request.mutable_data()->swap(data);
|
SrcInfo *p = new SrcInfo;
|
p->route.assign(msg.route().begin(), msg.route().end());
|
p->msg_id = msg.msg_id();
|
src_info = p;
|
return true;
|
}
|
}
|
return false;
|
}
|
|
bool SocketReply::SendReply(void *src_info, const std::string &data, const int timeout_ms)
|
{
|
SrcInfo *p = static_cast<SrcInfo *>(src_info);
|
DEFER1(delete p);
|
if (!p || p->route.empty()) {
|
return false;
|
}
|
|
BHMsg msg(MakeReply(p->msg_id, data.data(), data.size()));
|
for (unsigned i = 0; i < p->route.size() - 1; ++i) {
|
msg.add_route()->Swap(&p->route[i]);
|
}
|
|
return SyncSend(p->route.back().mq_id().data(), msg, timeout_ms);
|
}
|
