SSU data receive batching

SSU.cpp

@@ -218,6 +218,7 @@ namespace transport
 			auto packet = it1;
 			if (!session || session->GetRemoteEndpoint () != packet->from) // we received packet for other session than previous
 			{
+				if (session) session->FlushData ();
 				auto it = m_Sessions.find (packet->from);
 				if (it != m_Sessions.end ())
 					session = it->second;
@@ -235,6 +236,7 @@ namespace transport
 			session->ProcessNextMessage (packet->buf, packet->len, packet->from);
 			delete packet;
 		}
+		if (session) session->FlushData ();
 	}
 
 	std::shared_ptr<SSUSession> SSUServer::FindSession (std::shared_ptr<const i2p::data::RouterInfo> router) const

SSUData.cpp

@@ -258,8 +258,11 @@ namespace transport
 				SendFragmentAck (msgID, fragmentNum);			
 			buf += fragmentSize;
 		}	
-		if (numFragments > 0)
-			m_Handler.Flush ();
+	}
+
+	void SSUData::FlushReceivedMessage ()
+	{
+		m_Handler.Flush ();
 	}	
 		
 	void SSUData::ProcessMessage (uint8_t * buf, size_t len)

SSUData.h

@@ -87,6 +87,7 @@ namespace transport
 			void Stop ();	
 			
 			void ProcessMessage (uint8_t * buf, size_t len);
+			void FlushReceivedMessage ();
 			void Send (i2p::I2NPMessage * msg);
 
 			void UpdatePacketSize (const i2p::data::IdentHash& remoteIdent);

SSUSession.cpp

@@ -17,7 +17,7 @@ namespace transport
 		std::shared_ptr<const i2p::data::RouterInfo> router, bool peerTest ): TransportSession (router), 
 		m_Server (server), m_RemoteEndpoint (remoteEndpoint), m_Timer (GetService ()), 
 		m_PeerTest (peerTest),m_State (eSessionStateUnknown), m_IsSessionKey (false), m_RelayTag (0),
-		m_NumSentBytes (0), m_NumReceivedBytes (0), m_Data (*this)
+		m_NumSentBytes (0), m_NumReceivedBytes (0), m_Data (*this), m_IsDataReceived (false)
 	{
 		m_CreationTime = i2p::util::GetSecondsSinceEpoch ();
 	}
@@ -866,8 +866,17 @@ namespace transport
 	void SSUSession::ProcessData (uint8_t * buf, size_t len)
 	{
 		m_Data.ProcessMessage (buf, len);
+		m_IsDataReceived = true;
 	}
 
+	void SSUSession::FlushData ()
+	{
+		if (m_IsDataReceived)
+		{	
+			m_Data.FlushReceivedMessage ();
+			m_IsDataReceived = false;
+		}		
+	}
 
 	void SSUSession::ProcessPeerTest (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint)
 	{

SSUSession.h

@@ -79,6 +79,8 @@ namespace transport
 			uint32_t GetRelayTag () const { return m_RelayTag; };	
 			uint32_t GetCreationTime () const { return m_CreationTime; };
 
+			void FlushData ();
+			
 		private:
 
 			boost::asio::io_service& GetService ();
@@ -139,6 +141,7 @@ namespace transport
 			size_t m_NumSentBytes, m_NumReceivedBytes;
 			uint32_t m_CreationTime; // seconds since epoch
 			SSUData m_Data;
+			bool m_IsDataReceived;
 	};