receive incoming SSU2 packets

libi2pd/SSU2.cpp

@@ -225,7 +225,10 @@ namespace transport
 						}	
 					}	
 					if (port)
+					{	
 						OpenSocket (port);
+						Receive ();
+					}	
 					else
 						LogPrint (eLogError, "SSU2: Can't start server because port not specified ");
 					break;
@@ -256,6 +259,36 @@ namespace transport
 		}
 	}
 
+	void SSU2Server::Receive ()
+	{
+		Packet * packet = m_PacketsPool.AcquireMt ();
+		m_Socket.async_receive_from (boost::asio::buffer (packet->buf, SSU2_MTU), packet->from,
+			std::bind (&SSU2Server::HandleReceivedFrom, this, std::placeholders::_1, std::placeholders::_2, packet));
+	}
+
+	void SSU2Server::HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred, Packet * packet)
+	{
+		if (!ecode)
+		{
+			packet->len = bytes_transferred;
+			ProcessNextPacket (packet->buf, packet->len, packet->from);
+			m_PacketsPool.ReleaseMt (packet);
+			Receive ();
+		}
+		else
+		{
+			m_PacketsPool.ReleaseMt (packet);
+			if (ecode != boost::asio::error::operation_aborted)
+			{
+				LogPrint (eLogError, "SSU2: Receive error: code ", ecode.value(), ": ", ecode.message ());
+				auto port = m_Socket.local_endpoint ().port ();
+				m_Socket.close ();
+				OpenSocket (port);
+				Receive ();
+			}
+		}
+	}
+		
 	void SSU2Server::AddSession (uint64_t connID, std::shared_ptr<SSU2Session> session)
 	{
 		m_Sessions.emplace (connID, session);

libi2pd/SSU2.h

@@ -24,6 +24,7 @@ namespace transport
 	const int SSU2_TERMINATION_TIMEOUT = 330; // 5.5 minutes
 	const size_t SSU2_SOCKET_RECEIVE_BUFFER_SIZE = 0x1FFFF; // 128K
 	const size_t SSU2_SOCKET_SEND_BUFFER_SIZE = 0x1FFFF; // 128K
+	const size_t SSU2_MTU = 1488;
 	
 	enum SSU2MessageType
 	{
@@ -78,6 +79,13 @@ namespace transport
 
 	class SSU2Server:  private i2p::util::RunnableServiceWithWork
 	{
+		struct Packet
+		{
+			uint8_t buf[SSU2_MTU]; 
+			size_t len;
+			boost::asio::ip::udp::endpoint from;
+		};	
+		
 		public:
 
 			SSU2Server ();
@@ -96,6 +104,8 @@ namespace transport
 		private:
 
 			void OpenSocket (int port);
+			void Receive ();
+			void HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred, Packet * packet);
 			void ProcessNextPacket (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint);
 			
 		private:
@@ -103,6 +113,7 @@ namespace transport
 			boost::asio::ip::udp::socket m_Socket;
 			std::unordered_map<uint64_t, std::shared_ptr<SSU2Session> > m_Sessions;
 			std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSU2Session> > m_PendingOutgoingSessions;
+			i2p::util::MemoryPoolMt<Packet> m_PacketsPool;
 	};	
 }
 }