use SendBufferQueue for queued messages from I2P

libi2pd/Streaming.cpp

@@ -21,8 +21,16 @@ namespace stream
 {
 	void SendBufferQueue::Add (const uint8_t * buf, size_t len, SendHandler handler)
 	{
-		m_Buffers.push_back (std::make_shared<SendBuffer>(buf, len, handler));
+		Add (std::make_shared<SendBuffer>(buf, len, handler));
-		m_Size += len;
+	}
+
+	void SendBufferQueue::Add (std::shared_ptr<SendBuffer> buf)
+	{
+		if (buf)
+		{	
+			m_Buffers.push_back (buf);
+			m_Size += buf->len;
+		}	
 	}	
 	
 	size_t SendBufferQueue::Get (uint8_t * buf, size_t len)

libi2pd/Streaming.h

@@ -111,6 +111,11 @@ namespace stream
 			buf = new uint8_t[len];
 			memcpy (buf, b, len);
 		}
+		SendBuffer (size_t l): // creat empty buffer
+			len(l), offset (0)	
+		{
+			buf = new uint8_t[len];
+		}	
 		~SendBuffer ()
 		{
 			delete[] buf;
@@ -129,6 +134,7 @@ namespace stream
 			~SendBufferQueue () { CleanUp (); };
 
 			void Add (const uint8_t * buf, size_t len, SendHandler handler);
+			void Add (std::shared_ptr<SendBuffer> buf);
 			size_t Get (uint8_t * buf, size_t len);
 			size_t GetSize () const { return m_Size; };
 			bool IsEmpty () const { return m_Buffers.empty (); };

libi2pd_client/I2CP.cpp

@@ -360,13 +360,8 @@ namespace client
 			m_Socket->close ();
 			m_Socket = nullptr;
 		}
-		if (m_SendQueue)
+		if (!m_SendQueue.IsEmpty ())
-		{
+			m_SendQueue.CleanUp ();
-			for (auto& it: *m_SendQueue)
-				delete[] boost::asio::buffer_cast<const uint8_t *>(it);
-			m_SendQueue->clear ();
-			m_SendQueue = nullptr;
-		}	
 		if (m_SessionID != 0xFFFF)
 		{	
 			m_Owner.RemoveSession (GetSessionID ());
@@ -383,75 +378,56 @@ namespace client
 			LogPrint (eLogError, "I2CP: Message to send is too long ", l);
 			return;
 		}	
-		uint8_t * buf = m_IsSending ? new uint8_t[l] : m_SendBuffer;
+		auto sendBuf = m_IsSending ? std::make_shared<i2p::stream::SendBuffer> (l) : nullptr;
+		uint8_t * buf = sendBuf ? sendBuf->buf : m_SendBuffer;
 		htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len);
 		buf[I2CP_HEADER_TYPE_OFFSET] = type;
 		memcpy (buf + I2CP_HEADER_SIZE, payload, len);
-		SendBuffer (buf, l);
+		if (sendBuf)
-	}
-
-	void I2CPSession::HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred)
 		{	
-		if (ecode)
+			if (m_SendQueue.GetSize () < I2CP_MAX_SEND_QUEUE_SIZE)
+				m_SendQueue.Add (sendBuf);
+			else	
 			{	
-			if (ecode != boost::asio::error::operation_aborted)
+				LogPrint (eLogWarning, "I2CP: send queue size exceeds ", I2CP_MAX_SEND_QUEUE_SIZE);	
-				Terminate ();
+				return;	
+			}		
 		}		
-		else if (m_SendQueue)
+		else
 		{
 			auto socket = m_Socket;
 			if (socket)
 			{	
-				auto queue = m_SendQueue;
+				m_IsSending = true;
-				m_SendQueue = nullptr;
+				boost::asio::async_write (*socket, boost::asio::buffer (m_SendBuffer, l), 
-				boost::asio::async_write (*socket, *queue, boost::asio::transfer_all (),
+					boost::asio::transfer_all (), std::bind(&I2CPSession::HandleI2CPMessageSent, 
-					std::bind(&I2CPSession::HandleI2CPMessageSentQueue, shared_from_this (),
+					shared_from_this (), std::placeholders::_1, std::placeholders::_2));
-					std::placeholders::_1, std::placeholders::_2, queue));
 			}	
 		}	
-		else
-			m_IsSending = false;
 	}
 
-	void I2CPSession::HandleI2CPMessageSentQueue (const boost::system::error_code& ecode, std::size_t bytes_transferred, SendQueue queue)
+	void I2CPSession::HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred)
 	{
-		for (auto& it: *queue)
+		if (ecode)
-			delete[] boost::asio::buffer_cast<const uint8_t *>(it);
+		{	
-		queue->clear ();
+			if (ecode != boost::asio::error::operation_aborted)
-		
+				Terminate ();
-		HandleI2CPMessageSent (ecode, bytes_transferred);
 		}
-
+		else if (!m_SendQueue.IsEmpty ())
-	void I2CPSession::SendBuffer (uint8_t * buf, size_t len)
 		{
 			auto socket = m_Socket;
 			if (socket)
 			{	
-			if (m_IsSending)
+				auto len = m_SendQueue.Get (m_SendBuffer, I2CP_MAX_MESSAGE_LENGTH);
-			{
+				boost::asio::async_write (*socket, boost::asio::buffer (m_SendBuffer, len), 
-				auto sendQueue = m_SendQueue;
+				    boost::asio::transfer_all (),std::bind(&I2CPSession::HandleI2CPMessageSent, 
-				if (!sendQueue)
+				    shared_from_this (), std::placeholders::_1, std::placeholders::_2));
-				{	
-					sendQueue = std::make_shared<SendQueue::element_type> ();
-					m_SendQueue = sendQueue;
-				}	
-				else if (sendQueue->size () > I2CP_MAX_SEND_QUEUE_SIZE)
-				{
-					LogPrint (eLogError, "I2CP: Queue size exceeds ", I2CP_MAX_SEND_QUEUE_SIZE);
-					return;
-				}
-				sendQueue->push_back ({buf, len});
 			}	
 			else
-			{
+				m_IsSending = false;
-				m_IsSending = true;
-				boost::asio::async_write (*socket, boost::asio::buffer (buf, len), boost::asio::transfer_all (),
-					std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (),
-						std::placeholders::_1, std::placeholders::_2));
-			}	
 		}
 		else
-			LogPrint (eLogError, "I2CP: Can't write to the socket");
+			m_IsSending = false;
 	}
 		
 	std::string I2CPSession::ExtractString (const uint8_t * buf, size_t len)
