diff --git a/lib/remote/apilistener.cpp b/lib/remote/apilistener.cpp index cf4de5952..75a676143 100644 --- a/lib/remote/apilistener.cpp +++ b/lib/remote/apilistener.cpp @@ -1277,7 +1277,7 @@ void ApiListener::ReplayLog(const JsonRpcConnection::Ptr& client) } try { - client->SendMessage(JsonDecode(pmessage->Get("message"))); + client->SendRawMessage(pmessage->Get("message")); count++; } catch (const std::exception& ex) { Log(LogWarning, "ApiListener") diff --git a/lib/remote/jsonrpc.cpp b/lib/remote/jsonrpc.cpp index 5f4b6ae91..03f3c7d0e 100644 --- a/lib/remote/jsonrpc.cpp +++ b/lib/remote/jsonrpc.cpp @@ -68,8 +68,18 @@ size_t JsonRpc::SendMessage(const Stream::Ptr& stream, const Dictionary::Ptr& me */ size_t JsonRpc::SendMessage(const std::shared_ptr& stream, const Dictionary::Ptr& message, boost::asio::yield_context yc) { - String json = JsonEncode(message); + return JsonRpc::SendRawMessage(stream, JsonEncode(message), yc); +} +/** + * Sends a message to the connected peer and returns the bytes sent. + * + * @param message The message. + * + * @return The amount of bytes sent. + */ +size_t JsonRpc::SendRawMessage(const std::shared_ptr& stream, const String& json, boost::asio::yield_context yc) +{ #ifdef I2_DEBUG if (GetDebugJsonRpcCached()) std::cerr << ConsoleColorTag(Console_ForegroundBlue) << ">> " << json << ConsoleColorTag(Console_Normal) << "\n"; diff --git a/lib/remote/jsonrpc.hpp b/lib/remote/jsonrpc.hpp index 137d42a3b..faf9c07e8 100644 --- a/lib/remote/jsonrpc.hpp +++ b/lib/remote/jsonrpc.hpp @@ -23,6 +23,7 @@ class JsonRpc public: static size_t SendMessage(const Stream::Ptr& stream, const Dictionary::Ptr& message); static size_t SendMessage(const std::shared_ptr& stream, const Dictionary::Ptr& message, boost::asio::yield_context yc); + static size_t SendRawMessage(const std::shared_ptr& stream, const String& json, boost::asio::yield_context yc); static StreamReadStatus ReadMessage(const Stream::Ptr& stream, String *message, StreamReadContext& src, bool may_wait = false, ssize_t maxMessageLength = -1); static String ReadMessage(const std::shared_ptr& stream, boost::asio::yield_context yc, ssize_t maxMessageLength = -1); static Dictionary::Ptr DecodeMessage(const String& message); diff --git a/lib/remote/jsonrpcconnection.cpp b/lib/remote/jsonrpcconnection.cpp index 16066a6ae..e54a998d0 100644 --- a/lib/remote/jsonrpcconnection.cpp +++ b/lib/remote/jsonrpcconnection.cpp @@ -6,6 +6,7 @@ #include "remote/jsonrpc.hpp" #include "base/configtype.hpp" #include "base/io-engine.hpp" +#include "base/json.hpp" #include "base/objectlock.hpp" #include "base/utility.hpp" #include "base/logger.hpp" @@ -106,7 +107,7 @@ void JsonRpcConnection::WriteOutgoingMessages(boost::asio::yield_context yc) if (!queue.empty()) { try { for (auto& message : queue) { - size_t bytesSent = JsonRpc::SendMessage(m_Stream, message, yc); + size_t bytesSent = JsonRpc::SendRawMessage(m_Stream, message, yc); if (m_Endpoint) { m_Endpoint->AddMessageSent(bytesSent); @@ -163,9 +164,17 @@ void JsonRpcConnection::SendMessage(const Dictionary::Ptr& message) m_IoStrand.post([this, message]() { SendMessageInternal(message); }); } +void JsonRpcConnection::SendRawMessage(const String& message) +{ + m_IoStrand.post([this, message]() { + m_OutgoingMessagesQueue.emplace_back(message); + m_OutgoingMessagesQueued.Set(); + }); +} + void JsonRpcConnection::SendMessageInternal(const Dictionary::Ptr& message) { - m_OutgoingMessagesQueue.emplace_back(message); + m_OutgoingMessagesQueue.emplace_back(JsonEncode(message)); m_OutgoingMessagesQueued.Set(); } diff --git a/lib/remote/jsonrpcconnection.hpp b/lib/remote/jsonrpcconnection.hpp index 8f48fc4cd..994dd7368 100644 --- a/lib/remote/jsonrpcconnection.hpp +++ b/lib/remote/jsonrpcconnection.hpp @@ -55,6 +55,7 @@ public: void Disconnect(); void SendMessage(const Dictionary::Ptr& request); + void SendRawMessage(const String& request); static Value HeartbeatAPIHandler(const intrusive_ptr& origin, const Dictionary::Ptr& params); @@ -72,7 +73,7 @@ private: double m_Seen; double m_NextHeartbeat; boost::asio::io_service::strand m_IoStrand; - std::vector m_OutgoingMessagesQueue; + std::vector m_OutgoingMessagesQueue; AsioConditionVariable m_OutgoingMessagesQueued; AsioConditionVariable m_WriterDone; bool m_ShuttingDown;