From faeb64207fe9c300b92fbceb2ca4c8442fa0ab3d Mon Sep 17 00:00:00 2001 From: Calin Culianu Date: Tue, 30 Apr 2019 19:21:47 +0300 Subject: [PATCH] Finally.. finished refactoring out the RPC bits. Now we can re-use the RPC method<->result code in the TcpServer side which will face wallets. Phew! Took me long enough! It's a minimal JSON-RPC protocol impl.. but it'll do. The 1 nice bit is the Schema spec I came up with which more-or-less works well enough as a first-pass validatior. Further passes are needed in interested client code, but the initial pass validation can be done in a thread so as to not waste the main thread's time validating json or dict key presence/absense. Yay. --- AbstractConnection.cpp | 17 +++-- AbstractConnection.h | 9 ++- EXClient.cpp | 167 ++++------------------------------------- EXClient.h | 38 +--------- EXMgr.cpp | 139 ++++++++++++++-------------------- EXMgr.h | 8 +- RPC.cpp | 34 +++++---- RPC.h | 45 +++++++++-- TcpServer.cpp | 10 +-- TcpServer.h | 5 +- 10 files changed, 157 insertions(+), 315 deletions(-) diff --git a/AbstractConnection.cpp b/AbstractConnection.cpp index 8e1e408..0770a29 100644 --- a/AbstractConnection.cpp +++ b/AbstractConnection.cpp @@ -30,7 +30,7 @@ bool AbstractConnection::isStale() const return isGood() && Util::getTime() - lastGood > stale_threshold; } -void AbstractConnection::disconnect(bool graceful) +void AbstractConnection::do_disconnect(bool graceful) { status = status == Bad ? Bad : NotConnected; // try and keep Bad status around so EXMgr can decide when to reconnect based on it if (socket) { @@ -67,7 +67,7 @@ bool AbstractConnection::do_write(const QByteArray & data) qint64 written = socket->write(data2write); if (written < 0) { Error() << __FUNCTION__ << " error on write " << socket->error() << " (" << socket->errorString() << ") id=" << id; - disconnect(); + do_disconnect(); return false; } else if (written < data2write.length()) { writeBackLog = data2write.mid(int(written)); @@ -75,7 +75,7 @@ bool AbstractConnection::do_write(const QByteArray & data) nSent += written; if (writeBackLog.length() > MAX_BUFFER) { Error() << __FUNCTION__ << " MAX_BUFFER reached on write (" << MAX_BUFFER << ") id=" << id; - disconnect(); + do_disconnect(); return false; } return true; @@ -102,12 +102,14 @@ void AbstractConnection::on_pingTimer() do_ping(); } +void AbstractConnection::slot_on_readyRead() { on_readyRead(); } + void AbstractConnection::on_connected() { // runs in our thread's context Debug() << __FUNCTION__; connectedConns.push_back(connect(this, &AbstractConnection::send, this, &AbstractConnection::do_write)); - connectedConns.push_back(connect(socket, SIGNAL(readyRead()), this, SLOT(on_readyRead()))); + connectedConns.push_back(connect(socket, SIGNAL(readyRead()), this, SLOT(slot_on_readyRead()))); if (dynamic_cast(socket)) { // for some reason Qt can't find this old-style signal for QSslSocket so we do the below. // Additionally, bytesWritten is never emitted for QSslSocket, violating OOP! Thanks Qt. :P @@ -119,8 +121,9 @@ void AbstractConnection::on_connected() connect(socket, &QAbstractSocket::disconnected, this, [this]{ Debug() << prettyName() << " socket disconnected"; for (const auto & connection : connectedConns) { - disconnect(connection); + QObject::disconnect(connection); } + connectedConns.clear(); // be sure to empty the list out when we are done! kill_pingTimer(); on_disconnected(); emit lostConnection(this); @@ -132,7 +135,7 @@ void AbstractConnection::on_connected() void AbstractConnection::on_disconnected() { - /* nothing, here for derived classes */ + /* nothing, here; for derived classes to override if they wish. */ } void AbstractConnection::on_socketState(QAbstractSocket::SocketState s) @@ -171,6 +174,6 @@ void AbstractConnection::do_ping() void AbstractConnection::on_error(QAbstractSocket::SocketError err) { Warning() << prettyName() << ": error " << err << " (" << (socket ? socket->errorString() : "(null)") << ")"; - disconnect(); + do_disconnect(); } diff --git a/AbstractConnection.h b/AbstractConnection.h index 90565a9..1837c2b 100644 --- a/AbstractConnection.h +++ b/AbstractConnection.h @@ -31,11 +31,10 @@ signals: /// This is a low-level function subclasses should create their own high-level protocol-level signals / methods; void send(QByteArray); -protected slots: - virtual void on_readyRead() = 0; /**< Implement in subclasses -- required to read data */ - protected: + virtual void on_readyRead() = 0; /**< Implement in subclasses -- required to read data */ + enum Status { NotConnected = 0, Connecting, @@ -70,13 +69,15 @@ protected: virtual void on_disconnected(); ///< overrides can chain to this as well bool do_write(const QByteArray & = ""); - virtual void disconnect(bool graceful = false); /// does a socket->abort, sets status. Chain to this if you want on override. + /// does a socket->abort, sets status. Chain to this if you want on override. Named this way so as not to clash with QObject::disconnect + virtual void do_disconnect(bool graceful = false); private slots: void on_pingTimer(); void on_bytesWritten(); void on_error(QAbstractSocket::SocketError); void on_socketState(QAbstractSocket::SocketState); + void slot_on_readyRead(); ///< calls virtual method on_readyRead for us -- I was paranoid about Qt signal/slot binding semantics and prefer to call from within a function explicitly, hence this redundant method. private: void start_pingTimer(); void kill_pingTimer(); diff --git a/EXClient.cpp b/EXClient.cpp index 458f204..b1a5576 100644 --- a/EXClient.cpp +++ b/EXClient.cpp @@ -4,16 +4,8 @@ #include -class BadServerReply : public Exception { -public: - using Exception::Exception; /// bring in c'tor - ~BadServerReply(); -}; - -BadServerReply::~BadServerReply() {} // for vtable - EXClient::EXClient(EXMgr *mgr, qint64 id, const QString &host, quint16 tport, quint16 sport) - : AbstractConnection(id, nullptr), host(host), tport(tport), sport(sport), mgr(mgr) + : RPC::Connection(mgr->rpcMethods(), id, nullptr), host(host), tport(tport), sport(sport), mgr(mgr) { Debug() << __FUNCTION__ << " host:" << host << " t:" << tport << " s:" << sport; _thread.setObjectName(QString("%1 %2").arg("EXClient").arg(host)); @@ -33,12 +25,12 @@ QString EXClient::prettyName(bool dontTouchSocket) const Warning() << __PRETTY_FUNCTION__ << " called from another thread! FIXME!"; dontTouchSocket = true; } - return AbstractConnection::prettyName(dontTouchSocket); + return RPC::Connection::prettyName(dontTouchSocket); } bool EXClient::isGood() const { - return AbstractConnection::isGood() && _thread.isRunning() && info.isValid(); + return RPC::Connection::isGood() && _thread.isRunning() && info.isValid(); } @@ -61,7 +53,7 @@ void EXClient::killSocket() { if (socket && socket->state() != QAbstractSocket::UnconnectedState) { Debug() << host << " aborting connection"; - disconnect(); + do_disconnect(); } delete socket; socket = nullptr; // delete of nullptr ok status = NotConnected; @@ -97,21 +89,6 @@ void EXClient::reconnect() } -void EXClient::_sendRequest(qint64 id, const QString &method, const QVariantList ¶ms) -{ - if (status != Connected || !socket) { - Error() << __FUNCTION__ << " method: " << method << "; Not connected!"; - return; - } - while (idMethodMap.size() > 20000) { // prevent memory leaks in case of misbehaving server - idMethodMap.erase(idMethodMap.begin()); - } - idMethodMap[id] = method; - - emit send(makeRequestData(id, method, params)); // ends up calling do_write immediately (which is connected to send) -} - - void EXClient::do_ping() { emit sendRequest(mgr->newId(), "server.ping"); @@ -120,136 +97,20 @@ void EXClient::do_ping() void EXClient::on_connected() { // runs in thread - AbstractConnection::on_connected(); - connectedConns.push_back(connect(this, &EXClient::sendRequest, this, &EXClient::_sendRequest)); // connection will be auto-disconnected on socket disconnect + RPC::Connection::on_connected(); + connectedConns.push_back( + connect(this, &RPC::Connection::gotMessage, this, + [this](RPC::Connection *c, const RPC::Message &m) + { + if (this == c) emit EXClient::gotMessage(this, m); /// re-emits as EXClient * signal (different C++ signature) + else Error() << "this != c for gotMessage fwd! FIXME!"; + }) + ); // connection will be auto-disconnected on socket disconnect in superclass on_disconnected impl. emit newConnection(this); } void EXClient::on_disconnected() { - AbstractConnection::on_disconnected(); - idMethodMap.clear(); + RPC::Connection::on_disconnected(); emit lostConnection(this); /// re-emits as EXClient * signal (different method in C++) } -/* static */ -EXResponse EXResponse::fromJson(const QString &json) -{ - const auto m = Util::Json::parseString(json).toMap(); - const auto jsonrpc = m.value("jsonrpc", "").toString(); - if (jsonrpc != "2.0") - throw BadServerReply(QString("Unexpected or missing jsonrpc version: \"%1\"").arg(jsonrpc)); - const qint64 id = m.value("id", -1).toLongLong(); - QString method = m.value("method", "").toString(); - if (id < 0 && method.isEmpty()) - throw BadServerReply("Bad server reply, missing required id field in JSON"); - QVariantMap err = m.value("error", QVariantMap()).toMap(); - if (!err.isEmpty()) { - // error reply - const int code = err.value("code", 123456789).toInt(); - const QString message = err.value("message").toString(); - if (code == 123456789 || message.isEmpty()) - throw BadServerReply("Bad server reply, error field in JSON is not of the expected format"); - return EXResponse{ - jsonrpc, id, method, QVariant(), code, message - }; - } - QVariant result = m.value("result", QVariant()); - if (result.isNull()) { - result = m.value("params", QVariant()); - } - - return EXResponse{ - jsonrpc, - id, - method, - result - }; -} - -QString -EXResponse::toString() const -{ - return QString("jsonrpc: %1 ; id: %2 ; method: %3 ; result: %4 ; error code: %5 ; error message: %6") - .arg(jsonRpcVersion).arg(id).arg(method).arg(result.isNull() ? "(null)" : Util::Json::toString(result, true)) - .arg(errorCode).arg(errorMessage); -} - -void EXResponse::validate() -{ - if (!errorMessage.isEmpty()) - return; - if (method == "server.version" && !result.isNull()) { - QVariantList l = result.toList(); - if (l.count() < 2 || l[0].toString().isNull() || l[1].toString().isNull()) - throw BadServerReply(QString("%1 expected string list of size 2").arg(method)); - return; // ok - } else if (method == "blockchain.headers.subscribe" && !result.isNull()) { - QVariantMap m = result.toMap(); - QVariantList l = result.toList(); - if (m.isEmpty() && !l.isEmpty()) { - // spontaneous "subscribe" callbacks pass a list containing a dict rather than a straight up dict, - // so mogrify ourselves to always contain the dict - m = l.last().toMap(); - result = m; // save back result as a map rather than a list - } - if (m.isEmpty() || m.count() < 2 || m.value("height", -1).toInt() < 0 || m.value("hex", "").toString().isEmpty()) { - throw BadServerReply(QString("%1 expected map with 'height' and 'hex'").arg(method)); - } - return; // ok - } else if (method == "server.ping") { - // always accept - return; - } - throw BadServerReply(QString("Unexpected method \"%1\", and/or incomplete/missing results").arg(method)); -} - -void EXClient::on_readyRead() -{ - Debug() << __FUNCTION__; - try { - while (socket->canReadLine()) { - auto data = socket->readLine(); - nReceived += data.length(); - auto line = data.trimmed(); - Debug() << "Got: " << line; - // testing - //mgr->testCheckMethod(line); - // /testing - auto resp = EXResponse::fromJson(line); - auto meth = resp.id > 0 ? idMethodMap.take(resp.id) : resp.method; - if (meth.isEmpty()) { - throw BadServerReply(QString("Unexpected/unknown message id (%1) in server reply").arg(resp.id)); - } - resp.method = meth; - resp.validate(); // may throw, may modify resp - Debug() << "Parsed response: " << resp.toString(); - lastGood = Util::getTime(); - emit gotResponse(this, resp); - } - if (socket->bytesAvailable() > MAX_BUFFER) { - // bad server.. sending us garbage data not containing newlines. Kill connection. - throw BadServerReply(QString("Server has sent us more than %1 bytes without a newline! Bad server?").arg(MAX_BUFFER)); - } - } catch (const Exception &e) { - Error() << "Error reading/parsing response: " << e.what(); - disconnect(); - status = Bad; - } -} - - -/* static */ -QByteArray EXClient::makeRequestData(qint64 id, const QString &method, const QVariantList ¶ms) -{ - QVariantMap m; - m["id"] = id; - m["method"] = method; - m["params"] = params; - try { - static const QChar nl(012); - return QString("%1%2").arg(Util::Json::toString(m, true)).arg(nl).toUtf8(); - } catch (const Util::Json::Error &e) { - Error() << __FUNCTION__ << ": " << e.what(); - } - return QByteArray(); -} diff --git a/EXClient.h b/EXClient.h index 4ad22f2..93fde09 100644 --- a/EXClient.h +++ b/EXClient.h @@ -11,29 +11,11 @@ #include "Common.h" #include "AbstractConnection.h" #include "Mixins.h" - -struct EXResponse -{ - static EXResponse fromJson(const QString &json); ///< may throw Exception - - void validate(); ///< checks the QVariant is the expected format for each method. throws BadServerReply if it's not - - QString toString() const; - - QString jsonRpcVersion; - qint64 id; - QString method; - QVariant result; // 'params' also gets put here - - int errorCode = 0; - QString errorMessage = ""; -}; - -Q_DECLARE_METATYPE(EXResponse); +#include "RPC.h" class EXMgr; -class EXClient : public AbstractConnection, protected ThreadObjectMixin +class EXClient : public RPC::Connection, protected ThreadObjectMixin { Q_OBJECT public: @@ -56,18 +38,10 @@ public: bool isGood() const override; signals: - void gotResponse(EXClient *, EXResponse); void newConnection(EXClient *); - void lostConnection(EXClient *); ///< overrides lostConnection(AbstractClient *) by dynamic_casting it down and re-emitting - /// call (emit) this to send a requesst to the server - void sendRequest(qint64 reqid, const QString &method, const QVariantList & params = QVariantList()); + void lostConnection(EXClient *); ///< overrides lostConnection(AbstractConnection *) by dynamic_casting it down and re-emitting + void gotMessage(EXClient *, const RPC::Message &); ///< overrides gotMessage(RPC::Connection *..) by re-emitting with narrowed type. -protected slots: - /// Actual implentation that prepares the request. Is connected to sendRequest() above. Runs in this object's thread. - void _sendRequest(qint64 reqid, const QString &method, const QVariantList & params = QVariantList()); - - /// called from socket connection - void on_readyRead() override; protected: friend class EXMgr; @@ -83,10 +57,6 @@ protected: private: EXMgr *mgr = nullptr; - QMap idMethodMap; - - /// returns utf-8 encoded JSON data for a request - static QByteArray makeRequestData(qint64 id, const QString &method, const QVariantList & params = QVariantList()); void on_started() override; void on_finished() override; diff --git a/EXMgr.cpp b/EXMgr.cpp index 1ca3655..9a9a913 100644 --- a/EXMgr.cpp +++ b/EXMgr.cpp @@ -7,11 +7,6 @@ EXMgr::EXMgr(const QString & serversFile, QObject *parent) : Mgr(parent), serversFile(serversFile) { - static bool initted_meta = false; - if (!initted_meta) { - qRegisterMetaType(); - initted_meta = true; - } } EXMgr::~EXMgr() @@ -37,7 +32,7 @@ void EXMgr::cleanup() } clients.clear(); clientsById.clear(); - rpcMethods.clear(); + _rpcMethods.clear(); } void EXMgr::loadServers() @@ -58,7 +53,7 @@ void EXMgr::loadServers() clientsById[client->id] = client; connect(client, &EXClient::newConnection, this, &EXMgr::onNewConnection); connect(client, &EXClient::lostConnection, this, &EXMgr::onLostConnection); - connect(client, &EXClient::gotResponse, this, &EXMgr::onResponse); + connect(client, &EXClient::gotMessage, this, &EXMgr::onMessage); client->start(); } else { Warning() << "Bad server entry: " << host; @@ -78,9 +73,6 @@ void EXMgr::onNewConnection(EXClient *client) Debug () << "New connection for " << client->host; emit client->sendRequest(newId(), "server.version", QVariantList({QString("%1/%2").arg(APPNAME).arg(VERSION), QString("1.4")})); emit client->sendRequest(newId(), "blockchain.headers.subscribe"); - // testing - //testComposeRequest(123, "blockchain.headers.subscribe"); - //testComposeRequest(123, "server.version", QVariantList({QString("%1/%2").arg(APPNAME).arg(VERSION), QString("1.4")})); } void EXMgr::onLostConnection(EXClient *client) @@ -90,17 +82,25 @@ void EXMgr::onLostConnection(EXClient *client) client->info.clear(); } -void EXMgr::onResponse(EXClient *client, EXResponse r) +void EXMgr::onMessage(EXClient *client, const RPC::Message &m) { - Debug() << "(" << client->host << ") Got response in mgr to " << r.method; - if (r.method == "server.version") { - client->info.serverVersion.first = r.result.toList()[0].toString(); - client->info.serverVersion.second = r.result.toList()[1].toString(); - Debug() << "Got server version: " << client->info.serverVersion.first << " / " << client->info.serverVersion.second; - } else if (r.method == "blockchain.headers.subscribe") { - int ht = r.result.toMap().value("height", 0).toInt(); - QString hdr = r.result.toMap().value("hex", "").toString(); - if (ht > 0) { + Debug() << "(" << client->host << ") Got message in mgr, method: " << m.method; + if (m.method == "server.version") { + QVariantList l = m.data.toList(); + if (l.size() == 2) { + client->info.serverVersion.first = l[0].toString(); + client->info.serverVersion.second = l[1].toString(); + Debug() << "Got server version: " << client->info.serverVersion.first << " / " << client->info.serverVersion.second; + } else { + Error() << "Bad server version reply! Schema should have handled this. FIXME! Json: " << m.toJsonString(); + } + } else if (m.method == "blockchain.headers.subscribe") { + // list of dicts.. or a simple value.. handle either. TODO: make this a more general mechanism. + const QVariantList list ( m.data.toList() ); + const QVariantMap map ( list.isEmpty() ? m.data.toMap() : list.back().toMap() ); + int ht = map.value("height", 0).toInt(); + QString hdr = map.value("hex", "").toString(); + if (ht > 0 && !hdr.isEmpty()) { if (ht > height.height) { height.height = ht; height.ts = Util::getTime(); @@ -111,12 +111,15 @@ void EXMgr::onResponse(EXClient *client, EXResponse r) height.seenBy.insert(client->id); client->info.height = ht; client->info.header = hdr; + } else { + Error() << "Bad server headers reply! Schema should have handled this. FIXME! Json: " << m.toJsonString(); } Debug() << "Got header subscribe: " << client->info.height << " / " << client->info.header << " (count for height = " << height.seenBy.count() << ")"; - } else if (r.method == "server.ping") { - // ignore; timestamps updated in EXClient + } else if (m.method == "server.ping") { + // ignore; timestamps updated in EXClient and RPC::Connection + //Debug() << "server.ping reply... yay"; } else { - Error() << "Unknown method \"" << r.method << "\" from " << client->host; + Error() << "Unknown method \"" << m.method << "\" from " << client->host; } } @@ -180,6 +183,35 @@ EXClient * EXMgr::pick() return nullptr; } +void EXMgr::initRPCMethods() +{ + QString m, d; + m = "blockchain.headers.subscribe"; + d = "{\"hex\" : \"somestring\", \"height\" : 1}"; + _rpcMethods.insert(m, QSharedPointer(new RPC::Method( + m, + RPC::schemaMethod + QString(" { \"method\" : \"%1!\", \"params\" : [%2] }").arg(m).arg(d), // in schema (asynch from server -> us) + RPC::schemaResult + QString(" { \"result\" : %1}").arg(d), // result schema (synch. server -> us) + RPC::schemaMethod + QString(" { \"method\" : \"%1!\", \"params\" : [\"=0\"] }").arg(m) // out schema (req. us -> server) + ))); + + m = "server.version"; + _rpcMethods.insert(m, QSharedPointer(new RPC::Method( + m, + RPC::Schema(), // in schema (asynch from server -> us) -- DISABLED for server.version + RPC::schemaResult + QString(" { \"result\" : [\"=2\"] }"), // result schema (synch. server -> us) -- enforce must have 2 string args + RPC::schemaMethod + QString(" { \"method\" : \"%1!\", \"params\" : [\"=2\"] }").arg(m) // out schema (req. us -> server) -- enforce must have 2 string args + ))); + + m = "server.ping"; + _rpcMethods.insert(m, QSharedPointer(new RPC::Method( + m, + RPC::Schema(), // in schema (asynch from server -> us) -- DISABLED for server.ping + RPC::schemaResult + QString(" { \"result\" : null }"), // result schema -- 'result' arg should be there and be null. + RPC::schemaMethodNoParams // out schema, ping to server takes no args + ))); +} + void EXMgr::pickTest() { for (int i = 0; i < 100; ++i) { @@ -193,64 +225,3 @@ void EXMgr::pickTest() if (qApp) qApp->processEvents(); } } - -void EXMgr::initRPCMethods() -{ - QString m, d; - m = "blockchain.headers.subscribe"; - d = "{\"hex\" : \"somestring\", \"height\" : 1}"; - rpcMethods.insert(m, QSharedPointer(new RPC::Method( - m, - RPC::schemaMethod + QString(" { \"method\" : \"%1!\", \"params\" : [%2] }").arg(m).arg(d), // in schema (asynch from server -> us) - RPC::schemaResult + QString(" { \"result\" : %1}").arg(d), // result schema (synch. server -> us) - RPC::schemaMethod + QString(" { \"method\" : \"%1!\", \"params\" : [\"=0\"] }").arg(m) // out schema (req. us -> server) - ))); - - m = "server.version"; - rpcMethods.insert(m, QSharedPointer(new RPC::Method( - m, - RPC::Schema(), // in schema (asynch from server -> us) -- DISABLED for server.version - RPC::schemaResult + QString(" { \"result\" : [\"=2\"] }"), // result schema (synch. server -> us) - RPC::schemaMethod + QString(" { \"method\" : \"%1!\", \"params\" : [\"=2\"] }").arg(m) // out schema (req. us -> server) - ))); -} - -void EXMgr::testCheckMethod(const QString &json) const -{ - try { - const auto m = Util::Json::parseString(json, true).toMap(); - for (auto it = rpcMethods.cbegin(); it != rpcMethods.cend(); ++it) { - QList schemas({&(it.value()->inSchema), &(it.value()->resultSchema)}); - for (auto s : schemas) { - if (!s->isValid()) - // disabled schema - continue; - QString err; - if (auto res = s->match(m, &err); !res.isEmpty()) { - Debug() << "---> testCheckMethod on " << it.key() << ": parsed -> " << Util::Json::toString(res, true); - } else { - Debug() << "---> testCheckMethod on " << it.key() << ": failed -> " << err; - } - } - } - } catch (const Exception &e) { - Warning() << "testCheckMethod: " << e.what() << " (" << json << ")"; - } -} - -QVariantMap EXMgr::testComposeRequest(qint64 id, const QString &method, const QVariantList ¶ms) const -{ - QVariantMap ret; - if (auto it = rpcMethods.find(method); it != rpcMethods.end()) { - ret = it.value()->outSchema.toStrippedMap(); - ret["id"] = id; - ret["params"] = params; - if (QString err; ! (ret = it.value()->outSchema.match(ret, &err)).isEmpty()) { - Debug() << method << " ---> compose --> matched, json = " << Util::Json::toString(ret, true); - return ret; - } else { - Debug() << method << " ---> compose error: " << err; - } - } - return ret; -} diff --git a/EXMgr.h b/EXMgr.h index baa86b0..584b2e2 100644 --- a/EXMgr.h +++ b/EXMgr.h @@ -30,9 +30,7 @@ public: EXClient *pick(); - /// here for testing, thread-safe - void testCheckMethod(const QString &json) const; - QVariantMap testComposeRequest(qint64 id, const QString &method, const QVariantList ¶ms = QVariantList()) const; + const RPC::MethodMap & rpcMethods() const { return _rpcMethods; } signals: @@ -41,7 +39,7 @@ public slots: protected slots: void onNewConnection(EXClient *); void onLostConnection(EXClient *); - void onResponse(EXClient *, EXResponse); + void onMessage(EXClient *, const RPC::Message &); private slots: void checkClients(); @@ -68,7 +66,7 @@ private: void pickTest(); - RPC::MethodMap rpcMethods; + RPC::MethodMap _rpcMethods; }; #endif // ECMGR_H diff --git a/RPC.cpp b/RPC.cpp index 7b27555..aa7172f 100644 --- a/RPC.cpp +++ b/RPC.cpp @@ -413,15 +413,15 @@ namespace RPC { ret.data = params; map["params"] = params; ret.jsonRpcVersion = map.value("jsonrpc").toString(); -#ifdef QT_DEBUG - QString err; - ret.jsonData = ret.schema.match(map, &err); - if (ret.jsonData.isEmpty()) { - Error() << __FUNCTION__ << " schema verify failure: " << err << "; FIXME!"; - } -#else +//#ifdef QT_DEBUG +// QString err; +// ret.jsonData = ret.schema.match(map, &err); +// if (ret.jsonData.isEmpty()) { +// Error() << __FUNCTION__ << " schema verify failure: " << err << "; FIXME!"; +// } +//#else ret.jsonData = map; -#endif +//#endif return ret; } @@ -430,6 +430,8 @@ namespace RPC { { static bool initted_meta = false; if (!initted_meta) { + // finish registering RPC::Message metatype so that signals/slots work. This needs to only happen + // once in main thread at app init. qRegisterMetaType(); initted_meta = true; } @@ -440,12 +442,13 @@ namespace RPC { void Connection::on_connected() { AbstractConnection::on_connected(); - connectedConns.push_back(connect(this, &Connection::sendRequest, this, &Connection::_sendRequest)); // connection will be auto-disconnected on socket disconnect + // connection will be auto-disconnected on socket disconnect + connectedConns.push_back(connect(this, &Connection::sendRequest, this, &Connection::_sendRequest)); } void Connection::on_disconnected() { - AbstractConnection::on_disconnected(); + AbstractConnection::on_disconnected(); // will auto-disconnect all QMetaObject::Connections appearing in connectedConns idMethodMap.clear(); } @@ -468,11 +471,12 @@ namespace RPC { while (idMethodMap.size() > 20000) { // prevent memory leaks in case of misbehaving server idMethodMap.erase(idMethodMap.begin()); } - idMethodMap[id] = method; + idMethodMap[reqid] = method; // remember method sent out to associate it back. auto data = json.toUtf8(); Debug() << "Sending json: " << data; - emit send(data); // ends up calling do_write immediately (which is connected to send) + // below send() ends up calling do_write immediately (which is connected to send) + emit send( data + "\n" /* "\n" <-- is crucial! (Protocol is linefeed-based) */); } void Connection::on_readyRead() @@ -515,8 +519,8 @@ namespace RPC { message = Message::fromJsonData(jsonData, schema); message.method = mptr->method; // write to the method var again in case it was a result with no method name in the json } - lastGood = Util::getTime(); - Debug() << "Re-parsed message: " << message.toJsonString(); + lastGood = Util::getTime(); // update "lastGood" as this is used to determine if stale or not. + //Debug() << "Re-parsed message: " << message.toJsonString(); emit gotMessage(this, message); } if (socket->bytesAvailable() > MAX_BUFFER) { @@ -525,7 +529,7 @@ namespace RPC { } } catch (const std::exception &e) { Error() << "Error reading/parsing data coming in: " << e.what(); - disconnect(); + do_disconnect(); status = Bad; } } diff --git a/RPC.h b/RPC.h index e8bedb0..9135cfb 100644 --- a/RPC.h +++ b/RPC.h @@ -130,6 +130,32 @@ namespace RPC { typedef QMap > MethodMap; + /// A concrete derived class of AbstractConnection implementing a JSON-RPC + /// based method<->result protocol similar to ElectrumX's protocol. This + /// class is client/server agnostic and it just operates in terms of JSON + /// RPC methods and results. It can be used for either a client or a + /// server. + /// + /// Methods invoked on the peer need an id, and this id is used to track + /// the reply back and associate it with the method that was invoked on + /// the peer (see idMethodMap instance var). + /// + /// We use this protocol on the client-facing side too to negotiate + /// shuffles, hence why this is abstracted out into a separate class. This + /// class is responsible for parsing the JSON and closing the connections + /// on malformed input that doesn't match the expected 'Schema'. This + /// class is configured for which methods it supports and what the various + /// method schemas are via the 'MethodMap' passed to it at construction. + /// + /// See EXMgr for an example class that constructs a MethodMap and passes + /// it down. + /// + /// Classes that manage rpc methods should register for the gotMessage() + /// signal and process incoming messages further. All incoming messages + /// are either errors (errorCode != 0) or have a valid message.method name. + /// + /// Both EXClient and TCPServer's 'Client' class derive from this. + /// class Connection : public AbstractConnection { Q_OBJECT @@ -146,24 +172,33 @@ namespace RPC { signals: /// call (emit) this to send a requesst to the server void sendRequest(qint64 reqid, const QString &method, const QVariantList & params = QVariantList()); - void gotMessage(Connection *, Message m); + /// this is emitted when a new message arrives that was successfully parsed and matches + /// a known method described in the 'methods' MethodMap. Unknown messages will result + /// in auto-disconnect. (TODO: Implement error JSON replies to peer as well as tolerance for some malformed + /// data up until a threshold is reached?) + void gotMessage(RPC::Connection *, const RPC::Message & m); protected slots: - /// Actual implentation that prepares the request. Is connected to sendRequest() above. Runs in thread. Eventually calls send() -> do_write() + /// Actual implentation that prepares the request. Is connected to sendRequest() above. Runs in this object's + /// thread context. Eventually calls send() -> do_write() (from superclass). virtual void _sendRequest(qint64 reqid, const QString &method, const QVariantList & params = QVariantList()); - /// parses RPC, implements pure virtual from base - void on_readyRead() override; protected: + /// parses RPC, implements pure virtual from super to handle line-based JSON. + void on_readyRead() override; /// chains to base, connects sendRequest signal to _sendRequest slot void on_connected() override; + /// Chains to base, clears idMethodMap void on_disconnected() override; - /// map of requests that were generated via _sendRequest to method names to build a more meaningful Message object. + /// map of requests that were generated via _sendRequest to method names to build a more meaningful Message + /// object (which has a .method defined even on 'result=' messages). It is an error to receive a result= + /// message from the peer with its id= parameter not having an entry in this map. QMap idMethodMap; }; } +/// So that Qt signal/slots work with this type. Q_DECLARE_METATYPE(RPC::Message); #endif // SHUFFLEUP_RPC_H diff --git a/TcpServer.cpp b/TcpServer.cpp index 84a823b..d81992b 100644 --- a/TcpServer.cpp +++ b/TcpServer.cpp @@ -119,7 +119,7 @@ void TcpServer::killClient(Client *client) return; Debug() << __FUNCTION__ << " (id: " << client->id << ")"; clientsById.remove(client->id); // ensure gone from map asap so future lookups fail - client->disconnect(); + client->do_disconnect(); } void TcpServer::killClient(qint64 id) { @@ -152,22 +152,22 @@ void Client::on_readyRead() lastGood = Util::getTime(); if (line == "exit") { send("sayonara\n"); - disconnect(true); + do_disconnect(true); } else send("Thanks fam.\n"); } if (socket->bytesAvailable() > MAX_BUFFER) { // bad server.. sending us garbage data not containing newlines. Kill connection. Error() << QString("client has sent us more than %1 bytes without a newline! Bad client? (id: %2)").arg(MAX_BUFFER).arg(id); - disconnect(); + do_disconnect(); status = Bad; } } -void Client::disconnect(bool graceful) +void Client::do_disconnect(bool graceful) { const bool wasConnected = socket ? socket->state() == QAbstractSocket::ConnectedState : false; - AbstractConnection::disconnect(graceful); // if 'graceful' *AND* was connected, a disconnected state will be entered later at which point we will delete socket. + AbstractConnection::do_disconnect(graceful); // if 'graceful' *AND* was connected, a disconnected state will be entered later at which point we will delete socket. if (socket && (!graceful || !wasConnected)) /// delete the socket if we weren't connected or if !graceful. /// If graceful && connected, then a disconnect signal will be sent later and then we will diff --git a/TcpServer.h b/TcpServer.h index 7156b17..2666e7e 100644 --- a/TcpServer.h +++ b/TcpServer.h @@ -73,12 +73,11 @@ public: explicit Client(qint64 id, TcpServer *srv, QTcpSocket *sock); ~Client() override; -protected slots: +protected: void on_readyRead() override; -protected: void do_ping() override; - void disconnect(bool graceful = false) override; + void do_disconnect(bool graceful = false) override; TcpServer *srv; friend class TcpServer;