Count still queued messages per JsonRpcConnection

Author: Al2Klimov

lib/remote/jsonrpcconnection.cpp

@@ -154,6 +154,7 @@ void JsonRpcConnection::WriteOutgoingMessages(boost::asio::yield_context yc)
 					}
 
 					size_t bytesSent = JsonRpc::SendRawMessage(m_Stream, message, yc);
+					m_PendingOutgoingMessages--;
 
 					if (m_Endpoint) {
 						m_Endpoint->AddMessageSent(bytesSent);
@@ -230,6 +231,7 @@ void JsonRpcConnection::SendRawMessage(const String& message)
 
 		m_OutgoingMessagesQueue.emplace_back(message);
 		m_OutgoingMessagesQueued.Set();
+		m_PendingOutgoingMessages++;
 	});
 }
 
@@ -241,6 +243,7 @@ void JsonRpcConnection::SendMessageInternal(const Dictionary::Ptr& message)
 
 	m_OutgoingMessagesQueue.emplace_back(JsonEncode(message));
 	m_OutgoingMessagesQueued.Set();
+	m_PendingOutgoingMessages++;
 }
 
 void JsonRpcConnection::Disconnect()

lib/remote/jsonrpcconnection.hpp

@@ -54,6 +54,11 @@ class JsonRpcConnection final : public Object
 	Shared<AsioTlsStream>::Ptr GetStream() const;
 	ConnectionRole GetRole() const;
 
+	auto GetPendingOutgoingMessages() const noexcept
+	{
+		return m_PendingOutgoingMessages.load();
+	}
+
 	void Disconnect();
 
 	void SendMessage(const Dictionary::Ptr& request);
@@ -76,6 +81,7 @@ class JsonRpcConnection final : public Object
 	boost::asio::io_context::strand m_IoStrand;
 	std::vector<String> m_OutgoingMessagesQueue;
 	AsioConditionVariable m_OutgoingMessagesQueued;
+	Atomic<decltype(m_OutgoingMessagesQueue)::size_type, std::memory_order_relaxed> m_PendingOutgoingMessages {0};
 	AsioConditionVariable m_WriterDone;
 	Atomic<bool> m_ShuttingDown;
 	boost::asio::deadline_timer m_CheckLivenessTimer, m_HeartbeatTimer;