mirror of
https://github.com/Xahau/xahaud.git
synced 2025-12-06 17:27:52 +00:00
120 lines
2.5 KiB
C++
120 lines
2.5 KiB
C++
#include <boost/thread.hpp>
|
|
|
|
#include "Application.h"
|
|
#include "RPCSub.h"
|
|
|
|
#include "CallRPC.h"
|
|
|
|
SETUP_LOG();
|
|
|
|
RPCSub::RPCSub(const std::string& strUrl, const std::string& strUsername, const std::string& strPassword)
|
|
: mUrl(strUrl), mSSL(false), mUsername(strUsername), mPassword(strPassword), mSending(false)
|
|
{
|
|
std::string strScheme;
|
|
|
|
if (!parseUrl(strUrl, strScheme, mIp, mPort, mPath))
|
|
{
|
|
throw std::runtime_error("Failed to parse url.");
|
|
}
|
|
else if (strScheme == "https")
|
|
{
|
|
mSSL = true;
|
|
}
|
|
else if (strScheme != "http")
|
|
{
|
|
throw std::runtime_error("Only http and https is supported.");
|
|
}
|
|
|
|
mSeq = 1;
|
|
|
|
if (mPort < 0)
|
|
mPort = mSSL ? 443 : 80;
|
|
|
|
cLog(lsINFO) << boost::str(boost::format("callRPC sub: ip='%s' port=%d ssl=%d path='%s'")
|
|
% mIp
|
|
% mPort
|
|
% mSSL
|
|
% mPath);
|
|
}
|
|
|
|
// XXX Could probably create a bunch of send jobs in a single get of the lock.
|
|
void RPCSub::sendThread()
|
|
{
|
|
Json::Value jvEvent;
|
|
bool bSend;
|
|
|
|
do
|
|
{
|
|
{
|
|
// Obtain the lock to manipulate the queue and change sending.
|
|
boost::mutex::scoped_lock sl(mLockInfo);
|
|
|
|
if (mDeque.empty())
|
|
{
|
|
mSending = false;
|
|
bSend = false;
|
|
}
|
|
else
|
|
{
|
|
std::pair<int, Json::Value> pEvent = mDeque.front();
|
|
|
|
mDeque.pop_front();
|
|
|
|
jvEvent = pEvent.second;
|
|
jvEvent["seq"] = pEvent.first;
|
|
|
|
bSend = true;
|
|
}
|
|
}
|
|
|
|
// Send outside of the lock.
|
|
if (bSend)
|
|
{
|
|
// XXX Might not need this in a try.
|
|
try
|
|
{
|
|
cLog(lsINFO) << boost::str(boost::format("callRPC calling: %s") % mIp);
|
|
|
|
callRPC(
|
|
theApp->getIOService(),
|
|
mIp, mPort,
|
|
mUsername, mPassword,
|
|
mPath, "event",
|
|
jvEvent,
|
|
mSSL);
|
|
}
|
|
catch (const std::exception& e)
|
|
{
|
|
cLog(lsINFO) << boost::str(boost::format("callRPC exception: %s") % e.what());
|
|
}
|
|
}
|
|
} while (bSend);
|
|
}
|
|
|
|
void RPCSub::send(const Json::Value& jvObj, bool broadcast)
|
|
{
|
|
boost::mutex::scoped_lock sl(mLockInfo);
|
|
|
|
if (RPC_EVENT_QUEUE_MAX == mDeque.size())
|
|
{
|
|
// Drop the previous event.
|
|
cLog(lsWARNING) << boost::str(boost::format("callRPC drop"));
|
|
mDeque.pop_back();
|
|
}
|
|
|
|
cLog(broadcast ? lsDEBUG : lsINFO) << boost::str(boost::format("callRPC push: %s") % jvObj);
|
|
|
|
mDeque.push_back(std::make_pair(mSeq++, jvObj));
|
|
|
|
if (!mSending)
|
|
{
|
|
// Start a sending thread.
|
|
mSending = true;
|
|
|
|
cLog(lsINFO) << boost::str(boost::format("callRPC start"));
|
|
boost::thread(boost::bind(&RPCSub::sendThread, this)).detach();
|
|
}
|
|
}
|
|
|
|
// vim:ts=4
|