mirror of
https://github.com/logos-co/logos-cpp-sdk.git
synced 2026-08-31 17:51:07 +00:00
298 lines
10 KiB
C++
298 lines
10 KiB
C++
#include "plain_transport_host.h"
|
|
|
|
#include "cbor_codec.h"
|
|
#include "io_context_pool.h"
|
|
#include "json_codec.h"
|
|
#include "qvariant_rpc_value.h"
|
|
|
|
#include "../../module_proxy.h"
|
|
|
|
#include <QDebug>
|
|
#include <QMetaObject>
|
|
|
|
#include <boost/asio/ssl/context.hpp>
|
|
|
|
namespace logos::plain {
|
|
|
|
namespace {
|
|
|
|
std::shared_ptr<IWireCodec> makeCodec(LogosWireCodec kind)
|
|
{
|
|
switch (kind) {
|
|
case LogosWireCodec::Cbor: return std::make_shared<CborCodec>();
|
|
case LogosWireCodec::Json:
|
|
default: return std::make_shared<JsonCodec>();
|
|
}
|
|
}
|
|
|
|
boost::asio::ssl::context buildSslCtx(const LogosTransportConfig& cfg, bool server)
|
|
{
|
|
boost::asio::ssl::context ctx(server
|
|
? boost::asio::ssl::context::tls_server
|
|
: boost::asio::ssl::context::tls_client);
|
|
ctx.set_options(boost::asio::ssl::context::default_workarounds
|
|
| boost::asio::ssl::context::no_sslv2
|
|
| boost::asio::ssl::context::no_sslv3
|
|
| boost::asio::ssl::context::single_dh_use);
|
|
if (!cfg.certFile.empty())
|
|
ctx.use_certificate_chain_file(cfg.certFile);
|
|
if (!cfg.keyFile.empty())
|
|
ctx.use_private_key_file(cfg.keyFile, boost::asio::ssl::context::pem);
|
|
if (!cfg.caFile.empty())
|
|
ctx.load_verify_file(cfg.caFile);
|
|
if (cfg.verifyPeer && !server) {
|
|
ctx.set_verify_mode(boost::asio::ssl::verify_peer);
|
|
} else if (!server) {
|
|
ctx.set_verify_mode(boost::asio::ssl::verify_none);
|
|
}
|
|
return ctx;
|
|
}
|
|
|
|
} // anonymous namespace
|
|
|
|
PlainTransportHost::PlainTransportHost(LogosTransportConfig cfg)
|
|
: m_cfg(std::move(cfg))
|
|
{
|
|
}
|
|
|
|
PlainTransportHost::~PlainTransportHost()
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
for (auto& [name, pub] : m_published) {
|
|
QObject::disconnect(pub.eventConn);
|
|
}
|
|
if (m_tcp) m_tcp->stop();
|
|
if (m_ssl) m_ssl->stop();
|
|
}
|
|
|
|
bool PlainTransportHost::start()
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
if (m_started) return true;
|
|
auto codec = makeCodec(m_cfg.codec);
|
|
auto& ioc = IoContextPool::shared().ioContext();
|
|
|
|
if (m_cfg.protocol == LogosProtocol::Tcp) {
|
|
m_tcp = std::make_shared<RpcServerTcp>(ioc, m_cfg.host, m_cfg.port, codec, this);
|
|
if (!m_tcp->start()) {
|
|
qCritical() << "PlainTransportHost: TCP bind failed on"
|
|
<< QString::fromStdString(m_cfg.host) << m_cfg.port;
|
|
m_tcp.reset();
|
|
return false;
|
|
}
|
|
m_boundPort = m_tcp->boundPort();
|
|
} else if (m_cfg.protocol == LogosProtocol::TcpSsl) {
|
|
try {
|
|
auto ctx = buildSslCtx(m_cfg, /*server=*/true);
|
|
m_ssl = std::make_shared<RpcServerSsl>(ioc, m_cfg.host, m_cfg.port,
|
|
std::move(ctx), codec, this);
|
|
if (!m_ssl->start()) {
|
|
qCritical() << "PlainTransportHost: TLS bind failed";
|
|
m_ssl.reset();
|
|
return false;
|
|
}
|
|
m_boundPort = m_ssl->boundPort();
|
|
} catch (const std::exception& e) {
|
|
qCritical() << "PlainTransportHost: SSL context setup failed:" << e.what();
|
|
return false;
|
|
}
|
|
} else {
|
|
qCritical() << "PlainTransportHost: unsupported protocol";
|
|
return false;
|
|
}
|
|
m_started = true;
|
|
return true;
|
|
}
|
|
|
|
QString PlainTransportHost::endpoint() const
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
if (m_boundPort == 0) return QString();
|
|
return QString("tcp://%1:%2")
|
|
.arg(QString::fromStdString(m_cfg.host))
|
|
.arg(m_boundPort);
|
|
}
|
|
|
|
QString PlainTransportHost::bindUrl(const QString& /*instanceId*/,
|
|
const QString& /*moduleName*/)
|
|
{
|
|
// One PlainTransportHost listens on a single host:port and serves every
|
|
// published module over the same socket; URL is independent of module.
|
|
return endpoint();
|
|
}
|
|
|
|
bool PlainTransportHost::publishObject(const QString& name, QObject* object)
|
|
{
|
|
if (!object) return false;
|
|
auto* proxy = qobject_cast<ModuleProxy*>(object);
|
|
if (!proxy) {
|
|
qWarning() << "PlainTransportHost::publishObject: expected ModuleProxy for"
|
|
<< name << "(plain transport only publishes ModuleProxy for now)";
|
|
return false;
|
|
}
|
|
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
Published pub;
|
|
pub.object = object;
|
|
const std::string stdName = name.toStdString();
|
|
|
|
// Hook the QObject's eventResponse(QString, QVariantList) signal so every
|
|
// Q_INVOKABLE-style event emission fans out to subscribed connections.
|
|
pub.eventConn = QObject::connect(proxy, &ModuleProxy::eventResponse,
|
|
[this, stdName](const QString& eventName, const QVariantList& data) {
|
|
EventMessage msg;
|
|
msg.object = stdName;
|
|
msg.eventName = eventName.toStdString();
|
|
msg.data = qvariantListToRpcList(data);
|
|
fanOutEvent(stdName, std::move(msg));
|
|
});
|
|
|
|
m_published[stdName] = std::move(pub);
|
|
return true;
|
|
}
|
|
|
|
void PlainTransportHost::unpublishObject(const QString& name)
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
auto it = m_published.find(name.toStdString());
|
|
if (it == m_published.end()) return;
|
|
QObject::disconnect(it->second.eventConn);
|
|
m_published.erase(it);
|
|
}
|
|
|
|
void PlainTransportHost::fanOutEvent(const std::string& name, EventMessage msg)
|
|
{
|
|
std::vector<EventSink> sinks;
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
auto it = m_published.find(name);
|
|
if (it == m_published.end()) return;
|
|
// Named subscribers + wildcard ("") subscribers get the event.
|
|
for (auto which : {msg.eventName, std::string{}}) {
|
|
auto evtIt = it->second.sinksByEvent.find(which);
|
|
if (evtIt == it->second.sinksByEvent.end()) continue;
|
|
for (auto& [key, sink] : evtIt->second) sinks.push_back(sink);
|
|
}
|
|
}
|
|
for (auto& sink : sinks) {
|
|
try { sink(msg); } catch (...) {}
|
|
}
|
|
}
|
|
|
|
void PlainTransportHost::onCall(const CallMessage& req, CallReply reply)
|
|
{
|
|
QObject* obj = nullptr;
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
auto it = m_published.find(req.object);
|
|
if (it != m_published.end()) obj = it->second.object;
|
|
}
|
|
if (!obj) {
|
|
ResultMessage res; res.id = req.id; res.ok = false;
|
|
res.err = "object not published: " + req.object;
|
|
res.errCode = "MODULE_NOT_LOADED";
|
|
reply(std::move(res));
|
|
return;
|
|
}
|
|
|
|
QString authToken = QString::fromStdString(req.authToken);
|
|
QString methodName = QString::fromStdString(req.method);
|
|
QVariantList args = rpcListToQVariantList(req.args);
|
|
uint64_t id = req.id;
|
|
|
|
QMetaObject::invokeMethod(obj, [obj, authToken, methodName, args, id, reply]() {
|
|
QVariant ret;
|
|
bool ok = QMetaObject::invokeMethod(obj, "callRemoteMethod",
|
|
Qt::DirectConnection,
|
|
Q_RETURN_ARG(QVariant, ret),
|
|
Q_ARG(QString, authToken),
|
|
Q_ARG(QString, methodName),
|
|
Q_ARG(QVariantList, args));
|
|
ResultMessage res;
|
|
res.id = id;
|
|
if (ok) {
|
|
res.ok = true;
|
|
res.value = qvariantToRpcValue(ret);
|
|
} else {
|
|
res.ok = false;
|
|
res.err = "callRemoteMethod failed";
|
|
res.errCode = "METHOD_FAILED";
|
|
}
|
|
reply(std::move(res));
|
|
}, Qt::QueuedConnection);
|
|
}
|
|
|
|
void PlainTransportHost::onMethods(const MethodsMessage& req, MethodsReply reply)
|
|
{
|
|
QObject* obj = nullptr;
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
auto it = m_published.find(req.object);
|
|
if (it != m_published.end()) obj = it->second.object;
|
|
}
|
|
if (!obj) {
|
|
MethodsResultMessage res; res.id = req.id; res.ok = false;
|
|
res.err = "object not published";
|
|
reply(std::move(res));
|
|
return;
|
|
}
|
|
uint64_t id = req.id;
|
|
QMetaObject::invokeMethod(obj, [obj, id, reply]() {
|
|
QJsonArray arr;
|
|
QMetaObject::invokeMethod(obj, "getPluginMethods",
|
|
Qt::DirectConnection,
|
|
Q_RETURN_ARG(QJsonArray, arr));
|
|
MethodsResultMessage res;
|
|
res.id = id;
|
|
res.ok = true;
|
|
res.methods = methodsFromJsonArray(arr);
|
|
reply(std::move(res));
|
|
}, Qt::QueuedConnection);
|
|
}
|
|
|
|
void PlainTransportHost::onSubscribe(const SubscribeMessage& req, EventSink sink)
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
auto it = m_published.find(req.object);
|
|
if (it == m_published.end()) return;
|
|
// Sinks are keyed by functor-shared-pointer address. The caller (RpcConnection)
|
|
// moves the sink in; we key by its move-captured lambda-object's address,
|
|
// which is unique per subscription.
|
|
static std::atomic<uint64_t> counter{0};
|
|
it->second.sinksByEvent[req.eventName][
|
|
reinterpret_cast<const void*>(counter.fetch_add(1) + 1)] = std::move(sink);
|
|
}
|
|
|
|
void PlainTransportHost::onUnsubscribe(const UnsubscribeMessage& req)
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
auto it = m_published.find(req.object);
|
|
if (it == m_published.end()) return;
|
|
it->second.sinksByEvent.erase(req.eventName);
|
|
}
|
|
|
|
void PlainTransportHost::onToken(const TokenMessage& req)
|
|
{
|
|
QObject* obj = nullptr;
|
|
{
|
|
std::lock_guard<std::mutex> g(m_mu);
|
|
// Route token to the module matching req.moduleName if we host
|
|
// it; otherwise the first published module (matches today's behavior
|
|
// for the single-published-object provider pattern).
|
|
auto it = m_published.find(req.moduleName);
|
|
if (it != m_published.end()) obj = it->second.object;
|
|
else if (!m_published.empty()) obj = m_published.begin()->second.object;
|
|
}
|
|
if (!obj) return;
|
|
QString authToken = QString::fromStdString(req.authToken);
|
|
QString moduleName = QString::fromStdString(req.moduleName);
|
|
QString token = QString::fromStdString(req.token);
|
|
QMetaObject::invokeMethod(obj, "informModuleToken",
|
|
Qt::QueuedConnection,
|
|
Q_ARG(QString, authToken),
|
|
Q_ARG(QString, moduleName),
|
|
Q_ARG(QString, token));
|
|
}
|
|
|
|
} // namespace logos::plain
|