@@ -885,14 +861,35 @@ namespace client
 			LogPrint (eLogError, "I2CP: Message to send is too long ", l);
 			return;
 		}			
-		uint8_t * buf = m_IsSending ? new uint8_t[l] : m_SendBuffer;
+		auto sendBuf = m_IsSending ? std::make_shared<i2p::stream::SendBuffer> (l) : nullptr;
+		uint8_t * buf = sendBuf ? sendBuf->buf : m_SendBuffer;
 		htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len + 10);
 		buf[I2CP_HEADER_TYPE_OFFSET] = I2CP_MESSAGE_PAYLOAD_MESSAGE;
 		htobe16buf (buf + I2CP_HEADER_SIZE, m_SessionID);
 		htobe32buf (buf + I2CP_HEADER_SIZE + 2, m_MessageID++);
 		htobe32buf (buf + I2CP_HEADER_SIZE + 6, len);
 		memcpy (buf + I2CP_HEADER_SIZE + 10, payload, len);
-		SendBuffer (buf, l);
+		if (sendBuf)
+		{	
+			if (m_SendQueue.GetSize () < I2CP_MAX_SEND_QUEUE_SIZE)
+				m_SendQueue.Add (sendBuf);
+			else	
+			{	
+				LogPrint (eLogWarning, "I2CP: send queue size exceeds ", I2CP_MAX_SEND_QUEUE_SIZE);	
+				return;	
+			}		
+		}		
+		else
+		{
+			auto socket = m_Socket;
+			if (socket)
+			{	
+				m_IsSending = true;
+				boost::asio::async_write (*socket, boost::asio::buffer (m_SendBuffer, l), 
+					boost::asio::transfer_all (), std::bind(&I2CPSession::HandleI2CPMessageSent, 
+					shared_from_this (), std::placeholders::_1, std::placeholders::_2));
+			}	
+		}	
 	}
 
 	I2CPServer::I2CPServer (const std::string& interface, int port, bool isSingleThread):

libi2pd_client/I2CP.h

@@ -17,6 +17,7 @@
 #include <boost/asio.hpp>
 #include "util.h"
 #include "Destination.h"
+#include "Streaming.h"
 
 namespace i2p
 {
@@ -25,7 +26,7 @@ namespace client
 	const uint8_t I2CP_PROTOCOL_BYTE = 0x2A;
 	const size_t I2CP_SESSION_BUFFER_SIZE = 4096;
 	const size_t I2CP_MAX_MESSAGE_LENGTH = 65535;
-	const size_t I2CP_MAX_SEND_QUEUE_SIZE = 256;
+	const size_t I2CP_MAX_SEND_QUEUE_SIZE = 1024*1024; // in bytes, 1M
 
 	const size_t I2CP_HEADER_LENGTH_OFFSET = 0;
 	const size_t I2CP_HEADER_TYPE_OFFSET = I2CP_HEADER_LENGTH_OFFSET + 4;
@@ -125,8 +126,6 @@ namespace client
 	class I2CPServer;
 	class I2CPSession: public std::enable_shared_from_this<I2CPSession>
 	{
-		typedef std::shared_ptr<std::vector<boost::asio::const_buffer> > SendQueue;	
-		
 		public:
 
 #ifdef ANDROID
@@ -171,14 +170,12 @@ namespace client
 			void HandleReceivedPayload (const boost::system::error_code& ecode, std::size_t bytes_transferred);
 			void HandleMessage ();
 			void Terminate ();
-			void SendBuffer (uint8_t * buf, size_t len);
 			
 			void HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred);
-			void HandleI2CPMessageSentQueue (const boost::system::error_code& ecode, std::size_t bytes_transferred, SendQueue queue);
+			
 			std::string ExtractString (const uint8_t * buf, size_t len);
 			size_t PutString (uint8_t * buf, size_t len, const std::string& str);
 			void ExtractMapping (const uint8_t * buf, size_t len, std::map<std::string, std::string>& mapping);
-
 			void SendSessionStatusMessage (uint8_t status);
 			void SendHostReplyMessage (uint32_t requestID, std::shared_ptr<const i2p::data::IdentityEx> identity);
 
@@ -197,7 +194,7 @@ namespace client
 			// to client
 			bool m_IsSending;
 			uint8_t m_SendBuffer[I2CP_MAX_MESSAGE_LENGTH];
-			SendQueue m_SendQueue; 
+			i2p::stream::SendBufferQueue m_SendQueue; 
 	};
 	typedef void (I2CPSession::*I2CPMessageHandler)(const uint8_t * buf, size_t len);