mirror of
https://github.com/logos-co/logos-cpp-sdk.git
synced 2026-08-31 09:41:06 +00:00
support non-local remote transports (#57)
* support non-local remote transports * fix LogosResult * allow getting client over specific transport * fix ssl * investiage ssl error * pr comments * allow transport set configuration on any module * pr comments * add docs * pr comments * propagate only non-qt dependencies * restore ABI compatibility
This commit is contained in:
Dario Lipicar
@@ -0,0 +1,27 @@
|
||||
#include "cbor_codec.h"
|
||||
#include "json_mapping.h"
|
||||
|
||||
#include <nlohmann/json.hpp>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
using json = nlohmann::json;
|
||||
|
||||
std::vector<uint8_t> CborCodec::encode(const AnyMessage& msg)
|
||||
{
|
||||
const json j = messageToJson(msg);
|
||||
return json::to_cbor(j);
|
||||
}
|
||||
|
||||
AnyMessage CborCodec::decode(MessageType tag, const uint8_t* data, std::size_t len)
|
||||
{
|
||||
json j;
|
||||
try {
|
||||
j = json::from_cbor(data, data + len, /*strict=*/true, /*allow_exceptions=*/true);
|
||||
} catch (const std::exception& e) {
|
||||
throw CodecError(std::string("cbor parse failed: ") + e.what());
|
||||
}
|
||||
return jsonToMessage(tag, j);
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,28 @@
|
||||
#ifndef LOGOS_PLAIN_CBOR_CODEC_H
|
||||
#define LOGOS_PLAIN_CBOR_CODEC_H
|
||||
|
||||
#include "wire_codec.h"
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// CborCodec — same logical message layout as JsonCodec (shared via
|
||||
// json_mapping.{h,cpp}), serialized with nlohmann::json::to_cbor /
|
||||
// from_cbor. Matches JSON wire-for-wire in logical content; wire bytes
|
||||
// are binary CBOR rather than UTF-8 JSON text.
|
||||
//
|
||||
// Useful when you want smaller/faster on-the-wire encoding without
|
||||
// swapping to a wholly different codec family. Paired transports on the
|
||||
// daemon can offer both JSON and CBOR; clients pick per-connection via
|
||||
// --client-codec.
|
||||
class CborCodec : public IWireCodec {
|
||||
public:
|
||||
std::vector<uint8_t> encode(const AnyMessage&) override;
|
||||
AnyMessage decode(MessageType tag,
|
||||
const uint8_t* data,
|
||||
std::size_t len) override;
|
||||
std::string name() const override { return "cbor"; }
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_CBOR_CODEC_H
|
||||
@@ -0,0 +1,60 @@
|
||||
#ifndef LOGOS_PLAIN_INCOMING_CALL_HANDLER_H
|
||||
#define LOGOS_PLAIN_INCOMING_CALL_HANDLER_H
|
||||
|
||||
#include "rpc_message.h"
|
||||
|
||||
#include <functional>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// IncomingCallHandler — provider-side dispatch hook.
|
||||
//
|
||||
// rpc_connection hands inbound Call / Methods / Subscribe / Unsubscribe /
|
||||
// Token messages to a handler that the Qt-boundary layer implements. The
|
||||
// handler is what talks to the published QObject (ModuleProxy); this
|
||||
// interface deliberately speaks only plain C++ types so the wire stack
|
||||
// stays Qt-free.
|
||||
//
|
||||
// The reply callbacks can be invoked synchronously (from inside the
|
||||
// handler) or asynchronously from a different thread — rpc_connection
|
||||
// serializes the actual frame write internally.
|
||||
// -----------------------------------------------------------------------------
|
||||
class IncomingCallHandler {
|
||||
public:
|
||||
virtual ~IncomingCallHandler() = default;
|
||||
|
||||
using CallReply = std::function<void(ResultMessage)>;
|
||||
using MethodsReply = std::function<void(MethodsResultMessage)>;
|
||||
using EventSink = std::function<void(EventMessage)>;
|
||||
|
||||
virtual void onCall(const CallMessage& req, CallReply reply) = 0;
|
||||
|
||||
virtual void onMethods(const MethodsMessage& req, MethodsReply reply) = 0;
|
||||
|
||||
// `sink` stays alive until onUnsubscribe fires or the connection
|
||||
// dies. The handler must call `sink(evt)` on every matching emission.
|
||||
//
|
||||
// `connectionId` is an opaque per-connection token (the rpc layer
|
||||
// passes the connection's `this` pointer). The handler keys sinks
|
||||
// by it so a subsequent onUnsubscribe / onConnectionClosed can
|
||||
// remove only the sinks belonging to that connection — sub/unsub
|
||||
// frames don't carry a subscriber identifier on the wire.
|
||||
virtual void onSubscribe(const SubscribeMessage& req, EventSink sink,
|
||||
const void* connectionId) = 0;
|
||||
|
||||
virtual void onUnsubscribe(const UnsubscribeMessage& req,
|
||||
const void* connectionId) = 0;
|
||||
|
||||
// Called when a connection is torn down (graceful close or error)
|
||||
// so the handler can drop any sinks still keyed to it. Without
|
||||
// this, a dropped client leaks subscriptions and the host keeps
|
||||
// fanning events into dead sinks.
|
||||
virtual void onConnectionClosed(const void* connectionId) = 0;
|
||||
|
||||
virtual void onToken(const TokenMessage& req) = 0;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_INCOMING_CALL_HANDLER_H
|
||||
@@ -0,0 +1,28 @@
|
||||
#include "io_context_pool.h"
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
IoContextPool::IoContextPool()
|
||||
: m_ioc()
|
||||
, m_guard(boost::asio::make_work_guard(m_ioc))
|
||||
, m_worker([this]{ m_ioc.run(); })
|
||||
{
|
||||
}
|
||||
|
||||
IoContextPool::~IoContextPool()
|
||||
{
|
||||
// Drop the work guard so run() can return once all outstanding work
|
||||
// completes, then stop forcefully if something lingers.
|
||||
m_guard.reset();
|
||||
m_ioc.stop();
|
||||
if (m_worker.joinable())
|
||||
m_worker.join();
|
||||
}
|
||||
|
||||
IoContextPool& IoContextPool::shared()
|
||||
{
|
||||
static IoContextPool pool;
|
||||
return pool;
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,44 @@
|
||||
#ifndef LOGOS_PLAIN_IO_CONTEXT_POOL_H
|
||||
#define LOGOS_PLAIN_IO_CONTEXT_POOL_H
|
||||
|
||||
#include <boost/asio/executor_work_guard.hpp>
|
||||
#include <boost/asio/io_context.hpp>
|
||||
|
||||
#include <memory>
|
||||
#include <thread>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// IoContextPool — owns a single boost::asio::io_context and a worker thread
|
||||
// that runs it until the pool is destroyed.
|
||||
//
|
||||
// One pool per SDK process is sufficient for our traffic (a handful of
|
||||
// concurrent connections). If we ever need more parallelism, swap for a
|
||||
// multi-thread pool (one io_context per thread + round-robin dispatch).
|
||||
//
|
||||
// Access the shared pool via `sharedPool()`; tests / special cases can
|
||||
// construct their own.
|
||||
// -----------------------------------------------------------------------------
|
||||
class IoContextPool {
|
||||
public:
|
||||
IoContextPool();
|
||||
~IoContextPool();
|
||||
|
||||
IoContextPool(const IoContextPool&) = delete;
|
||||
IoContextPool& operator=(const IoContextPool&) = delete;
|
||||
|
||||
boost::asio::io_context& ioContext() { return m_ioc; }
|
||||
|
||||
// Process-wide default pool. Thread-safe lazy init.
|
||||
static IoContextPool& shared();
|
||||
|
||||
private:
|
||||
boost::asio::io_context m_ioc;
|
||||
boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_guard;
|
||||
std::thread m_worker;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_IO_CONTEXT_POOL_H
|
||||
@@ -0,0 +1,28 @@
|
||||
#include "json_codec.h"
|
||||
#include "json_mapping.h"
|
||||
|
||||
#include <nlohmann/json.hpp>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
using json = nlohmann::json;
|
||||
|
||||
std::vector<uint8_t> JsonCodec::encode(const AnyMessage& msg)
|
||||
{
|
||||
const json j = messageToJson(msg);
|
||||
const std::string s = j.dump();
|
||||
return std::vector<uint8_t>(s.begin(), s.end());
|
||||
}
|
||||
|
||||
AnyMessage JsonCodec::decode(MessageType tag, const uint8_t* data, std::size_t len)
|
||||
{
|
||||
json j;
|
||||
try {
|
||||
j = json::parse(data, data + len);
|
||||
} catch (const std::exception& e) {
|
||||
throw CodecError(std::string("json parse failed: ") + e.what());
|
||||
}
|
||||
return jsonToMessage(tag, j);
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,22 @@
|
||||
#ifndef LOGOS_PLAIN_JSON_CODEC_H
|
||||
#define LOGOS_PLAIN_JSON_CODEC_H
|
||||
|
||||
#include "wire_codec.h"
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// JsonCodec — uses nlohmann::json::dump / parse for the payload bytes.
|
||||
// Default codec for now; a future CborCodec will swap in by using
|
||||
// json::to_cbor / from_cbor on the same message structs.
|
||||
class JsonCodec : public IWireCodec {
|
||||
public:
|
||||
std::vector<uint8_t> encode(const AnyMessage&) override;
|
||||
AnyMessage decode(MessageType tag,
|
||||
const uint8_t* data,
|
||||
std::size_t len) override;
|
||||
std::string name() const override { return "json"; }
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_JSON_CODEC_H
|
||||
@@ -0,0 +1,329 @@
|
||||
#include "json_mapping.h"
|
||||
|
||||
#include <type_traits>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
using json = nlohmann::json;
|
||||
|
||||
// ── RpcValue ↔ json ─────────────────────────────────────────────────────────
|
||||
//
|
||||
// Mapping:
|
||||
// null ↔ json null
|
||||
// bool ↔ json boolean
|
||||
// int64 ↔ json integer
|
||||
// double ↔ json number (non-integer)
|
||||
// string ↔ json string
|
||||
// bytes ↔ {"_bytes": base64url}
|
||||
// list ↔ json array
|
||||
// map ↔ json object (we disambiguate bytes via the "_bytes" key)
|
||||
//
|
||||
// JSON has no bytes primitive, so `bytes` round-trip via a tagged object.
|
||||
// CBOR has native byte strings; when we want a "real" CBOR byte
|
||||
// representation we can upgrade CborCodec to bypass this hack and use
|
||||
// `json::binary_t` — for now identical behaviour keeps the code paths
|
||||
// uniform and tested.
|
||||
|
||||
namespace {
|
||||
|
||||
std::string b64url_encode(const std::vector<uint8_t>& bytes)
|
||||
{
|
||||
static const char* alpha =
|
||||
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
|
||||
std::string out;
|
||||
out.reserve(((bytes.size() + 2) / 3) * 4);
|
||||
size_t i = 0;
|
||||
while (i + 3 <= bytes.size()) {
|
||||
uint32_t n = (uint32_t(bytes[i]) << 16) | (uint32_t(bytes[i+1]) << 8) | uint32_t(bytes[i+2]);
|
||||
out.push_back(alpha[(n >> 18) & 0x3f]);
|
||||
out.push_back(alpha[(n >> 12) & 0x3f]);
|
||||
out.push_back(alpha[(n >> 6) & 0x3f]);
|
||||
out.push_back(alpha[ n & 0x3f]);
|
||||
i += 3;
|
||||
}
|
||||
if (i < bytes.size()) {
|
||||
uint32_t n = uint32_t(bytes[i]) << 16;
|
||||
if (i + 1 < bytes.size()) n |= uint32_t(bytes[i+1]) << 8;
|
||||
out.push_back(alpha[(n >> 18) & 0x3f]);
|
||||
out.push_back(alpha[(n >> 12) & 0x3f]);
|
||||
if (i + 1 < bytes.size())
|
||||
out.push_back(alpha[(n >> 6) & 0x3f]);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
std::vector<uint8_t> b64url_decode(const std::string& s)
|
||||
{
|
||||
auto idx = [](char c) -> int {
|
||||
if (c >= 'A' && c <= 'Z') return c - 'A';
|
||||
if (c >= 'a' && c <= 'z') return c - 'a' + 26;
|
||||
if (c >= '0' && c <= '9') return c - '0' + 52;
|
||||
if (c == '-') return 62;
|
||||
if (c == '_') return 63;
|
||||
return -1;
|
||||
};
|
||||
std::vector<uint8_t> out;
|
||||
out.reserve((s.size() * 3) / 4);
|
||||
size_t i = 0;
|
||||
while (i + 4 <= s.size()) {
|
||||
int a = idx(s[i]), b = idx(s[i+1]), c = idx(s[i+2]), d = idx(s[i+3]);
|
||||
if (a < 0 || b < 0 || c < 0 || d < 0)
|
||||
throw CodecError("invalid base64url input");
|
||||
uint32_t n = (uint32_t(a) << 18) | (uint32_t(b) << 12) | (uint32_t(c) << 6) | uint32_t(d);
|
||||
out.push_back((n >> 16) & 0xff);
|
||||
out.push_back((n >> 8) & 0xff);
|
||||
out.push_back( n & 0xff);
|
||||
i += 4;
|
||||
}
|
||||
size_t rem = s.size() - i;
|
||||
if (rem == 2 || rem == 3) {
|
||||
int a = idx(s[i]), b = idx(s[i+1]);
|
||||
if (a < 0 || b < 0) throw CodecError("invalid base64url input");
|
||||
uint32_t n = (uint32_t(a) << 18) | (uint32_t(b) << 12);
|
||||
out.push_back((n >> 16) & 0xff);
|
||||
if (rem == 3) {
|
||||
int c = idx(s[i+2]);
|
||||
if (c < 0) throw CodecError("invalid base64url input");
|
||||
n |= uint32_t(c) << 6;
|
||||
out.push_back((n >> 8) & 0xff);
|
||||
}
|
||||
} else if (rem != 0) {
|
||||
throw CodecError("invalid base64url length");
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
json valueToJson(const RpcValue& v);
|
||||
RpcValue jsonToValue(const json& j);
|
||||
|
||||
json valueToJson(const RpcValue& v)
|
||||
{
|
||||
if (v.isNull()) return nullptr;
|
||||
if (v.isBool()) return v.asBool();
|
||||
if (v.isInt()) return v.asInt();
|
||||
if (v.isDouble()) return v.asDouble();
|
||||
if (v.isString()) return v.asString();
|
||||
if (v.isBytes()) return json{{"_bytes", b64url_encode(v.asBytes().data)}};
|
||||
if (v.isList()) {
|
||||
json arr = json::array();
|
||||
for (const auto& item : v.asList().items) arr.push_back(valueToJson(item));
|
||||
return arr;
|
||||
}
|
||||
if (v.isMap()) {
|
||||
json obj = json::object();
|
||||
for (const auto& [k, val] : v.asMap().entries) obj[k] = valueToJson(val);
|
||||
return obj;
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
RpcValue jsonToValue(const json& j)
|
||||
{
|
||||
if (j.is_null()) return RpcValue{std::monostate{}};
|
||||
if (j.is_boolean()) return RpcValue{j.get<bool>()};
|
||||
if (j.is_number_integer() || j.is_number_unsigned())
|
||||
return RpcValue{j.get<int64_t>()};
|
||||
if (j.is_number_float()) return RpcValue{j.get<double>()};
|
||||
if (j.is_string()) return RpcValue{j.get<std::string>()};
|
||||
if (j.is_array()) {
|
||||
RpcList list;
|
||||
list.items.reserve(j.size());
|
||||
for (const auto& e : j) list.items.push_back(jsonToValue(e));
|
||||
return RpcValue{std::move(list)};
|
||||
}
|
||||
if (j.is_object()) {
|
||||
// Disambiguate bytes.
|
||||
if (j.size() == 1 && j.contains("_bytes") && j["_bytes"].is_string()) {
|
||||
return RpcValue{RpcBytes{b64url_decode(j["_bytes"].get<std::string>())}};
|
||||
}
|
||||
RpcMap map;
|
||||
for (auto it = j.begin(); it != j.end(); ++it)
|
||||
map.emplace(it.key(), jsonToValue(it.value()));
|
||||
return RpcValue{std::move(map)};
|
||||
}
|
||||
// CBOR round-trips through nlohmann::json::binary as binary_t. Convert
|
||||
// to our bytes representation so CborCodec and JsonCodec end up with
|
||||
// the same logical RpcValue shape.
|
||||
if (j.is_binary()) {
|
||||
const auto& b = j.get_binary();
|
||||
return RpcValue{RpcBytes{std::vector<uint8_t>(b.begin(), b.end())}};
|
||||
}
|
||||
return RpcValue{std::monostate{}};
|
||||
}
|
||||
|
||||
// ── Message struct ↔ json helpers ──────────────────────────────────────────
|
||||
|
||||
json methodToJson(const MethodMetadata& m)
|
||||
{
|
||||
json o = json::object();
|
||||
o["name"] = m.name;
|
||||
o["signature"] = m.signature;
|
||||
o["returnType"] = m.returnType;
|
||||
o["isInvokable"] = m.isInvokable;
|
||||
json pa = json::array();
|
||||
for (const auto& p : m.parameters.items) pa.push_back(valueToJson(p));
|
||||
o["parameters"] = std::move(pa);
|
||||
return o;
|
||||
}
|
||||
|
||||
MethodMetadata methodFromJson(const json& j)
|
||||
{
|
||||
MethodMetadata m;
|
||||
m.name = j.value("name", std::string{});
|
||||
m.signature = j.value("signature", std::string{});
|
||||
m.returnType = j.value("returnType", std::string{});
|
||||
m.isInvokable = j.value("isInvokable", true);
|
||||
if (j.contains("parameters") && j["parameters"].is_array()) {
|
||||
for (const auto& p : j["parameters"]) m.parameters.items.push_back(jsonToValue(p));
|
||||
}
|
||||
return m;
|
||||
}
|
||||
|
||||
json argsToJson(const std::vector<RpcValue>& args)
|
||||
{
|
||||
json a = json::array();
|
||||
for (const auto& v : args) a.push_back(valueToJson(v));
|
||||
return a;
|
||||
}
|
||||
|
||||
std::vector<RpcValue> argsFromJson(const json& j)
|
||||
{
|
||||
std::vector<RpcValue> out;
|
||||
if (j.is_array()) {
|
||||
out.reserve(j.size());
|
||||
for (const auto& e : j) out.push_back(jsonToValue(e));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
// ── Public entry points ────────────────────────────────────────────────────
|
||||
|
||||
json messageToJson(const AnyMessage& msg)
|
||||
{
|
||||
return std::visit([](const auto& m) -> json {
|
||||
using T = std::decay_t<decltype(m)>;
|
||||
json o = json::object();
|
||||
if constexpr (std::is_same_v<T, CallMessage>) {
|
||||
o["id"] = m.id;
|
||||
o["authToken"] = m.authToken;
|
||||
o["object"] = m.object;
|
||||
o["method"] = m.method;
|
||||
o["args"] = argsToJson(m.args);
|
||||
} else if constexpr (std::is_same_v<T, ResultMessage>) {
|
||||
o["id"] = m.id;
|
||||
o["ok"] = m.ok;
|
||||
if (m.ok) {
|
||||
o["value"] = valueToJson(m.value);
|
||||
} else {
|
||||
o["err"] = m.err;
|
||||
o["errCode"] = m.errCode;
|
||||
}
|
||||
} else if constexpr (std::is_same_v<T, SubscribeMessage>) {
|
||||
o["object"] = m.object;
|
||||
o["event"] = m.eventName;
|
||||
} else if constexpr (std::is_same_v<T, UnsubscribeMessage>) {
|
||||
o["object"] = m.object;
|
||||
o["event"] = m.eventName;
|
||||
} else if constexpr (std::is_same_v<T, EventMessage>) {
|
||||
o["object"] = m.object;
|
||||
o["event"] = m.eventName;
|
||||
o["data"] = argsToJson(m.data);
|
||||
} else if constexpr (std::is_same_v<T, TokenMessage>) {
|
||||
o["authToken"] = m.authToken;
|
||||
o["moduleName"] = m.moduleName;
|
||||
o["token"] = m.token;
|
||||
} else if constexpr (std::is_same_v<T, MethodsMessage>) {
|
||||
o["id"] = m.id;
|
||||
o["authToken"] = m.authToken;
|
||||
o["object"] = m.object;
|
||||
} else if constexpr (std::is_same_v<T, MethodsResultMessage>) {
|
||||
o["id"] = m.id;
|
||||
o["ok"] = m.ok;
|
||||
if (m.ok) {
|
||||
json ma = json::array();
|
||||
for (const auto& md : m.methods) ma.push_back(methodToJson(md));
|
||||
o["methods"] = std::move(ma);
|
||||
} else {
|
||||
o["err"] = m.err;
|
||||
}
|
||||
}
|
||||
return o;
|
||||
}, msg);
|
||||
}
|
||||
|
||||
AnyMessage jsonToMessage(MessageType tag, const json& j)
|
||||
{
|
||||
if (!j.is_object()) throw CodecError("expected top-level object");
|
||||
switch (tag) {
|
||||
case MessageType::Call: {
|
||||
CallMessage m;
|
||||
m.id = j.value("id", uint64_t{0});
|
||||
m.authToken = j.value("authToken", std::string{});
|
||||
m.object = j.value("object", std::string{});
|
||||
m.method = j.value("method", std::string{});
|
||||
if (j.contains("args")) m.args = argsFromJson(j["args"]);
|
||||
return m;
|
||||
}
|
||||
case MessageType::Result: {
|
||||
ResultMessage m;
|
||||
m.id = j.value("id", uint64_t{0});
|
||||
m.ok = j.value("ok", false);
|
||||
if (m.ok) {
|
||||
if (j.contains("value")) m.value = jsonToValue(j["value"]);
|
||||
} else {
|
||||
m.err = j.value("err", std::string{});
|
||||
m.errCode = j.value("errCode", std::string{});
|
||||
}
|
||||
return m;
|
||||
}
|
||||
case MessageType::Subscribe: {
|
||||
SubscribeMessage m;
|
||||
m.object = j.value("object", std::string{});
|
||||
m.eventName = j.value("event", std::string{});
|
||||
return m;
|
||||
}
|
||||
case MessageType::Unsubscribe: {
|
||||
UnsubscribeMessage m;
|
||||
m.object = j.value("object", std::string{});
|
||||
m.eventName = j.value("event", std::string{});
|
||||
return m;
|
||||
}
|
||||
case MessageType::Event: {
|
||||
EventMessage m;
|
||||
m.object = j.value("object", std::string{});
|
||||
m.eventName = j.value("event", std::string{});
|
||||
if (j.contains("data")) m.data = argsFromJson(j["data"]);
|
||||
return m;
|
||||
}
|
||||
case MessageType::Token: {
|
||||
TokenMessage m;
|
||||
m.authToken = j.value("authToken", std::string{});
|
||||
m.moduleName = j.value("moduleName", std::string{});
|
||||
m.token = j.value("token", std::string{});
|
||||
return m;
|
||||
}
|
||||
case MessageType::Methods: {
|
||||
MethodsMessage m;
|
||||
m.id = j.value("id", uint64_t{0});
|
||||
m.authToken = j.value("authToken", std::string{});
|
||||
m.object = j.value("object", std::string{});
|
||||
return m;
|
||||
}
|
||||
case MessageType::MethodsResult: {
|
||||
MethodsResultMessage m;
|
||||
m.id = j.value("id", uint64_t{0});
|
||||
m.ok = j.value("ok", false);
|
||||
if (m.ok && j.contains("methods") && j["methods"].is_array()) {
|
||||
for (const auto& md : j["methods"]) m.methods.push_back(methodFromJson(md));
|
||||
} else if (!m.ok) {
|
||||
m.err = j.value("err", std::string{});
|
||||
}
|
||||
return m;
|
||||
}
|
||||
}
|
||||
throw CodecError("unknown message tag");
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,22 @@
|
||||
#ifndef LOGOS_PLAIN_JSON_MAPPING_H
|
||||
#define LOGOS_PLAIN_JSON_MAPPING_H
|
||||
|
||||
// Shared RpcValue ↔ nlohmann::json conversion used by both JsonCodec
|
||||
// (dump / parse as text) and CborCodec (to_cbor / from_cbor as bytes).
|
||||
// The JSON representation is the canonical in-memory form; codecs differ
|
||||
// only in how they serialize that form to the wire.
|
||||
|
||||
#include "rpc_message.h"
|
||||
#include "rpc_value.h"
|
||||
#include "wire_codec.h"
|
||||
|
||||
#include <nlohmann/json.hpp>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
nlohmann::json messageToJson(const AnyMessage& msg);
|
||||
AnyMessage jsonToMessage(MessageType tag, const nlohmann::json& j);
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_JSON_MAPPING_H
|
||||
@@ -0,0 +1,219 @@
|
||||
#include "plain_logos_object.h"
|
||||
|
||||
#include "qvariant_rpc_value.h"
|
||||
|
||||
#include <QCoreApplication>
|
||||
#include <QDebug>
|
||||
#include <QMetaObject>
|
||||
#include <QTimer>
|
||||
|
||||
#include <chrono>
|
||||
#include <future>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
PlainLogosObject::PlainLogosObject(std::string objectName,
|
||||
std::shared_ptr<RpcConnectionBase> conn)
|
||||
: m_objectName(std::move(objectName))
|
||||
, m_conn(std::move(conn))
|
||||
{
|
||||
}
|
||||
|
||||
PlainLogosObject::~PlainLogosObject()
|
||||
{
|
||||
disconnectEvents();
|
||||
}
|
||||
|
||||
QVariant PlainLogosObject::callMethod(const QString& authToken,
|
||||
const QString& methodName,
|
||||
const QVariantList& args,
|
||||
int timeoutMs)
|
||||
{
|
||||
if (!m_conn || !m_conn->isOpen()) return QVariant();
|
||||
|
||||
CallMessage msg;
|
||||
msg.id = m_conn->nextId();
|
||||
msg.authToken = authToken.toStdString();
|
||||
msg.object = m_objectName;
|
||||
msg.method = methodName.toStdString();
|
||||
msg.args = qvariantListToRpcList(args);
|
||||
|
||||
auto fut = m_conn->sendCall(std::move(msg));
|
||||
|
||||
if (fut.wait_for(std::chrono::milliseconds(timeoutMs)) != std::future_status::ready) {
|
||||
qWarning() << "PlainLogosObject::callMethod: timeout for" << methodName;
|
||||
return QVariant();
|
||||
}
|
||||
auto res = fut.get();
|
||||
if (!res.ok) {
|
||||
qWarning() << "PlainLogosObject::callMethod:" << methodName
|
||||
<< "failed:" << QString::fromStdString(res.err);
|
||||
return QVariant();
|
||||
}
|
||||
return rpcValueToQVariant(res.value);
|
||||
}
|
||||
|
||||
namespace {
|
||||
|
||||
// Hand `callback(result)` over to the Qt event loop so PlainLogosObject's
|
||||
// async path matches LogosObject's interface contract: callbacks are
|
||||
// always delivered on a subsequent event-loop iteration, on the Qt
|
||||
// thread, never synchronously and never racing with QObjects/UI code.
|
||||
//
|
||||
// Using QCoreApplication::instance() as the anchor means the queued
|
||||
// invocation lands on whichever thread runs the Qt event loop in this
|
||||
// process, regardless of which worker thread completed the future.
|
||||
// If the application has shut down (instance() is null), we drop the
|
||||
// callback rather than invoke it from an arbitrary thread.
|
||||
void postToQtEventLoop(PlainLogosObject::AsyncResultCallback callback,
|
||||
QVariant result)
|
||||
{
|
||||
QCoreApplication* app = QCoreApplication::instance();
|
||||
if (!app) return;
|
||||
QMetaObject::invokeMethod(app,
|
||||
[callback = std::move(callback), result = std::move(result)]() mutable {
|
||||
callback(result);
|
||||
},
|
||||
Qt::QueuedConnection);
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
void PlainLogosObject::callMethodAsync(const QString& authToken,
|
||||
const QString& methodName,
|
||||
const QVariantList& args,
|
||||
int timeoutMs,
|
||||
AsyncResultCallback callback)
|
||||
{
|
||||
if (!callback) return;
|
||||
if (!m_conn || !m_conn->isOpen()) {
|
||||
// Defer even the failure path — LogosObject's contract requires
|
||||
// callbacks on a subsequent event-loop iteration, never inline.
|
||||
postToQtEventLoop(std::move(callback), QVariant());
|
||||
return;
|
||||
}
|
||||
|
||||
CallMessage msg;
|
||||
msg.id = m_conn->nextId();
|
||||
msg.authToken = authToken.toStdString();
|
||||
msg.object = m_objectName;
|
||||
msg.method = methodName.toStdString();
|
||||
msg.args = qvariantListToRpcList(args);
|
||||
|
||||
auto fut = std::make_shared<std::future<ResultMessage>>(
|
||||
m_conn->sendCall(std::move(msg)));
|
||||
|
||||
// Waiter thread is per-call but the callback hops back to the Qt
|
||||
// event loop before running, so it never races with Qt objects. A
|
||||
// future iteration can fold this wait into the shared Asio
|
||||
// io_context (the connection already runs on it) so we don't spin
|
||||
// up a thread per pending RPC.
|
||||
std::thread([fut, timeoutMs, callback = std::move(callback)]() mutable {
|
||||
if (fut->wait_for(std::chrono::milliseconds(timeoutMs))
|
||||
!= std::future_status::ready) {
|
||||
postToQtEventLoop(std::move(callback), QVariant());
|
||||
return;
|
||||
}
|
||||
auto res = fut->get();
|
||||
QVariant value = res.ok ? rpcValueToQVariant(res.value) : QVariant();
|
||||
postToQtEventLoop(std::move(callback), std::move(value));
|
||||
}).detach();
|
||||
}
|
||||
|
||||
bool PlainLogosObject::informModuleToken(const QString& authToken,
|
||||
const QString& moduleName,
|
||||
const QString& token,
|
||||
int /*timeoutMs*/)
|
||||
{
|
||||
if (!m_conn || !m_conn->isOpen()) return false;
|
||||
TokenMessage msg;
|
||||
msg.authToken = authToken.toStdString();
|
||||
msg.moduleName = moduleName.toStdString();
|
||||
msg.token = token.toStdString();
|
||||
m_conn->sendToken(std::move(msg));
|
||||
return true; // fire-and-forget
|
||||
}
|
||||
|
||||
void PlainLogosObject::onEvent(const QString& eventName, EventCallback callback)
|
||||
{
|
||||
if (!m_conn || !m_conn->isOpen() || !callback) return;
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_subs.emplace_back(eventName, callback);
|
||||
}
|
||||
|
||||
SubscribeMessage msg;
|
||||
msg.object = m_objectName;
|
||||
msg.eventName = eventName.toStdString();
|
||||
|
||||
// Bridge RPC event → Qt-flavored callback.
|
||||
m_conn->sendSubscribe(std::move(msg), [callback](EventMessage evt) {
|
||||
callback(QString::fromStdString(evt.eventName),
|
||||
rpcListToQVariantList(evt.data));
|
||||
});
|
||||
}
|
||||
|
||||
void PlainLogosObject::disconnectEvents()
|
||||
{
|
||||
std::vector<std::pair<QString, EventCallback>> subs;
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
subs.swap(m_subs);
|
||||
}
|
||||
if (!m_conn) return;
|
||||
for (const auto& [name, _] : subs) {
|
||||
UnsubscribeMessage msg;
|
||||
msg.object = m_objectName;
|
||||
msg.eventName = name.toStdString();
|
||||
m_conn->sendUnsubscribe(std::move(msg));
|
||||
}
|
||||
}
|
||||
|
||||
void PlainLogosObject::emitEvent(const QString& eventName, const QVariantList& data)
|
||||
{
|
||||
if (!m_conn || !m_conn->isOpen()) return;
|
||||
EventMessage msg;
|
||||
msg.object = m_objectName;
|
||||
msg.eventName = eventName.toStdString();
|
||||
msg.data = qvariantListToRpcList(data);
|
||||
m_conn->sendEvent(std::move(msg));
|
||||
}
|
||||
|
||||
QJsonArray PlainLogosObject::getMethods()
|
||||
{
|
||||
if (!m_conn || !m_conn->isOpen()) return QJsonArray();
|
||||
|
||||
MethodsMessage msg;
|
||||
msg.id = m_conn->nextId();
|
||||
msg.object = m_objectName;
|
||||
|
||||
auto fut = m_conn->sendMethods(std::move(msg));
|
||||
if (fut.wait_for(std::chrono::seconds(5)) != std::future_status::ready) {
|
||||
return QJsonArray();
|
||||
}
|
||||
auto res = fut.get();
|
||||
if (!res.ok) return QJsonArray();
|
||||
return methodsToJsonArray(res.methods);
|
||||
}
|
||||
|
||||
void PlainLogosObject::release()
|
||||
{
|
||||
// The RpcConnection is SHARED across every PlainLogosObject a single
|
||||
// PlainTransportConnection hands out. Stopping it here would kill
|
||||
// the connection for every other holder too, so just unsubscribe our
|
||||
// own events and drop our reference — the connection stays alive
|
||||
// until PlainTransportConnection itself is destroyed.
|
||||
disconnectEvents();
|
||||
m_conn.reset();
|
||||
delete this;
|
||||
}
|
||||
|
||||
quintptr PlainLogosObject::id() const
|
||||
{
|
||||
return reinterpret_cast<quintptr>(m_conn.get());
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,62 @@
|
||||
#ifndef LOGOS_PLAIN_LOGOS_OBJECT_H
|
||||
#define LOGOS_PLAIN_LOGOS_OBJECT_H
|
||||
|
||||
#include "logos_object.h"
|
||||
|
||||
#include "rpc_connection.h"
|
||||
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// PlainLogosObject — consumer-side LogosObject backed by the plain-C++
|
||||
// RPC runtime. Identical public shape to LocalLogosObject / RemoteLogosObject
|
||||
// so LogosAPIConsumer doesn't care which backend it's talking to.
|
||||
//
|
||||
// Owns a shared_ptr<RpcConnectionBase>; the transport layer hands the
|
||||
// connection over after opening the socket. release() stops the connection.
|
||||
// -----------------------------------------------------------------------------
|
||||
class PlainLogosObject : public LogosObject {
|
||||
public:
|
||||
PlainLogosObject(std::string objectName,
|
||||
std::shared_ptr<RpcConnectionBase> conn);
|
||||
~PlainLogosObject() override;
|
||||
|
||||
QVariant callMethod(const QString& authToken,
|
||||
const QString& methodName,
|
||||
const QVariantList& args,
|
||||
int timeoutMs) override;
|
||||
|
||||
void callMethodAsync(const QString& authToken,
|
||||
const QString& methodName,
|
||||
const QVariantList& args,
|
||||
int timeoutMs,
|
||||
AsyncResultCallback callback) override;
|
||||
|
||||
bool informModuleToken(const QString& authToken,
|
||||
const QString& moduleName,
|
||||
const QString& token,
|
||||
int timeoutMs) override;
|
||||
|
||||
void onEvent(const QString& eventName, EventCallback callback) override;
|
||||
void disconnectEvents() override;
|
||||
void emitEvent(const QString& eventName, const QVariantList& data) override;
|
||||
QJsonArray getMethods() override;
|
||||
void release() override;
|
||||
quintptr id() const override;
|
||||
|
||||
private:
|
||||
std::string m_objectName;
|
||||
std::shared_ptr<RpcConnectionBase> m_conn;
|
||||
std::mutex m_mu;
|
||||
std::vector<std::pair<QString, EventCallback>> m_subs;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_LOGOS_OBJECT_H
|
||||
@@ -0,0 +1,148 @@
|
||||
#include "plain_transport_connection.h"
|
||||
|
||||
#include "cbor_codec.h"
|
||||
#include "io_context_pool.h"
|
||||
#include "json_codec.h"
|
||||
#include "plain_logos_object.h"
|
||||
#include "rpc_server.h"
|
||||
|
||||
#include <QDebug>
|
||||
|
||||
#include <boost/asio/connect.hpp>
|
||||
#include <boost/asio/ip/tcp.hpp>
|
||||
#include <boost/asio/ssl/context.hpp>
|
||||
#include <boost/asio/ssl/host_name_verification.hpp>
|
||||
#include <boost/asio/ssl/stream.hpp>
|
||||
|
||||
#include <openssl/ssl.h>
|
||||
|
||||
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 buildClientSslCtx(const LogosTransportConfig& cfg)
|
||||
{
|
||||
boost::asio::ssl::context ctx(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);
|
||||
if (!cfg.caFile.empty())
|
||||
ctx.load_verify_file(cfg.caFile);
|
||||
ctx.set_verify_mode(cfg.verifyPeer
|
||||
? boost::asio::ssl::verify_peer
|
||||
: boost::asio::ssl::verify_none);
|
||||
return ctx;
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
PlainTransportConnection::PlainTransportConnection(LogosTransportConfig cfg)
|
||||
: m_cfg(std::move(cfg))
|
||||
{
|
||||
}
|
||||
|
||||
PlainTransportConnection::~PlainTransportConnection()
|
||||
{
|
||||
if (m_conn) m_conn->stop("connection destroyed");
|
||||
}
|
||||
|
||||
bool PlainTransportConnection::connectToHost()
|
||||
{
|
||||
if (m_connected) return true;
|
||||
|
||||
auto& ioc = IoContextPool::shared().ioContext();
|
||||
auto codec = makeCodec(m_cfg.codec);
|
||||
|
||||
try {
|
||||
boost::asio::ip::tcp::resolver resolver(ioc);
|
||||
auto endpoints = resolver.resolve(m_cfg.host, std::to_string(m_cfg.port));
|
||||
|
||||
if (m_cfg.protocol == LogosProtocol::Tcp) {
|
||||
boost::asio::ip::tcp::socket socket(ioc);
|
||||
boost::asio::connect(socket, endpoints);
|
||||
auto conn = std::make_shared<TcpConnection>(
|
||||
std::move(socket), codec, nullptr);
|
||||
conn->start();
|
||||
m_conn = conn;
|
||||
m_connected = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
if (m_cfg.protocol == LogosProtocol::TcpSsl) {
|
||||
auto ctx = buildClientSslCtx(m_cfg);
|
||||
SslStream stream(ioc, ctx);
|
||||
// Set SNI: TLS clients must advertise the target host name
|
||||
// in the ClientHello so the server picks the right cert
|
||||
// (and so any intermediate proxy can route correctly). Without
|
||||
// this, vhost-style deployments would terminate the handshake.
|
||||
// Cast through the OpenSSL macro because Asio doesn't expose
|
||||
// SNI configuration at the wrapper level.
|
||||
if (!SSL_set_tlsext_host_name(stream.native_handle(),
|
||||
m_cfg.host.c_str())) {
|
||||
qWarning() << "PlainTransportConnection: SSL_set_tlsext_host_name failed";
|
||||
}
|
||||
// Verify the peer's certificate name matches the host we
|
||||
// dialed when verifyPeer is on. verify_peer alone only
|
||||
// validates the chain — without host-name verification a
|
||||
// valid cert for a *different* name would still pass, which
|
||||
// is exactly the MITM hole verify_peer is meant to close.
|
||||
if (m_cfg.verifyPeer) {
|
||||
stream.set_verify_callback(
|
||||
boost::asio::ssl::host_name_verification(m_cfg.host));
|
||||
}
|
||||
boost::asio::connect(stream.lowest_layer(), endpoints);
|
||||
stream.handshake(boost::asio::ssl::stream_base::client);
|
||||
auto conn = std::make_shared<SslConnection>(
|
||||
std::move(stream), codec, nullptr);
|
||||
conn->start();
|
||||
m_conn = conn;
|
||||
m_connected = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
qCritical() << "PlainTransportConnection: unsupported protocol";
|
||||
return false;
|
||||
} catch (const std::exception& e) {
|
||||
qWarning() << "PlainTransportConnection::connectToHost failed:" << e.what();
|
||||
m_connected = false;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
bool PlainTransportConnection::isConnected() const
|
||||
{
|
||||
return m_connected && m_conn && m_conn->isOpen();
|
||||
}
|
||||
|
||||
bool PlainTransportConnection::reconnect()
|
||||
{
|
||||
if (m_conn) m_conn->stop("reconnecting");
|
||||
m_conn.reset();
|
||||
m_connected = false;
|
||||
return connectToHost();
|
||||
}
|
||||
|
||||
LogosObject* PlainTransportConnection::requestObject(const QString& objectName, int /*timeoutMs*/)
|
||||
{
|
||||
if (!isConnected()) return nullptr;
|
||||
return new PlainLogosObject(objectName.toStdString(), m_conn);
|
||||
}
|
||||
|
||||
QString PlainTransportConnection::endpointUrl(const QString& /*instanceId*/,
|
||||
const QString& /*moduleName*/)
|
||||
{
|
||||
return QString("tcp://%1:%2")
|
||||
.arg(QString::fromStdString(m_cfg.host))
|
||||
.arg(m_cfg.port);
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,41 @@
|
||||
#ifndef LOGOS_PLAIN_TRANSPORT_CONNECTION_H
|
||||
#define LOGOS_PLAIN_TRANSPORT_CONNECTION_H
|
||||
|
||||
#include "logos_transport.h"
|
||||
#include "logos_transport_config.h"
|
||||
|
||||
#include "rpc_connection.h"
|
||||
|
||||
#include <memory>
|
||||
#include <string>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// PlainTransportConnection — consumer-side LogosTransportConnection.
|
||||
//
|
||||
// connectToHost() opens a TCP (or TLS) socket to the daemon's endpoint from
|
||||
// the LogosTransportConfig and starts the RPC read loop. requestObject()
|
||||
// returns a PlainLogosObject sharing that connection.
|
||||
// -----------------------------------------------------------------------------
|
||||
class PlainTransportConnection : public LogosTransportConnection {
|
||||
public:
|
||||
explicit PlainTransportConnection(LogosTransportConfig cfg);
|
||||
~PlainTransportConnection() override;
|
||||
|
||||
bool connectToHost() override;
|
||||
bool isConnected() const override;
|
||||
bool reconnect() override;
|
||||
LogosObject* requestObject(const QString& objectName, int timeoutMs) override;
|
||||
QString endpointUrl(const QString& instanceId,
|
||||
const QString& moduleName) override;
|
||||
|
||||
private:
|
||||
LogosTransportConfig m_cfg;
|
||||
std::shared_ptr<RpcConnectionBase> m_conn;
|
||||
bool m_connected = false;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_TRANSPORT_CONNECTION_H
|
||||
@@ -0,0 +1,468 @@
|
||||
#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 <atomic>
|
||||
|
||||
#include <boost/asio/ssl/context.hpp>
|
||||
#include <boost/version.hpp>
|
||||
#include <openssl/ssl.h>
|
||||
#include <openssl/err.h>
|
||||
#include <openssl/opensslv.h>
|
||||
#include <openssl/x509.h>
|
||||
#include <openssl/evp.h>
|
||||
|
||||
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>();
|
||||
}
|
||||
}
|
||||
|
||||
// Print the last OpenSSL error to qWarning, clearing the error stack.
|
||||
// Used after any SSL_CTX_set_* call that returned 0 — silent failures
|
||||
// were how we missed the cipher-list misconfig for multiple rounds.
|
||||
void dumpSslErrors(const char* where)
|
||||
{
|
||||
unsigned long e;
|
||||
while ((e = ERR_get_error()) != 0) {
|
||||
char buf[256];
|
||||
ERR_error_string_n(e, buf, sizeof(buf));
|
||||
qWarning() << "buildSslCtx/" << where << ":" << buf;
|
||||
}
|
||||
}
|
||||
|
||||
boost::asio::ssl::context buildSslCtx(const LogosTransportConfig& cfg, bool server)
|
||||
{
|
||||
// One-time diagnostic: print Boost + OpenSSL build-vs-runtime
|
||||
// versions on first call. We've spent multiple rounds on
|
||||
// TLS configuration that *appeared* to take effect (strings
|
||||
// baked into the binary) but didn't change runtime behaviour;
|
||||
// a version mismatch between compile-time headers and runtime
|
||||
// libs would do that, and this prints the smoking gun.
|
||||
static bool versionsLogged = false;
|
||||
if (!versionsLogged) {
|
||||
versionsLogged = true;
|
||||
qInfo().nospace()
|
||||
<< "buildSslCtx versions: "
|
||||
<< "Boost build=" << BOOST_VERSION
|
||||
<< " (" << BOOST_LIB_VERSION << "), "
|
||||
<< "OpenSSL build=0x" << Qt::hex << OPENSSL_VERSION_NUMBER
|
||||
<< Qt::dec << " (" << OPENSSL_VERSION_TEXT << "), "
|
||||
<< "runtime=" << OpenSSL_version(OPENSSL_VERSION);
|
||||
}
|
||||
|
||||
qInfo().nospace() << "buildSslCtx: cfg.certFile='"
|
||||
<< QString::fromStdString(cfg.certFile)
|
||||
<< "' cfg.keyFile='"
|
||||
<< QString::fromStdString(cfg.keyFile)
|
||||
<< "' cfg.caFile='"
|
||||
<< QString::fromStdString(cfg.caFile)
|
||||
<< "' role=" << (server ? "server" : "client");
|
||||
|
||||
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);
|
||||
|
||||
// Require TLS 1.2+. Check return values on every set_* below and
|
||||
// dump any pending OpenSSL errors — previous silent-fail behaviour
|
||||
// was the root cause of multiple debugging rounds landing no fix.
|
||||
if (!SSL_CTX_set_min_proto_version(ctx.native_handle(), TLS1_2_VERSION)) {
|
||||
qWarning() << "buildSslCtx: SSL_CTX_set_min_proto_version(TLS1_2) failed";
|
||||
dumpSslErrors("set_min_proto_version");
|
||||
}
|
||||
if (!SSL_CTX_set_max_proto_version(ctx.native_handle(), TLS1_3_VERSION)) {
|
||||
qWarning() << "buildSslCtx: SSL_CTX_set_max_proto_version(TLS1_3) failed";
|
||||
dumpSslErrors("set_max_proto_version");
|
||||
}
|
||||
if (!SSL_CTX_set1_groups_list(ctx.native_handle(),
|
||||
"X25519:P-256:P-384:P-521")) {
|
||||
qWarning() << "buildSslCtx: SSL_CTX_set1_groups_list failed";
|
||||
dumpSslErrors("set1_groups_list");
|
||||
}
|
||||
if (!SSL_CTX_set_ciphersuites(ctx.native_handle(),
|
||||
"TLS_AES_128_GCM_SHA256:"
|
||||
"TLS_AES_256_GCM_SHA384:"
|
||||
"TLS_CHACHA20_POLY1305_SHA256")) {
|
||||
qWarning() << "buildSslCtx: SSL_CTX_set_ciphersuites failed";
|
||||
dumpSslErrors("set_ciphersuites");
|
||||
}
|
||||
if (!SSL_CTX_set_cipher_list(ctx.native_handle(),
|
||||
"ECDHE+AESGCM:ECDHE+CHACHA20:"
|
||||
"DHE+AESGCM:DHE+CHACHA20:"
|
||||
"!aNULL:!MD5:!DSS:!RC4:!3DES")) {
|
||||
qWarning() << "buildSslCtx: SSL_CTX_set_cipher_list failed";
|
||||
dumpSslErrors("set_cipher_list");
|
||||
}
|
||||
|
||||
// Log what stuck. If min/max_proto read back as 0, the platform
|
||||
// doesn't support bounded proto versions (very old OpenSSL) and
|
||||
// nothing we did above will have capped anything. Cipher count is
|
||||
// the second-most-likely silent failure: a non-zero count from
|
||||
// SSL_CTX_get_ciphers means the TLS 1.2 cipher list got applied.
|
||||
{
|
||||
auto* sk = SSL_CTX_get_ciphers(ctx.native_handle());
|
||||
const int n = sk ? sk_SSL_CIPHER_num(sk) : 0;
|
||||
QString first;
|
||||
if (n > 0) {
|
||||
const SSL_CIPHER* c = sk_SSL_CIPHER_value(sk, 0);
|
||||
first = QString::fromLatin1(SSL_CIPHER_get_name(c));
|
||||
}
|
||||
qInfo().nospace() << "buildSslCtx: role="
|
||||
<< (server ? "server" : "client")
|
||||
<< " min_proto=0x" << Qt::hex
|
||||
<< SSL_CTX_get_min_proto_version(ctx.native_handle())
|
||||
<< " max_proto=0x"
|
||||
<< SSL_CTX_get_max_proto_version(ctx.native_handle())
|
||||
<< " options=0x"
|
||||
<< SSL_CTX_get_options(ctx.native_handle())
|
||||
<< Qt::dec
|
||||
<< " cipher_count=" << n
|
||||
<< " first_cipher=" << first;
|
||||
}
|
||||
|
||||
if (!cfg.certFile.empty()) {
|
||||
ctx.use_certificate_chain_file(cfg.certFile);
|
||||
dumpSslErrors("use_certificate_chain_file");
|
||||
}
|
||||
if (!cfg.keyFile.empty()) {
|
||||
ctx.use_private_key_file(cfg.keyFile, boost::asio::ssl::context::pem);
|
||||
dumpSslErrors("use_private_key_file");
|
||||
}
|
||||
if (!cfg.caFile.empty()) {
|
||||
ctx.load_verify_file(cfg.caFile);
|
||||
dumpSslErrors("load_verify_file");
|
||||
}
|
||||
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);
|
||||
}
|
||||
|
||||
// Final check: is a cert + matching key actually attached to the
|
||||
// SSL_CTX? "no shared cipher" / "unsupported protocol" can both
|
||||
// result from a server that has no usable cert at all (no PKI
|
||||
// cipher suites can negotiate without one). use_certificate_*
|
||||
// / use_private_key_* throw on outright failure but can leave the
|
||||
// ctx in a "loaded but the CTX-level slot is empty" state if the
|
||||
// file's first PEM block was something other than a CERTIFICATE.
|
||||
{
|
||||
X509* serverCert = SSL_CTX_get0_certificate(ctx.native_handle());
|
||||
EVP_PKEY* serverKey = SSL_CTX_get0_privatekey(ctx.native_handle());
|
||||
const int checkOk = SSL_CTX_check_private_key(ctx.native_handle());
|
||||
qInfo().nospace() << "buildSslCtx: cert_attached="
|
||||
<< (serverCert ? "yes" : "no")
|
||||
<< " key_attached=" << (serverKey ? "yes" : "no")
|
||||
<< " check_private_key=" << checkOk;
|
||||
if (!checkOk) dumpSslErrors("SSL_CTX_check_private_key");
|
||||
}
|
||||
return ctx;
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
PlainTransportHost::PlainTransportHost(LogosTransportConfig cfg)
|
||||
: m_cfg(std::move(cfg))
|
||||
{
|
||||
}
|
||||
|
||||
PlainTransportHost::~PlainTransportHost()
|
||||
{
|
||||
// The two stop() calls below are blocking: they tear down all
|
||||
// open RpcConnection sessions, each of which calls
|
||||
// IncomingCallHandler::onConnectionClosed() — which re-acquires
|
||||
// `m_mu` to remove its publisher mapping. Holding m_mu across
|
||||
// stop() therefore self-deadlocks.
|
||||
//
|
||||
// Move the published map and the listeners out under the lock,
|
||||
// then drop it before driving the shutdowns. Once we've moved
|
||||
// them, no other thread can reach this object's data through
|
||||
// m_published / m_tcp / m_ssl.
|
||||
decltype(m_published) published;
|
||||
decltype(m_tcp) tcp;
|
||||
decltype(m_ssl) ssl;
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
published = std::move(m_published);
|
||||
tcp = std::move(m_tcp);
|
||||
ssl = std::move(m_ssl);
|
||||
}
|
||||
for (auto& [name, pub] : published) {
|
||||
QObject::disconnect(pub.eventConn);
|
||||
}
|
||||
if (tcp) tcp->stop();
|
||||
if (ssl) 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,
|
||||
const void* connectionId)
|
||||
{
|
||||
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 the originating connection so that
|
||||
// onUnsubscribe / onConnectionClosed can remove only sinks
|
||||
// belonging to that connection — sub/unsub frames don't carry a
|
||||
// subscriber id on the wire.
|
||||
it->second.sinksByEvent[req.eventName][connectionId] = std::move(sink);
|
||||
}
|
||||
|
||||
void PlainTransportHost::onUnsubscribe(const UnsubscribeMessage& req,
|
||||
const void* connectionId)
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
auto it = m_published.find(req.object);
|
||||
if (it == m_published.end()) return;
|
||||
auto evtIt = it->second.sinksByEvent.find(req.eventName);
|
||||
if (evtIt == it->second.sinksByEvent.end()) return;
|
||||
// Only drop the requesting connection's sink — other clients
|
||||
// subscribed to the same (object, event) keep theirs. Previously
|
||||
// this erased the entire eventName entry, taking every other
|
||||
// subscriber down with it.
|
||||
evtIt->second.erase(connectionId);
|
||||
if (evtIt->second.empty()) it->second.sinksByEvent.erase(evtIt);
|
||||
}
|
||||
|
||||
void PlainTransportHost::onConnectionClosed(const void* connectionId)
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
// Sweep every published object's per-event sink table and drop any
|
||||
// entries belonging to the closed connection. Without this, a
|
||||
// crashing/disconnecting client (which never sends Unsubscribe)
|
||||
// leaves dead sinks in the table forever.
|
||||
for (auto& [_name, pub] : m_published) {
|
||||
for (auto evtIt = pub.sinksByEvent.begin(); evtIt != pub.sinksByEvent.end(); ) {
|
||||
evtIt->second.erase(connectionId);
|
||||
if (evtIt->second.empty()) evtIt = pub.sinksByEvent.erase(evtIt);
|
||||
else ++evtIt;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
@@ -0,0 +1,83 @@
|
||||
#ifndef LOGOS_PLAIN_TRANSPORT_HOST_H
|
||||
#define LOGOS_PLAIN_TRANSPORT_HOST_H
|
||||
|
||||
#include "logos_transport.h"
|
||||
#include "logos_transport_config.h"
|
||||
|
||||
#include "incoming_call_handler.h"
|
||||
#include "rpc_server.h"
|
||||
|
||||
#include <QObject>
|
||||
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// PlainTransportHost — publishes QObjects over plain-C++ TCP or TCP+SSL.
|
||||
//
|
||||
// Owns an RpcServer (TCP or SSL variant), an IWireCodec (per config), and
|
||||
// a registry mapping object name → published QObject. For each object it
|
||||
// hooks into the QObject's `eventResponse(QString, QVariantList)` Qt signal
|
||||
// so emitted events fan out to every subscribed RPC connection.
|
||||
// -----------------------------------------------------------------------------
|
||||
class PlainTransportHost
|
||||
: public LogosTransportHost
|
||||
, public IncomingCallHandler
|
||||
{
|
||||
public:
|
||||
explicit PlainTransportHost(LogosTransportConfig cfg);
|
||||
~PlainTransportHost() override;
|
||||
|
||||
// LogosTransportHost
|
||||
bool publishObject(const QString& name, QObject* object) override;
|
||||
void unpublishObject(const QString& name) override;
|
||||
QString bindUrl(const QString& instanceId,
|
||||
const QString& moduleName) override;
|
||||
|
||||
// Reports the bound endpoint URL ("tcp://host:port") once start() has
|
||||
// succeeded. Empty string until then.
|
||||
QString endpoint() const;
|
||||
|
||||
// Must be called once after constructing + publishing is wired up,
|
||||
// so the acceptor starts listening. Idempotent.
|
||||
bool start();
|
||||
|
||||
// IncomingCallHandler
|
||||
void onCall(const CallMessage& req, CallReply reply) override;
|
||||
void onMethods(const MethodsMessage& req, MethodsReply reply) override;
|
||||
void onSubscribe(const SubscribeMessage& req, EventSink sink,
|
||||
const void* connectionId) override;
|
||||
void onUnsubscribe(const UnsubscribeMessage& req,
|
||||
const void* connectionId) override;
|
||||
void onConnectionClosed(const void* connectionId) override;
|
||||
void onToken(const TokenMessage& req) override;
|
||||
|
||||
// Internal: deliver an event emitted by the wrapped QObject to every
|
||||
// subscribed connection (both matching-name and wildcard subscribers).
|
||||
void fanOutEvent(const std::string& name, EventMessage msg);
|
||||
|
||||
private:
|
||||
struct Published {
|
||||
QObject* object = nullptr;
|
||||
// Tracked event subscribers per event name (including "" wildcard).
|
||||
std::map<std::string, std::map<const void*, EventSink>> sinksByEvent;
|
||||
QMetaObject::Connection eventConn;
|
||||
};
|
||||
|
||||
LogosTransportConfig m_cfg;
|
||||
std::shared_ptr<RpcServerTcp> m_tcp;
|
||||
std::shared_ptr<RpcServerSsl> m_ssl;
|
||||
uint16_t m_boundPort = 0;
|
||||
|
||||
mutable std::mutex m_mu;
|
||||
std::map<std::string, Published> m_published;
|
||||
bool m_started = false;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_TRANSPORT_HOST_H
|
||||
@@ -0,0 +1,255 @@
|
||||
#include "qvariant_rpc_value.h"
|
||||
|
||||
#include "../../logos_types.h"
|
||||
|
||||
#include <QMetaType>
|
||||
|
||||
#include <cmath>
|
||||
#include <limits>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
namespace {
|
||||
|
||||
RpcValue fromJsonValue(const QJsonValue& v);
|
||||
QJsonValue toJsonValue(const RpcValue& v);
|
||||
|
||||
RpcValue fromJsonValue(const QJsonValue& v)
|
||||
{
|
||||
switch (v.type()) {
|
||||
case QJsonValue::Null: return RpcValue{std::monostate{}};
|
||||
case QJsonValue::Bool: return RpcValue{v.toBool()};
|
||||
case QJsonValue::Double: {
|
||||
double d = v.toDouble();
|
||||
double intPart = 0.0;
|
||||
if (std::modf(d, &intPart) == 0.0 &&
|
||||
d >= double(std::numeric_limits<int64_t>::min()) &&
|
||||
d <= double(std::numeric_limits<int64_t>::max()))
|
||||
return RpcValue{int64_t(d)};
|
||||
return RpcValue{d};
|
||||
}
|
||||
case QJsonValue::String: return RpcValue{v.toString().toStdString()};
|
||||
case QJsonValue::Array: {
|
||||
RpcList out;
|
||||
const auto arr = v.toArray();
|
||||
out.items.reserve(arr.size());
|
||||
for (const QJsonValue& e : arr) out.items.push_back(fromJsonValue(e));
|
||||
return RpcValue{std::move(out)};
|
||||
}
|
||||
case QJsonValue::Object: {
|
||||
RpcMap out;
|
||||
const auto obj = v.toObject();
|
||||
for (auto it = obj.begin(); it != obj.end(); ++it)
|
||||
out.emplace(it.key().toStdString(), fromJsonValue(it.value()));
|
||||
return RpcValue{std::move(out)};
|
||||
}
|
||||
default:
|
||||
return RpcValue{std::monostate{}};
|
||||
}
|
||||
}
|
||||
|
||||
QJsonValue toJsonValue(const RpcValue& v)
|
||||
{
|
||||
if (v.isNull()) return QJsonValue(QJsonValue::Null);
|
||||
if (v.isBool()) return QJsonValue(v.asBool());
|
||||
if (v.isInt()) return QJsonValue(static_cast<double>(v.asInt()));
|
||||
if (v.isDouble()) return QJsonValue(v.asDouble());
|
||||
if (v.isString()) return QJsonValue(QString::fromStdString(v.asString()));
|
||||
if (v.isBytes()) {
|
||||
// QJsonValue has no bytes primitive; encode as base64 string.
|
||||
const auto& b = v.asBytes().data;
|
||||
QByteArray ba(reinterpret_cast<const char*>(b.data()),
|
||||
static_cast<int>(b.size()));
|
||||
return QJsonValue(QString::fromLatin1(ba.toBase64(QByteArray::Base64UrlEncoding)));
|
||||
}
|
||||
if (v.isList()) {
|
||||
QJsonArray arr;
|
||||
for (const auto& e : v.asList().items) arr.append(toJsonValue(e));
|
||||
return arr;
|
||||
}
|
||||
if (v.isMap()) {
|
||||
QJsonObject obj;
|
||||
for (const auto& kv : v.asMap().entries)
|
||||
obj.insert(QString::fromStdString(kv.first), toJsonValue(kv.second));
|
||||
return obj;
|
||||
}
|
||||
return QJsonValue(QJsonValue::Null);
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
RpcValue qvariantToRpcValue(const QVariant& v)
|
||||
{
|
||||
if (!v.isValid()) return RpcValue{std::monostate{}};
|
||||
|
||||
// LogosResult is a user-defined struct registered via qRegisterMetaType;
|
||||
// its metatype id is assigned at runtime so we can't put it in the
|
||||
// switch on QMetaType::Type below. Check it first — if we let it fall
|
||||
// through to the default, we'd stringify it via QVariant::toString()
|
||||
// (returning "" because LogosResult has no QString converter) or,
|
||||
// earlier, lose it as std::monostate{} and the receiver would see null.
|
||||
//
|
||||
// Wire shape: {"success": bool, "value": <any>, "error": <any>}.
|
||||
// That matches the struct's fields and recursively reuses the RpcValue
|
||||
// conversion for `value` and `error`, which themselves are QVariants
|
||||
// carrying primitives / QVariantMap / QVariantList / etc.
|
||||
//
|
||||
// Look up the metatype id per call (not cached in a `static`): the
|
||||
// first `qvariantToRpcValue` call might land before any `LogosAPI`
|
||||
// has called `qRegisterMetaType<LogosResult>`, and we don't want to
|
||||
// permanently cache `UnknownType` in that case. The lookup is a
|
||||
// hash probe — trivially cheap compared to the actual RPC work.
|
||||
{
|
||||
const int logosResultId = QMetaType::fromName("LogosResult").id();
|
||||
if (logosResultId != QMetaType::UnknownType && v.userType() == logosResultId) {
|
||||
const LogosResult r = v.value<LogosResult>();
|
||||
RpcMap m;
|
||||
m.emplace("success", RpcValue{r.success});
|
||||
m.emplace("value", qvariantToRpcValue(r.value));
|
||||
m.emplace("error", qvariantToRpcValue(r.error));
|
||||
return RpcValue{std::move(m)};
|
||||
}
|
||||
}
|
||||
|
||||
// Fast path for the common scalar types.
|
||||
switch (static_cast<QMetaType::Type>(v.userType())) {
|
||||
case QMetaType::Bool: return RpcValue{v.toBool()};
|
||||
case QMetaType::Int:
|
||||
case QMetaType::Long:
|
||||
case QMetaType::LongLong:
|
||||
case QMetaType::Short:
|
||||
case QMetaType::Char:
|
||||
case QMetaType::SChar:
|
||||
return RpcValue{int64_t(v.toLongLong())};
|
||||
case QMetaType::UInt:
|
||||
case QMetaType::ULong:
|
||||
case QMetaType::ULongLong:
|
||||
case QMetaType::UShort:
|
||||
case QMetaType::UChar:
|
||||
return RpcValue{int64_t(v.toULongLong())};
|
||||
case QMetaType::Float:
|
||||
case QMetaType::Double:
|
||||
return RpcValue{v.toDouble()};
|
||||
case QMetaType::QString:
|
||||
return RpcValue{v.toString().toStdString()};
|
||||
case QMetaType::QByteArray: {
|
||||
QByteArray ba = v.toByteArray();
|
||||
RpcBytes b;
|
||||
b.data.assign(reinterpret_cast<const uint8_t*>(ba.data()),
|
||||
reinterpret_cast<const uint8_t*>(ba.data()) + ba.size());
|
||||
return RpcValue{std::move(b)};
|
||||
}
|
||||
case QMetaType::QVariantList: {
|
||||
RpcList list;
|
||||
const QVariantList src = v.toList();
|
||||
list.items.reserve(src.size());
|
||||
for (const QVariant& e : src) list.items.push_back(qvariantToRpcValue(e));
|
||||
return RpcValue{std::move(list)};
|
||||
}
|
||||
case QMetaType::QVariantMap: {
|
||||
RpcMap map;
|
||||
const QVariantMap src = v.toMap();
|
||||
for (auto it = src.begin(); it != src.end(); ++it)
|
||||
map.emplace(it.key().toStdString(), qvariantToRpcValue(it.value()));
|
||||
return RpcValue{std::move(map)};
|
||||
}
|
||||
case QMetaType::QJsonValue:
|
||||
return fromJsonValue(v.toJsonValue());
|
||||
case QMetaType::QJsonArray: {
|
||||
RpcList list;
|
||||
const QJsonArray arr = v.toJsonArray();
|
||||
list.items.reserve(arr.size());
|
||||
for (const QJsonValue& e : arr) list.items.push_back(fromJsonValue(e));
|
||||
return RpcValue{std::move(list)};
|
||||
}
|
||||
case QMetaType::QJsonObject: {
|
||||
RpcMap map;
|
||||
const QJsonObject obj = v.toJsonObject();
|
||||
for (auto it = obj.begin(); it != obj.end(); ++it)
|
||||
map.emplace(it.key().toStdString(), fromJsonValue(it.value()));
|
||||
return RpcValue{std::move(map)};
|
||||
}
|
||||
default:
|
||||
// Best-effort fallback: stringify.
|
||||
if (v.canConvert<QString>()) return RpcValue{v.toString().toStdString()};
|
||||
return RpcValue{std::monostate{}};
|
||||
}
|
||||
}
|
||||
|
||||
QVariant rpcValueToQVariant(const RpcValue& v)
|
||||
{
|
||||
if (v.isNull()) return QVariant();
|
||||
if (v.isBool()) return QVariant(v.asBool());
|
||||
if (v.isInt()) return QVariant(static_cast<qlonglong>(v.asInt()));
|
||||
if (v.isDouble()) return QVariant(v.asDouble());
|
||||
if (v.isString()) return QVariant(QString::fromStdString(v.asString()));
|
||||
if (v.isBytes()) {
|
||||
const auto& b = v.asBytes().data;
|
||||
return QVariant(QByteArray(reinterpret_cast<const char*>(b.data()),
|
||||
static_cast<int>(b.size())));
|
||||
}
|
||||
if (v.isList()) return QVariant(rpcListToQVariantList(v.asList().items));
|
||||
if (v.isMap()) {
|
||||
QVariantMap map;
|
||||
for (const auto& kv : v.asMap().entries)
|
||||
map.insert(QString::fromStdString(kv.first), rpcValueToQVariant(kv.second));
|
||||
return QVariant(std::move(map));
|
||||
}
|
||||
return QVariant();
|
||||
}
|
||||
|
||||
std::vector<RpcValue> qvariantListToRpcList(const QVariantList& list)
|
||||
{
|
||||
std::vector<RpcValue> out;
|
||||
out.reserve(list.size());
|
||||
for (const QVariant& e : list) out.push_back(qvariantToRpcValue(e));
|
||||
return out;
|
||||
}
|
||||
|
||||
QVariantList rpcListToQVariantList(const std::vector<RpcValue>& list)
|
||||
{
|
||||
QVariantList out;
|
||||
out.reserve(list.size());
|
||||
for (const auto& e : list) out.append(rpcValueToQVariant(e));
|
||||
return out;
|
||||
}
|
||||
|
||||
QJsonArray methodsToJsonArray(const std::vector<MethodMetadata>& methods)
|
||||
{
|
||||
QJsonArray out;
|
||||
for (const auto& m : methods) {
|
||||
QJsonObject o;
|
||||
o["name"] = QString::fromStdString(m.name);
|
||||
o["signature"] = QString::fromStdString(m.signature);
|
||||
o["returnType"] = QString::fromStdString(m.returnType);
|
||||
o["isInvokable"] = m.isInvokable;
|
||||
QJsonArray params;
|
||||
for (const auto& p : m.parameters.items) params.append(toJsonValue(p));
|
||||
o["parameters"] = std::move(params);
|
||||
out.append(o);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
std::vector<MethodMetadata> methodsFromJsonArray(const QJsonArray& arr)
|
||||
{
|
||||
std::vector<MethodMetadata> out;
|
||||
out.reserve(arr.size());
|
||||
for (const QJsonValue& v : arr) {
|
||||
if (!v.isObject()) continue;
|
||||
const auto o = v.toObject();
|
||||
MethodMetadata m;
|
||||
m.name = o.value("name").toString().toStdString();
|
||||
m.signature = o.value("signature").toString().toStdString();
|
||||
m.returnType = o.value("returnType").toString().toStdString();
|
||||
m.isInvokable = o.value("isInvokable").toBool(true);
|
||||
if (o.contains("parameters") && o.value("parameters").isArray()) {
|
||||
for (const QJsonValue& p : o.value("parameters").toArray())
|
||||
m.parameters.items.push_back(fromJsonValue(p));
|
||||
}
|
||||
out.push_back(std::move(m));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,34 @@
|
||||
#ifndef LOGOS_PLAIN_QVARIANT_RPC_VALUE_H
|
||||
#define LOGOS_PLAIN_QVARIANT_RPC_VALUE_H
|
||||
|
||||
// Qt ↔ plain adapter. This is THE place the plain-C++ transport tier
|
||||
// touches Qt — isolated here so when Qt is eventually removed from the
|
||||
// SDK interface, there's one file to delete.
|
||||
|
||||
#include "rpc_message.h"
|
||||
#include "rpc_value.h"
|
||||
|
||||
#include <QByteArray>
|
||||
#include <QJsonArray>
|
||||
#include <QJsonObject>
|
||||
#include <QJsonValue>
|
||||
#include <QString>
|
||||
#include <QVariant>
|
||||
#include <QVariantList>
|
||||
#include <QVariantMap>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
RpcValue qvariantToRpcValue(const QVariant& v);
|
||||
QVariant rpcValueToQVariant(const RpcValue& v);
|
||||
|
||||
std::vector<RpcValue> qvariantListToRpcList(const QVariantList& list);
|
||||
QVariantList rpcListToQVariantList(const std::vector<RpcValue>& list);
|
||||
|
||||
// Method metadata round-trip (used for introspection).
|
||||
QJsonArray methodsToJsonArray(const std::vector<MethodMetadata>& methods);
|
||||
std::vector<MethodMetadata> methodsFromJsonArray(const QJsonArray& arr);
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_QVARIANT_RPC_VALUE_H
|
||||
@@ -0,0 +1,442 @@
|
||||
#ifndef LOGOS_PLAIN_RPC_CONNECTION_H
|
||||
#define LOGOS_PLAIN_RPC_CONNECTION_H
|
||||
|
||||
#include "incoming_call_handler.h"
|
||||
#include "rpc_framing.h"
|
||||
#include "rpc_message.h"
|
||||
#include "wire_codec.h"
|
||||
|
||||
#include <boost/asio/any_io_executor.hpp>
|
||||
#include <boost/asio/bind_executor.hpp>
|
||||
#include <boost/asio/buffer.hpp>
|
||||
#include <boost/asio/post.hpp>
|
||||
#include <boost/asio/read.hpp>
|
||||
#include <boost/asio/strand.hpp>
|
||||
#include <boost/asio/write.hpp>
|
||||
#include <boost/system/error_code.hpp>
|
||||
|
||||
#include <atomic>
|
||||
#include <cstdint>
|
||||
#include <deque>
|
||||
#include <functional>
|
||||
#include <future>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// RpcConnectionBase — type-erased public surface of RpcConnection<Stream>.
|
||||
//
|
||||
// Callers (plain_logos_object, plain_transport_host) hold a
|
||||
// shared_ptr<RpcConnectionBase> so they don't have to know whether the
|
||||
// underlying socket is plain TCP or TLS-wrapped TCP. All the async machinery
|
||||
// lives in the templated subclass.
|
||||
// -----------------------------------------------------------------------------
|
||||
class RpcConnectionBase {
|
||||
public:
|
||||
using ErrorHandler = std::function<void(const std::string& reason)>;
|
||||
|
||||
virtual ~RpcConnectionBase() = default;
|
||||
|
||||
virtual void start() = 0;
|
||||
virtual void stop(const std::string& reason = "stopped") = 0;
|
||||
virtual bool isOpen() const = 0;
|
||||
|
||||
virtual std::future<ResultMessage> sendCall(CallMessage msg) = 0;
|
||||
virtual std::future<MethodsResultMessage> sendMethods(MethodsMessage msg) = 0;
|
||||
|
||||
virtual void sendSubscribe(SubscribeMessage msg,
|
||||
std::function<void(EventMessage)> callback) = 0;
|
||||
virtual void sendUnsubscribe(UnsubscribeMessage msg) = 0;
|
||||
virtual void sendEvent(EventMessage msg) = 0;
|
||||
virtual void sendToken(TokenMessage msg) = 0;
|
||||
|
||||
virtual void setErrorHandler(ErrorHandler handler) = 0;
|
||||
virtual uint64_t nextId() = 0;
|
||||
};
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// RpcConnection<Stream> — one full-duplex RPC conversation over a Boost.Asio
|
||||
// stream-like socket (plain TCP or SSL-wrapped TCP, sharing this template).
|
||||
//
|
||||
// Roles: the same connection supports both directions. Either peer can
|
||||
// initiate Call / Methods / Subscribe / Token / Event messages. Provider-side
|
||||
// dispatch of inbound Call/Methods/Subscribe/Token goes through an
|
||||
// IncomingCallHandler supplied at construction (may be null for pure-consumer
|
||||
// connections).
|
||||
//
|
||||
// Lifecycle: heap-allocated via std::make_shared; call start() once the
|
||||
// socket is ready; call stop() (or destroy) to tear down.
|
||||
// -----------------------------------------------------------------------------
|
||||
template <typename Stream>
|
||||
class RpcConnection
|
||||
: public RpcConnectionBase
|
||||
, public std::enable_shared_from_this<RpcConnection<Stream>>
|
||||
{
|
||||
public:
|
||||
RpcConnection(Stream stream,
|
||||
std::shared_ptr<IWireCodec> codec,
|
||||
IncomingCallHandler* handler = nullptr);
|
||||
|
||||
void start() override;
|
||||
void stop(const std::string& reason = "stopped") override;
|
||||
bool isOpen() const override { return !m_stopped.load(); }
|
||||
|
||||
std::future<ResultMessage> sendCall(CallMessage msg) override;
|
||||
std::future<MethodsResultMessage> sendMethods(MethodsMessage msg) override;
|
||||
|
||||
void sendSubscribe(SubscribeMessage msg,
|
||||
std::function<void(EventMessage)> callback) override;
|
||||
void sendUnsubscribe(UnsubscribeMessage msg) override;
|
||||
void sendEvent(EventMessage msg) override;
|
||||
void sendToken(TokenMessage msg) override;
|
||||
|
||||
void setErrorHandler(ErrorHandler handler) override {
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_error = std::move(handler);
|
||||
}
|
||||
|
||||
uint64_t nextId() override {
|
||||
return m_nextId.fetch_add(1, std::memory_order_relaxed);
|
||||
}
|
||||
|
||||
private:
|
||||
void doRead();
|
||||
void handleFrame(MessageType tag, std::vector<uint8_t> payload);
|
||||
void dispatchIncoming(AnyMessage msg);
|
||||
void writeFrame(std::vector<uint8_t> frame);
|
||||
void doWrite();
|
||||
void fail(const std::string& reason);
|
||||
|
||||
Stream m_stream;
|
||||
std::shared_ptr<IWireCodec> m_codec;
|
||||
IncomingCallHandler* m_handler;
|
||||
boost::asio::strand<boost::asio::any_io_executor> m_strand;
|
||||
|
||||
// Read side
|
||||
FrameReader m_reader;
|
||||
std::vector<uint8_t> m_readBuf;
|
||||
|
||||
// Write side
|
||||
std::deque<std::vector<uint8_t>> m_writeQueue;
|
||||
bool m_writing = false;
|
||||
|
||||
// Outgoing-pending maps
|
||||
std::mutex m_mu;
|
||||
std::map<uint64_t, std::shared_ptr<std::promise<ResultMessage>>> m_pendingCalls;
|
||||
std::map<uint64_t, std::shared_ptr<std::promise<MethodsResultMessage>>> m_pendingMethods;
|
||||
|
||||
using EventKey = std::pair<std::string, std::string>; // object, event
|
||||
std::map<EventKey, std::function<void(EventMessage)>> m_eventCallbacks;
|
||||
|
||||
ErrorHandler m_error;
|
||||
std::atomic<uint64_t> m_nextId{1};
|
||||
std::atomic<bool> m_stopped{false};
|
||||
std::atomic<bool> m_started{false};
|
||||
};
|
||||
|
||||
// ── Template implementation (must be visible at instantiation sites) ─────
|
||||
|
||||
template <typename Stream>
|
||||
RpcConnection<Stream>::RpcConnection(Stream stream,
|
||||
std::shared_ptr<IWireCodec> codec,
|
||||
IncomingCallHandler* handler)
|
||||
: m_stream(std::move(stream))
|
||||
, m_codec(std::move(codec))
|
||||
, m_handler(handler)
|
||||
, m_strand(boost::asio::make_strand(m_stream.get_executor()))
|
||||
{
|
||||
m_readBuf.resize(4096);
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::start()
|
||||
{
|
||||
bool expected = false;
|
||||
if (!m_started.compare_exchange_strong(expected, true)) return;
|
||||
auto self = this->shared_from_this();
|
||||
boost::asio::post(m_strand, [self] { self->doRead(); });
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::stop(const std::string& reason)
|
||||
{
|
||||
fail(reason);
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::doRead()
|
||||
{
|
||||
auto self = this->shared_from_this();
|
||||
m_stream.async_read_some(boost::asio::buffer(m_readBuf),
|
||||
boost::asio::bind_executor(m_strand,
|
||||
[self](const boost::system::error_code& ec, std::size_t n) {
|
||||
if (ec) { self->fail(ec.message()); return; }
|
||||
try {
|
||||
self->m_reader.append(self->m_readBuf.data(), n);
|
||||
MessageType tag;
|
||||
std::vector<uint8_t> payload;
|
||||
while (self->m_reader.next(tag, payload)) {
|
||||
self->handleFrame(tag, std::move(payload));
|
||||
}
|
||||
} catch (const std::exception& e) {
|
||||
self->fail(std::string("frame error: ") + e.what());
|
||||
return;
|
||||
}
|
||||
self->doRead();
|
||||
}));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::handleFrame(MessageType tag, std::vector<uint8_t> payload)
|
||||
{
|
||||
AnyMessage msg;
|
||||
try {
|
||||
msg = m_codec->decode(tag, payload.data(), payload.size());
|
||||
} catch (const std::exception& e) {
|
||||
fail(std::string("decode error: ") + e.what());
|
||||
return;
|
||||
}
|
||||
dispatchIncoming(std::move(msg));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::dispatchIncoming(AnyMessage msg)
|
||||
{
|
||||
std::visit([this](auto&& m) {
|
||||
using T = std::decay_t<decltype(m)>;
|
||||
|
||||
if constexpr (std::is_same_v<T, ResultMessage>) {
|
||||
std::shared_ptr<std::promise<ResultMessage>> p;
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
auto it = m_pendingCalls.find(m.id);
|
||||
if (it != m_pendingCalls.end()) {
|
||||
p = std::move(it->second);
|
||||
m_pendingCalls.erase(it);
|
||||
}
|
||||
}
|
||||
if (p) p->set_value(std::forward<decltype(m)>(m));
|
||||
|
||||
} else if constexpr (std::is_same_v<T, MethodsResultMessage>) {
|
||||
std::shared_ptr<std::promise<MethodsResultMessage>> p;
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
auto it = m_pendingMethods.find(m.id);
|
||||
if (it != m_pendingMethods.end()) {
|
||||
p = std::move(it->second);
|
||||
m_pendingMethods.erase(it);
|
||||
}
|
||||
}
|
||||
if (p) p->set_value(std::forward<decltype(m)>(m));
|
||||
|
||||
} else if constexpr (std::is_same_v<T, EventMessage>) {
|
||||
std::function<void(EventMessage)> cb;
|
||||
std::function<void(EventMessage)> wildcardCb;
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
auto it = m_eventCallbacks.find({m.object, m.eventName});
|
||||
if (it != m_eventCallbacks.end()) cb = it->second;
|
||||
auto wit = m_eventCallbacks.find({m.object, std::string{}});
|
||||
if (wit != m_eventCallbacks.end()) wildcardCb = wit->second;
|
||||
}
|
||||
if (cb) cb(m);
|
||||
if (wildcardCb) wildcardCb(m);
|
||||
|
||||
} else if constexpr (std::is_same_v<T, CallMessage>) {
|
||||
if (!m_handler) return;
|
||||
auto self = this->shared_from_this();
|
||||
m_handler->onCall(m, [self](ResultMessage res) {
|
||||
self->writeFrame(encodeFrame(*self->m_codec, AnyMessage{std::move(res)}));
|
||||
});
|
||||
|
||||
} else if constexpr (std::is_same_v<T, MethodsMessage>) {
|
||||
if (!m_handler) return;
|
||||
auto self = this->shared_from_this();
|
||||
m_handler->onMethods(m, [self](MethodsResultMessage res) {
|
||||
self->writeFrame(encodeFrame(*self->m_codec, AnyMessage{std::move(res)}));
|
||||
});
|
||||
|
||||
} else if constexpr (std::is_same_v<T, SubscribeMessage>) {
|
||||
if (!m_handler) return;
|
||||
// weak_ptr capture so the host's stored sink doesn't keep the
|
||||
// connection alive past its natural lifetime — without this,
|
||||
// `[self]` would leak every subscribed connection until
|
||||
// unsubscribe (which a crashing client never sends).
|
||||
std::weak_ptr<RpcConnection<Stream>> weak = this->shared_from_this();
|
||||
const void* connId = static_cast<const void*>(this);
|
||||
m_handler->onSubscribe(m, [weak](EventMessage evt) {
|
||||
if (auto self = weak.lock()) self->sendEvent(std::move(evt));
|
||||
}, connId);
|
||||
|
||||
} else if constexpr (std::is_same_v<T, UnsubscribeMessage>) {
|
||||
if (m_handler)
|
||||
m_handler->onUnsubscribe(m, static_cast<const void*>(this));
|
||||
|
||||
} else if constexpr (std::is_same_v<T, TokenMessage>) {
|
||||
if (m_handler) m_handler->onToken(m);
|
||||
}
|
||||
}, std::move(msg));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
std::future<ResultMessage>
|
||||
RpcConnection<Stream>::sendCall(CallMessage msg)
|
||||
{
|
||||
auto p = std::make_shared<std::promise<ResultMessage>>();
|
||||
auto f = p->get_future();
|
||||
if (m_stopped.load()) {
|
||||
ResultMessage r;
|
||||
r.id = msg.id; r.ok = false;
|
||||
r.err = "connection stopped"; r.errCode = "TRANSPORT_CLOSED";
|
||||
p->set_value(std::move(r));
|
||||
return f;
|
||||
}
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_pendingCalls[msg.id] = p;
|
||||
}
|
||||
writeFrame(encodeFrame(*m_codec, AnyMessage{std::move(msg)}));
|
||||
return f;
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
std::future<MethodsResultMessage>
|
||||
RpcConnection<Stream>::sendMethods(MethodsMessage msg)
|
||||
{
|
||||
auto p = std::make_shared<std::promise<MethodsResultMessage>>();
|
||||
auto f = p->get_future();
|
||||
if (m_stopped.load()) {
|
||||
MethodsResultMessage r;
|
||||
r.id = msg.id; r.ok = false; r.err = "connection stopped";
|
||||
p->set_value(std::move(r));
|
||||
return f;
|
||||
}
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_pendingMethods[msg.id] = p;
|
||||
}
|
||||
writeFrame(encodeFrame(*m_codec, AnyMessage{std::move(msg)}));
|
||||
return f;
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::sendSubscribe(SubscribeMessage msg,
|
||||
std::function<void(EventMessage)> cb)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_eventCallbacks[{msg.object, msg.eventName}] = std::move(cb);
|
||||
}
|
||||
writeFrame(encodeFrame(*m_codec, AnyMessage{std::move(msg)}));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::sendUnsubscribe(UnsubscribeMessage msg)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_eventCallbacks.erase({msg.object, msg.eventName});
|
||||
}
|
||||
writeFrame(encodeFrame(*m_codec, AnyMessage{std::move(msg)}));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::sendEvent(EventMessage msg)
|
||||
{
|
||||
writeFrame(encodeFrame(*m_codec, AnyMessage{std::move(msg)}));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::sendToken(TokenMessage msg)
|
||||
{
|
||||
writeFrame(encodeFrame(*m_codec, AnyMessage{std::move(msg)}));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::writeFrame(std::vector<uint8_t> frame)
|
||||
{
|
||||
if (m_stopped.load()) return;
|
||||
auto self = this->shared_from_this();
|
||||
boost::asio::post(m_strand, [self, frame = std::move(frame)]() mutable {
|
||||
self->m_writeQueue.push_back(std::move(frame));
|
||||
if (!self->m_writing) {
|
||||
self->m_writing = true;
|
||||
self->doWrite();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::doWrite()
|
||||
{
|
||||
auto self = this->shared_from_this();
|
||||
boost::asio::async_write(m_stream,
|
||||
boost::asio::buffer(m_writeQueue.front()),
|
||||
boost::asio::bind_executor(m_strand,
|
||||
[self](const boost::system::error_code& ec, std::size_t /*n*/) {
|
||||
if (ec) { self->fail(ec.message()); return; }
|
||||
self->m_writeQueue.pop_front();
|
||||
if (self->m_writeQueue.empty()) {
|
||||
self->m_writing = false;
|
||||
} else {
|
||||
self->doWrite();
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
template <typename Stream>
|
||||
void RpcConnection<Stream>::fail(const std::string& reason)
|
||||
{
|
||||
bool expected = false;
|
||||
if (!m_stopped.compare_exchange_strong(expected, true)) return;
|
||||
|
||||
// Fail every pending promise with a transport-level error.
|
||||
std::map<uint64_t, std::shared_ptr<std::promise<ResultMessage>>> calls;
|
||||
std::map<uint64_t, std::shared_ptr<std::promise<MethodsResultMessage>>> methods;
|
||||
ErrorHandler errCb;
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
calls.swap(m_pendingCalls);
|
||||
methods.swap(m_pendingMethods);
|
||||
errCb.swap(m_error);
|
||||
m_eventCallbacks.clear();
|
||||
}
|
||||
for (auto& [id, p] : calls) {
|
||||
ResultMessage r; r.id = id; r.ok = false;
|
||||
r.err = reason; r.errCode = "TRANSPORT_ERROR";
|
||||
try { p->set_value(std::move(r)); } catch (...) {}
|
||||
}
|
||||
for (auto& [id, p] : methods) {
|
||||
MethodsResultMessage r; r.id = id; r.ok = false; r.err = reason;
|
||||
try { p->set_value(std::move(r)); } catch (...) {}
|
||||
}
|
||||
|
||||
boost::system::error_code ignore;
|
||||
try {
|
||||
// lowest_layer() works for plain asio::ip::tcp::socket (returns
|
||||
// itself) and for asio::ssl::stream (returns the underlying TCP
|
||||
// socket). Closing the lowest layer tears the stack down cleanly
|
||||
// without needing protocol-specific shutdown sequences.
|
||||
m_stream.lowest_layer().close(ignore);
|
||||
} catch (...) {}
|
||||
|
||||
// Notify the dispatch handler so it can drop any subscriptions still
|
||||
// keyed to this connection. Without this, a connection that drops
|
||||
// without sending Unsubscribe leaks sinks in the host's per-event map.
|
||||
if (m_handler) {
|
||||
try { m_handler->onConnectionClosed(static_cast<const void*>(this)); }
|
||||
catch (...) {}
|
||||
}
|
||||
|
||||
if (errCb) errCb(reason);
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_RPC_CONNECTION_H
|
||||
@@ -0,0 +1,76 @@
|
||||
#include "rpc_framing.h"
|
||||
|
||||
#include <cstring>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
namespace {
|
||||
|
||||
void writeBe32(std::vector<uint8_t>& out, uint32_t n)
|
||||
{
|
||||
out.push_back(static_cast<uint8_t>((n >> 24) & 0xff));
|
||||
out.push_back(static_cast<uint8_t>((n >> 16) & 0xff));
|
||||
out.push_back(static_cast<uint8_t>((n >> 8) & 0xff));
|
||||
out.push_back(static_cast<uint8_t>( n & 0xff));
|
||||
}
|
||||
|
||||
uint32_t readBe32(const uint8_t* p)
|
||||
{
|
||||
return (uint32_t(p[0]) << 24) | (uint32_t(p[1]) << 16)
|
||||
| (uint32_t(p[2]) << 8) | uint32_t(p[3]);
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
std::vector<uint8_t>
|
||||
encodeFrame(MessageType tag, const std::vector<uint8_t>& payload)
|
||||
{
|
||||
// Total frame body length = 1 byte (tag) + payload bytes. The 4-byte
|
||||
// length prefix is not included in the length value it describes.
|
||||
const uint64_t bodyLen = 1u + payload.size();
|
||||
if (bodyLen > kMaxFrameLength)
|
||||
throw FramingError("frame too large");
|
||||
|
||||
std::vector<uint8_t> out;
|
||||
out.reserve(4 + bodyLen);
|
||||
writeBe32(out, static_cast<uint32_t>(bodyLen));
|
||||
out.push_back(static_cast<uint8_t>(tag));
|
||||
out.insert(out.end(), payload.begin(), payload.end());
|
||||
return out;
|
||||
}
|
||||
|
||||
std::vector<uint8_t>
|
||||
encodeFrame(IWireCodec& codec, const AnyMessage& msg)
|
||||
{
|
||||
return encodeFrame(messageTypeOf(msg), codec.encode(msg));
|
||||
}
|
||||
|
||||
void FrameReader::append(const uint8_t* data, std::size_t len)
|
||||
{
|
||||
m_buf.insert(m_buf.end(), data, data + len);
|
||||
}
|
||||
|
||||
bool FrameReader::next(MessageType& tag, std::vector<uint8_t>& payload)
|
||||
{
|
||||
// Need at least 4 bytes for the length prefix.
|
||||
if (m_buf.size() < 4) return false;
|
||||
|
||||
const uint32_t bodyLen = readBe32(m_buf.data());
|
||||
if (bodyLen == 0) throw FramingError("zero-length frame");
|
||||
if (bodyLen > kMaxFrameLength) throw FramingError("frame length exceeds cap");
|
||||
|
||||
const std::size_t total = 4u + bodyLen;
|
||||
if (m_buf.size() < total) return false;
|
||||
|
||||
// Tag + payload.
|
||||
tag = static_cast<MessageType>(m_buf[4]);
|
||||
payload.assign(m_buf.begin() + 5, m_buf.begin() + total);
|
||||
|
||||
// Drop the consumed bytes. erase from front is O(n) in buffer size;
|
||||
// acceptable while frames are small and rare. Can switch to a ring
|
||||
// buffer if profiling ever flags this.
|
||||
m_buf.erase(m_buf.begin(), m_buf.begin() + total);
|
||||
return true;
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,62 @@
|
||||
#ifndef LOGOS_PLAIN_RPC_FRAMING_H
|
||||
#define LOGOS_PLAIN_RPC_FRAMING_H
|
||||
|
||||
#include "rpc_message.h"
|
||||
#include "wire_codec.h"
|
||||
|
||||
#include <cstdint>
|
||||
#include <stdexcept>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// Wire framing: one frame is
|
||||
//
|
||||
// [4-byte big-endian length N][1-byte MessageType tag][N-1 payload bytes]
|
||||
//
|
||||
// The length covers the tag + payload. 4 bytes give us 4 GiB max per frame
|
||||
// (way more than we'll ever need); 1-byte tag holds a MessageType enum
|
||||
// value. Payload bytes come from an IWireCodec.
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
class FramingError : public std::runtime_error {
|
||||
public:
|
||||
using std::runtime_error::runtime_error;
|
||||
};
|
||||
|
||||
// Maximum accepted frame length (length-prefix value). 16 MiB is plenty for
|
||||
// anything we'll RPC today, and guards against runaway allocation on
|
||||
// malformed input.
|
||||
constexpr uint32_t kMaxFrameLength = 16u * 1024u * 1024u;
|
||||
|
||||
// Build a complete frame around a codec-produced payload.
|
||||
std::vector<uint8_t>
|
||||
encodeFrame(MessageType tag, const std::vector<uint8_t>& payload);
|
||||
|
||||
// Convenience: encode message via codec + frame in one step.
|
||||
std::vector<uint8_t> encodeFrame(IWireCodec& codec, const AnyMessage& msg);
|
||||
|
||||
// Incremental reader: feed bytes via `append`, pull complete frames out via
|
||||
// `next`. Useful with Asio async reads that deliver arbitrary chunk sizes.
|
||||
class FrameReader {
|
||||
public:
|
||||
void append(const uint8_t* data, std::size_t len);
|
||||
void append(const std::vector<uint8_t>& chunk) {
|
||||
append(chunk.data(), chunk.size());
|
||||
}
|
||||
|
||||
// Returns true and populates `tag` + `payload` with one frame's worth
|
||||
// of data. Returns false if the buffer doesn't yet contain a complete
|
||||
// frame. Throws FramingError on unrecoverable corruption.
|
||||
bool next(MessageType& tag, std::vector<uint8_t>& payload);
|
||||
|
||||
std::size_t buffered() const { return m_buf.size(); }
|
||||
|
||||
private:
|
||||
std::vector<uint8_t> m_buf;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_RPC_FRAMING_H
|
||||
@@ -0,0 +1,22 @@
|
||||
#include "rpc_message.h"
|
||||
|
||||
#include <type_traits>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
MessageType messageTypeOf(const AnyMessage& m)
|
||||
{
|
||||
return std::visit([](const auto& v) -> MessageType {
|
||||
using T = std::decay_t<decltype(v)>;
|
||||
if constexpr (std::is_same_v<T, CallMessage>) return MessageType::Call;
|
||||
else if constexpr (std::is_same_v<T, ResultMessage>) return MessageType::Result;
|
||||
else if constexpr (std::is_same_v<T, SubscribeMessage>)return MessageType::Subscribe;
|
||||
else if constexpr (std::is_same_v<T, UnsubscribeMessage>) return MessageType::Unsubscribe;
|
||||
else if constexpr (std::is_same_v<T, EventMessage>) return MessageType::Event;
|
||||
else if constexpr (std::is_same_v<T, TokenMessage>) return MessageType::Token;
|
||||
else if constexpr (std::is_same_v<T, MethodsMessage>) return MessageType::Methods;
|
||||
else if constexpr (std::is_same_v<T, MethodsResultMessage>) return MessageType::MethodsResult;
|
||||
}, m);
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,120 @@
|
||||
#ifndef LOGOS_PLAIN_RPC_MESSAGE_H
|
||||
#define LOGOS_PLAIN_RPC_MESSAGE_H
|
||||
|
||||
#include "rpc_value.h"
|
||||
|
||||
#include <cstdint>
|
||||
#include <string>
|
||||
#include <variant>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// Wire message definitions.
|
||||
//
|
||||
// Each on-wire frame is [4-byte big-endian length][1-byte type tag][payload].
|
||||
// The payload encoding is decided by the codec (JSON today, CBOR planned).
|
||||
//
|
||||
// `MessageType` doubles as the 1-byte type tag so the codec can decode the
|
||||
// right struct without peeking inside the payload. Keep the values stable —
|
||||
// they're on the wire.
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
enum class MessageType : uint8_t {
|
||||
Call = 1,
|
||||
Result = 2,
|
||||
Subscribe = 3,
|
||||
Unsubscribe = 4,
|
||||
Event = 5,
|
||||
Token = 6,
|
||||
Methods = 7,
|
||||
MethodsResult = 8,
|
||||
};
|
||||
|
||||
struct MethodMetadata {
|
||||
std::string name;
|
||||
std::string signature;
|
||||
std::string returnType;
|
||||
bool isInvokable = true;
|
||||
RpcList parameters; // list of {name, type} maps — schema-flexible
|
||||
};
|
||||
|
||||
// Call <module>.<method>(args...). The response is a Result with the same id.
|
||||
struct CallMessage {
|
||||
uint64_t id;
|
||||
std::string authToken;
|
||||
std::string object;
|
||||
std::string method;
|
||||
std::vector<RpcValue> args;
|
||||
};
|
||||
|
||||
// Response to a Call (or Methods) message, matched by id.
|
||||
struct ResultMessage {
|
||||
uint64_t id;
|
||||
bool ok = false;
|
||||
RpcValue value; // present when ok
|
||||
std::string err; // present when !ok
|
||||
std::string errCode; // present when !ok
|
||||
};
|
||||
|
||||
// Subscribe / unsubscribe to a named event on an object.
|
||||
// After Subscribe, the peer pushes EventMessage frames whenever the provider
|
||||
// emits that event until an Unsubscribe arrives (or the connection closes).
|
||||
// eventName "" means "all events on this object" (wildcard).
|
||||
struct SubscribeMessage {
|
||||
std::string object;
|
||||
std::string eventName;
|
||||
};
|
||||
struct UnsubscribeMessage {
|
||||
std::string object;
|
||||
std::string eventName;
|
||||
};
|
||||
|
||||
// Fire-and-forget event delivery from provider → subscriber.
|
||||
struct EventMessage {
|
||||
std::string object;
|
||||
std::string eventName;
|
||||
std::vector<RpcValue> data;
|
||||
};
|
||||
|
||||
// Authorization token that the consumer wants registered for a specific
|
||||
// module name. Mirrors LogosObject::informModuleToken today.
|
||||
struct TokenMessage {
|
||||
std::string authToken;
|
||||
std::string moduleName;
|
||||
std::string token;
|
||||
};
|
||||
|
||||
// Query the set of methods a published object exposes. Response is a
|
||||
// MethodsResult keyed to the same id.
|
||||
struct MethodsMessage {
|
||||
uint64_t id;
|
||||
std::string authToken;
|
||||
std::string object;
|
||||
};
|
||||
struct MethodsResultMessage {
|
||||
uint64_t id;
|
||||
bool ok = false;
|
||||
std::vector<MethodMetadata> methods;
|
||||
std::string err;
|
||||
};
|
||||
|
||||
// Tagged union of every message the wire stack knows.
|
||||
using AnyMessage = std::variant<
|
||||
CallMessage,
|
||||
ResultMessage,
|
||||
SubscribeMessage,
|
||||
UnsubscribeMessage,
|
||||
EventMessage,
|
||||
TokenMessage,
|
||||
MethodsMessage,
|
||||
MethodsResultMessage
|
||||
>;
|
||||
|
||||
// Lookup: AnyMessage variant ↔ MessageType tag.
|
||||
MessageType messageTypeOf(const AnyMessage& m);
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_RPC_MESSAGE_H
|
||||
@@ -0,0 +1,181 @@
|
||||
#include "rpc_server.h"
|
||||
|
||||
#include <boost/asio/ip/address.hpp>
|
||||
|
||||
#include <QDebug>
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// ── RpcServerTcp ──────────────────────────────────────────────────────────
|
||||
|
||||
RpcServerTcp::RpcServerTcp(boost::asio::io_context& ioc,
|
||||
const std::string& host,
|
||||
uint16_t port,
|
||||
std::shared_ptr<IWireCodec> codec,
|
||||
IncomingCallHandler* handler)
|
||||
: m_acceptor(ioc)
|
||||
, m_codec(std::move(codec))
|
||||
, m_handler(handler)
|
||||
, m_host(host)
|
||||
, m_port(port)
|
||||
{
|
||||
}
|
||||
|
||||
bool RpcServerTcp::start()
|
||||
{
|
||||
boost::system::error_code ec;
|
||||
boost::asio::ip::tcp::endpoint ep(
|
||||
boost::asio::ip::make_address(m_host, ec),
|
||||
m_port);
|
||||
if (ec) return false;
|
||||
|
||||
m_acceptor.open(ep.protocol(), ec); if (ec) return false;
|
||||
m_acceptor.set_option(boost::asio::socket_base::reuse_address(true), ec);
|
||||
m_acceptor.bind(ep, ec); if (ec) return false;
|
||||
m_acceptor.listen(boost::asio::socket_base::max_listen_connections, ec);
|
||||
if (ec) return false;
|
||||
|
||||
m_boundPort = m_acceptor.local_endpoint().port();
|
||||
doAccept();
|
||||
return true;
|
||||
}
|
||||
|
||||
void RpcServerTcp::stop()
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_stopped = true;
|
||||
boost::system::error_code ignore;
|
||||
m_acceptor.close(ignore);
|
||||
for (auto& c : m_conns) c->stop("server stopped");
|
||||
m_conns.clear();
|
||||
}
|
||||
|
||||
void RpcServerTcp::doAccept()
|
||||
{
|
||||
auto self = shared_from_this();
|
||||
m_acceptor.async_accept(
|
||||
[self](const boost::system::error_code& ec,
|
||||
boost::asio::ip::tcp::socket socket) {
|
||||
if (ec) return; // acceptor probably closed; quietly exit.
|
||||
auto conn = std::make_shared<TcpConnection>(
|
||||
std::move(socket), self->m_codec, self->m_handler);
|
||||
{
|
||||
std::lock_guard<std::mutex> g(self->m_mu);
|
||||
if (self->m_stopped) { conn->stop("server stopped"); return; }
|
||||
self->m_conns.push_back(conn);
|
||||
}
|
||||
std::weak_ptr<RpcServerTcp> weakSelf = self;
|
||||
conn->setErrorHandler([weakSelf, conn](const std::string&) {
|
||||
auto s = weakSelf.lock();
|
||||
if (!s) return;
|
||||
std::lock_guard<std::mutex> g(s->m_mu);
|
||||
s->m_conns.erase(
|
||||
std::remove(s->m_conns.begin(), s->m_conns.end(), conn),
|
||||
s->m_conns.end());
|
||||
});
|
||||
conn->start();
|
||||
self->doAccept();
|
||||
});
|
||||
}
|
||||
|
||||
// ── RpcServerSsl ──────────────────────────────────────────────────────────
|
||||
|
||||
RpcServerSsl::RpcServerSsl(boost::asio::io_context& ioc,
|
||||
const std::string& host,
|
||||
uint16_t port,
|
||||
boost::asio::ssl::context sslCtx,
|
||||
std::shared_ptr<IWireCodec> codec,
|
||||
IncomingCallHandler* handler)
|
||||
: m_acceptor(ioc)
|
||||
, m_sslCtx(std::move(sslCtx))
|
||||
, m_codec(std::move(codec))
|
||||
, m_handler(handler)
|
||||
, m_host(host)
|
||||
, m_port(port)
|
||||
{
|
||||
}
|
||||
|
||||
bool RpcServerSsl::start()
|
||||
{
|
||||
boost::system::error_code ec;
|
||||
boost::asio::ip::tcp::endpoint ep(
|
||||
boost::asio::ip::make_address(m_host, ec),
|
||||
m_port);
|
||||
if (ec) return false;
|
||||
|
||||
m_acceptor.open(ep.protocol(), ec); if (ec) return false;
|
||||
m_acceptor.set_option(boost::asio::socket_base::reuse_address(true), ec);
|
||||
m_acceptor.bind(ep, ec); if (ec) return false;
|
||||
m_acceptor.listen(boost::asio::socket_base::max_listen_connections, ec);
|
||||
if (ec) return false;
|
||||
|
||||
m_boundPort = m_acceptor.local_endpoint().port();
|
||||
doAccept();
|
||||
return true;
|
||||
}
|
||||
|
||||
void RpcServerSsl::stop()
|
||||
{
|
||||
std::lock_guard<std::mutex> g(m_mu);
|
||||
m_stopped = true;
|
||||
boost::system::error_code ignore;
|
||||
m_acceptor.close(ignore);
|
||||
for (auto& c : m_conns) c->stop("server stopped");
|
||||
m_conns.clear();
|
||||
}
|
||||
|
||||
void RpcServerSsl::doAccept()
|
||||
{
|
||||
auto self = shared_from_this();
|
||||
m_acceptor.async_accept(
|
||||
[self](const boost::system::error_code& ec,
|
||||
boost::asio::ip::tcp::socket socket) {
|
||||
if (ec) return;
|
||||
auto stream = std::make_shared<SslStream>(std::move(socket), self->m_sslCtx);
|
||||
stream->async_handshake(
|
||||
boost::asio::ssl::stream_base::server,
|
||||
[self, stream](const boost::system::error_code& hs) {
|
||||
if (hs) {
|
||||
// Handshake failed. Log the error — silently
|
||||
// discarding the socket used to be a debugging
|
||||
// dead-end (clients see a generic "alert 40"
|
||||
// and can't tell if the cert is bad, a curve
|
||||
// is unavailable, or the listener picked a
|
||||
// version the client refuses). Category + code
|
||||
// + message give enough to grep for.
|
||||
qWarning().nospace()
|
||||
<< "RpcServerSsl: TLS handshake failed: "
|
||||
<< hs.category().name() << ':' << hs.value()
|
||||
<< " (" << QString::fromStdString(hs.message()) << ")";
|
||||
return;
|
||||
}
|
||||
// Hand the SslStream off to a connection that owns
|
||||
// it. We held it in a shared_ptr only for the
|
||||
// duration of async_handshake (so the buffer
|
||||
// outlives the dispatch); now move the underlying
|
||||
// stream into the connection by value.
|
||||
auto conn = std::make_shared<SslConnection>(
|
||||
std::move(*stream), self->m_codec, self->m_handler);
|
||||
{
|
||||
std::lock_guard<std::mutex> g(self->m_mu);
|
||||
if (self->m_stopped) { conn->stop("server stopped"); return; }
|
||||
self->m_conns.push_back(conn);
|
||||
}
|
||||
std::weak_ptr<RpcServerSsl> weakSelf = self;
|
||||
conn->setErrorHandler([weakSelf, conn](const std::string&) {
|
||||
auto s = weakSelf.lock();
|
||||
if (!s) return;
|
||||
std::lock_guard<std::mutex> g(s->m_mu);
|
||||
s->m_conns.erase(
|
||||
std::remove(s->m_conns.begin(), s->m_conns.end(), conn),
|
||||
s->m_conns.end());
|
||||
});
|
||||
conn->start();
|
||||
});
|
||||
self->doAccept();
|
||||
});
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
@@ -0,0 +1,105 @@
|
||||
#ifndef LOGOS_PLAIN_RPC_SERVER_H
|
||||
#define LOGOS_PLAIN_RPC_SERVER_H
|
||||
|
||||
#include "incoming_call_handler.h"
|
||||
#include "rpc_connection.h"
|
||||
#include "wire_codec.h"
|
||||
|
||||
#include <boost/asio/ip/tcp.hpp>
|
||||
#include <boost/asio/ssl/context.hpp>
|
||||
#include <boost/asio/ssl/stream.hpp>
|
||||
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// RpcServer (TCP) — accepts TCP connections, wraps each in an
|
||||
// RpcConnection, keeps them alive until they drop.
|
||||
//
|
||||
// Every accepted connection uses the same shared IncomingCallHandler so
|
||||
// the provider layer can dispatch regardless of which client is talking.
|
||||
// The server doesn't multiplex objects by itself; it's the handler's job
|
||||
// to look up the target object for each incoming CallMessage.
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
using TcpStream = boost::asio::ip::tcp::socket;
|
||||
using TcpConnection = RpcConnection<TcpStream>;
|
||||
|
||||
class RpcServerTcp : public std::enable_shared_from_this<RpcServerTcp> {
|
||||
public:
|
||||
RpcServerTcp(boost::asio::io_context& ioc,
|
||||
const std::string& host,
|
||||
uint16_t port,
|
||||
std::shared_ptr<IWireCodec> codec,
|
||||
IncomingCallHandler* handler);
|
||||
|
||||
// Start accepting. Returns false if bind fails.
|
||||
bool start();
|
||||
|
||||
// Actual bound port (useful when the caller requested port=0).
|
||||
uint16_t boundPort() const { return m_boundPort; }
|
||||
|
||||
void stop();
|
||||
|
||||
private:
|
||||
void doAccept();
|
||||
|
||||
boost::asio::ip::tcp::acceptor m_acceptor;
|
||||
std::shared_ptr<IWireCodec> m_codec;
|
||||
IncomingCallHandler* m_handler;
|
||||
std::string m_host;
|
||||
uint16_t m_port;
|
||||
uint16_t m_boundPort = 0;
|
||||
|
||||
std::mutex m_mu;
|
||||
std::vector<std::shared_ptr<TcpConnection>> m_conns;
|
||||
bool m_stopped = false;
|
||||
};
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// RpcServer (TLS) — same as TCP but wraps every accepted socket in an
|
||||
// asio::ssl::stream and completes the handshake before spinning up the
|
||||
// RpcConnection.
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
using SslStream = boost::asio::ssl::stream<boost::asio::ip::tcp::socket>;
|
||||
using SslConnection = RpcConnection<SslStream>;
|
||||
|
||||
class RpcServerSsl : public std::enable_shared_from_this<RpcServerSsl> {
|
||||
public:
|
||||
RpcServerSsl(boost::asio::io_context& ioc,
|
||||
const std::string& host,
|
||||
uint16_t port,
|
||||
boost::asio::ssl::context sslCtx,
|
||||
std::shared_ptr<IWireCodec> codec,
|
||||
IncomingCallHandler* handler);
|
||||
|
||||
bool start();
|
||||
uint16_t boundPort() const { return m_boundPort; }
|
||||
void stop();
|
||||
|
||||
private:
|
||||
void doAccept();
|
||||
|
||||
boost::asio::ip::tcp::acceptor m_acceptor;
|
||||
boost::asio::ssl::context m_sslCtx;
|
||||
std::shared_ptr<IWireCodec> m_codec;
|
||||
IncomingCallHandler* m_handler;
|
||||
std::string m_host;
|
||||
uint16_t m_port;
|
||||
uint16_t m_boundPort = 0;
|
||||
|
||||
std::mutex m_mu;
|
||||
std::vector<std::shared_ptr<SslConnection>> m_conns;
|
||||
bool m_stopped = false;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_RPC_SERVER_H
|
||||
@@ -0,0 +1,128 @@
|
||||
#ifndef LOGOS_PLAIN_RPC_VALUE_H
|
||||
#define LOGOS_PLAIN_RPC_VALUE_H
|
||||
|
||||
#include <algorithm>
|
||||
#include <cstdint>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include <variant>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// RpcValue — plain C++ variant carried by the wire RPC layer.
|
||||
//
|
||||
// Covers the shapes we actually need (null / bool / int / double / string /
|
||||
// bytes / list / map). No Qt types. The JSON/CBOR codec converts to/from
|
||||
// `nlohmann::json`; Qt-side callers convert to/from `QVariant` at the Qt
|
||||
// boundary (see plain_logos_object.cpp, plain_transport_host.cpp).
|
||||
//
|
||||
// Uses recursive std::variant via wrapper structs so list/map can hold
|
||||
// RpcValue children without forward-declaration headaches.
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
struct RpcValue;
|
||||
|
||||
struct RpcList {
|
||||
std::vector<RpcValue> items;
|
||||
bool operator==(const RpcList& other) const { return items == other.items; }
|
||||
bool operator!=(const RpcList& other) const { return !(*this == other); }
|
||||
};
|
||||
|
||||
// std::map<string, RpcValue> would require RpcValue to be complete at this
|
||||
// point, which is impossible (RpcValue contains RpcMap as a variant alt).
|
||||
// Use a vector of pairs instead — also gives us deterministic encoding
|
||||
// order for free, which matters when we move to CBOR.
|
||||
//
|
||||
// Method bodies that dereference RpcValue are defined out-of-line below,
|
||||
// once RpcValue is complete.
|
||||
struct RpcMap {
|
||||
std::vector<std::pair<std::string, RpcValue>> entries;
|
||||
|
||||
void emplace(std::string key, RpcValue val);
|
||||
const RpcValue* find(const std::string& key) const;
|
||||
const RpcValue& at(const std::string& key) const;
|
||||
|
||||
bool operator==(const RpcMap& other) const;
|
||||
bool operator!=(const RpcMap& other) const { return !(*this == other); }
|
||||
};
|
||||
|
||||
struct RpcBytes {
|
||||
std::vector<uint8_t> data;
|
||||
bool operator==(const RpcBytes& other) const { return data == other.data; }
|
||||
bool operator!=(const RpcBytes& other) const { return !(*this == other); }
|
||||
};
|
||||
|
||||
struct RpcValue {
|
||||
using Variant = std::variant<
|
||||
std::monostate, // null
|
||||
bool,
|
||||
int64_t,
|
||||
double,
|
||||
std::string,
|
||||
RpcBytes,
|
||||
RpcList,
|
||||
RpcMap
|
||||
>;
|
||||
|
||||
Variant value;
|
||||
|
||||
RpcValue() = default;
|
||||
RpcValue(std::monostate) : value(std::monostate{}) {}
|
||||
RpcValue(bool b) : value(b) {}
|
||||
RpcValue(int i) : value(static_cast<int64_t>(i)) {}
|
||||
RpcValue(int64_t i) : value(i) {}
|
||||
RpcValue(double d) : value(d) {}
|
||||
RpcValue(const char* s) : value(std::string(s)) {}
|
||||
RpcValue(std::string s) : value(std::move(s)) {}
|
||||
RpcValue(RpcBytes b) : value(std::move(b)) {}
|
||||
RpcValue(RpcList l) : value(std::move(l)) {}
|
||||
RpcValue(RpcMap m) : value(std::move(m)) {}
|
||||
|
||||
bool isNull() const { return std::holds_alternative<std::monostate>(value); }
|
||||
bool isBool() const { return std::holds_alternative<bool>(value); }
|
||||
bool isInt() const { return std::holds_alternative<int64_t>(value); }
|
||||
bool isDouble() const { return std::holds_alternative<double>(value); }
|
||||
bool isString() const { return std::holds_alternative<std::string>(value); }
|
||||
bool isBytes() const { return std::holds_alternative<RpcBytes>(value); }
|
||||
bool isList() const { return std::holds_alternative<RpcList>(value); }
|
||||
bool isMap() const { return std::holds_alternative<RpcMap>(value); }
|
||||
|
||||
bool asBool() const { return std::get<bool>(value); }
|
||||
int64_t asInt() const { return std::get<int64_t>(value); }
|
||||
double asDouble() const { return std::get<double>(value); }
|
||||
const std::string& asString() const { return std::get<std::string>(value); }
|
||||
const RpcBytes& asBytes() const { return std::get<RpcBytes>(value); }
|
||||
const RpcList& asList() const { return std::get<RpcList>(value); }
|
||||
const RpcMap& asMap() const { return std::get<RpcMap>(value); }
|
||||
|
||||
bool operator==(const RpcValue& other) const { return value == other.value; }
|
||||
bool operator!=(const RpcValue& other) const { return !(*this == other); }
|
||||
};
|
||||
|
||||
// ── RpcMap out-of-line methods (need complete RpcValue) ────────────────────
|
||||
|
||||
inline void RpcMap::emplace(std::string key, RpcValue val) {
|
||||
entries.emplace_back(std::move(key), std::move(val));
|
||||
}
|
||||
|
||||
inline const RpcValue* RpcMap::find(const std::string& key) const {
|
||||
for (const auto& kv : entries) if (kv.first == key) return &kv.second;
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
inline const RpcValue& RpcMap::at(const std::string& key) const {
|
||||
const RpcValue* v = find(key);
|
||||
if (!v) throw std::out_of_range("RpcMap::at: key not found: " + key);
|
||||
return *v;
|
||||
}
|
||||
|
||||
inline bool RpcMap::operator==(const RpcMap& other) const {
|
||||
return entries == other.entries;
|
||||
}
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_RPC_VALUE_H
|
||||
@@ -0,0 +1,50 @@
|
||||
#ifndef LOGOS_PLAIN_WIRE_CODEC_H
|
||||
#define LOGOS_PLAIN_WIRE_CODEC_H
|
||||
|
||||
#include "rpc_message.h"
|
||||
|
||||
#include <cstdint>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace logos::plain {
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
// IWireCodec — pluggable (de)serializer for the wire message set.
|
||||
//
|
||||
// The plan calls out CDDL/CBOR as the intended future format. Shipping
|
||||
// implementation is `JsonCodec` (nlohmann::json::dump / parse). A future
|
||||
// `CborCodec` uses `to_cbor` / `from_cbor` on the same message structs; the
|
||||
// messages don't change.
|
||||
//
|
||||
// `encode` / `decode` produce / consume bare payload bytes — they do NOT
|
||||
// handle framing (length prefix or type tag). Framing is `rpc_framing.{h,cpp}`.
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
class CodecError : public std::runtime_error {
|
||||
public:
|
||||
using std::runtime_error::runtime_error;
|
||||
};
|
||||
|
||||
class IWireCodec {
|
||||
public:
|
||||
virtual ~IWireCodec() = default;
|
||||
|
||||
// Encode one message to bytes. The caller stamps the type tag and
|
||||
// length prefix around the returned payload.
|
||||
virtual std::vector<uint8_t> encode(const AnyMessage&) = 0;
|
||||
|
||||
// Decode one payload of the given type (learned from the 1-byte tag).
|
||||
// Throws CodecError on malformed input.
|
||||
virtual AnyMessage decode(MessageType tag,
|
||||
const uint8_t* data,
|
||||
std::size_t len) = 0;
|
||||
|
||||
// Human-readable name: "json" | "cbor" | ...
|
||||
virtual std::string name() const = 0;
|
||||
};
|
||||
|
||||
} // namespace logos::plain
|
||||
|
||||
#endif // LOGOS_PLAIN_WIRE_CODEC_H
|
||||
Reference in New Issue
Block a user