/*
|
* =====================================================================================
|
*
|
* Filename: topic_request.cpp
|
*
|
* Description: topic request sockets
|
*
|
* Version: 1.0
|
* Created: 2021年04月01日 09时35分35秒
|
* Revision: none
|
* Compiler: gcc
|
*
|
* Author: Li Chao (),
|
* Organization:
|
*
|
* =====================================================================================
|
*/
|
#include "topic_request.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 = [&](RecvBHMsgCB &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;
|
}
|
};
|
|
RecvBHMsgCB cb;
|
if (Find(cb) && cb) {
|
cb(msg);
|
} else if (msg.type() == kMsgTypeRequestTopicReply && rrcb) {
|
MsgRequestTopicReply reply;
|
if (reply.ParseFromString(msg.body())) {
|
rrcb(reply.data());
|
}
|
} else {
|
// ignored, or dropped
|
}
|
};
|
|
return Start(AsyncRecvProc, nworker);
|
}
|
|
bool SocketRequest::AsyncRequest(const Topic &topic, const void *data, const size_t size, const int timeout_ms)
|
{
|
try {
|
BHAddress addr;
|
if (QueryRPCTopic(topic, addr, timeout_ms)) {
|
const BHMsg &msg(MakeRequest(mq().Id(), topic, data, size));
|
return AsyncSend(addr.mq_id().data(), &msg, timeout_ms);
|
} else {
|
return false;
|
}
|
} catch (...) {
|
return false;
|
}
|
}
|
bool SocketRequest::AsyncRequest(const Topic &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() == kMsgTypeRequestTopicReply) {
|
MsgRequestTopicReply 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 Topic &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() == kMsgTypeRequestTopicReply) {
|
MsgRequestTopicReply dr;
|
if (dr.ParseFromString(reply.body())) {
|
dr.mutable_data()->swap(out);
|
return true;
|
} else {
|
printf("error parse reply.\n");
|
}
|
} else {
|
printf("error recv data. line: %d\n", __LINE__);
|
}
|
} else {
|
printf("error recv data. line: %d\n", __LINE__);
|
}
|
} catch (...) {
|
printf("error recv data. line: %d\n", __LINE__);
|
}
|
return false;
|
}
|
|
bool SocketRequest::AsyncSend(const void *remote, const void *pmsg, const int timeout_ms)
|
{
|
assert(remote && pmsg);
|
try {
|
const BHMsg &msg = *static_cast<const BHMsg *>(pmsg);
|
return mq().Send(*static_cast<const MQId *>(remote), msg, timeout_ms);
|
} catch (...) {
|
return false;
|
}
|
}
|
bool SocketRequest::AsyncSend(const void *remote, const void *pmsg, const int timeout_ms, const RecvBHMsgCB &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 Topic &topic, bhome::msg::BHAddress &addr, const int timeout_ms)
|
{
|
if (topic_cache_.Find(topic, addr)) {
|
return true;
|
}
|
|
BHMsg result;
|
const BHMsg &msg = MakeQueryTopic(mq().Id(), topic);
|
if (SyncSendAndRecv(&kBHTopicReqRepCenter, &msg, &result, timeout_ms)) {
|
if (result.type() == kMsgTypeQueryTopicReply) {
|
MsgQueryTopicReply reply;
|
if (reply.ParseFromString(result.body())) {
|
addr = reply.address();
|
if (addr.mq_id().empty()) {
|
return false;
|
} else {
|
topic_cache_.Update(topic, addr);
|
return true;
|
}
|
}
|
}
|
} else {
|
}
|
return false;
|
}
|
