introduced ClientDestination

Destination.cpp

@@ -9,9 +9,9 @@
 
 namespace i2p
 {
-namespace stream
+namespace client
 {
-	StreamingDestination::StreamingDestination (bool isPublic, i2p::data::SigningKeyType sigType): 
+	ClientDestination::ClientDestination (bool isPublic, i2p::data::SigningKeyType sigType): 
 		m_IsRunning (false), m_Thread (nullptr), m_Service (nullptr), m_Work (nullptr), 
 		m_CurrentOutboundTunnel (nullptr), m_LeaseSet (nullptr), m_IsPublic (isPublic)
 	{		
@@ -23,7 +23,7 @@ namespace stream
 			LogPrint ("Local address ", GetIdentHash ().ToBase32 (), ".b32.i2p created");
 	}
 
-	StreamingDestination::StreamingDestination (const std::string& fullPath, bool isPublic):
+	ClientDestination::ClientDestination (const std::string& fullPath, bool isPublic):
 		m_IsRunning (false), m_Thread (nullptr), m_Service (nullptr), m_Work (nullptr),
 		m_CurrentOutboundTunnel (nullptr), m_LeaseSet (nullptr), m_IsPublic (isPublic) 
 	{
@@ -58,7 +58,7 @@ namespace stream
 		m_Pool = i2p::tunnel::tunnels.CreateTunnelPool (*this, 3); // 3-hops tunnel 
 	}
 
-	StreamingDestination::StreamingDestination (const i2p::data::PrivateKeys& keys, bool isPublic):
+	ClientDestination::ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic):
 		m_IsRunning (false), m_Thread (nullptr), m_Service (nullptr), m_Work (nullptr),	
 		m_Keys (keys), m_CurrentOutboundTunnel (nullptr), m_LeaseSet (nullptr), m_IsPublic (isPublic)
 	{
@@ -69,7 +69,7 @@ namespace stream
 			LogPrint ("Local address ", GetIdentHash ().ToBase32 (), ".b32.i2p created");
 	}
 
-	StreamingDestination::~StreamingDestination ()
+	ClientDestination::~ClientDestination ()
 	{
 		Stop ();
 		for (auto it: m_RemoteLeaseSets)
@@ -81,30 +81,23 @@ namespace stream
 		delete m_Service;
 	}	
 
-	void StreamingDestination::Run ()
+	void ClientDestination::Run ()
 	{
 		if (m_Service)
 			m_Service->run ();
 	}	
 
-	void StreamingDestination::Start ()
+	void ClientDestination::Start ()
 	{	
 		m_Service = new boost::asio::io_service;
 		m_Work = new boost::asio::io_service::work (*m_Service);
 		m_Pool->SetActive (true);
 		m_IsRunning = true;
-		m_Thread = new std::thread (std::bind (&StreamingDestination::Run, this));
+		m_Thread = new std::thread (std::bind (&ClientDestination::Run, this));
 	}
 		
-	void StreamingDestination::Stop ()
+	void ClientDestination::Stop ()
-	{	
-		ResetAcceptor ();
 	{	
-			std::unique_lock<std::mutex> l(m_StreamsMutex);
-			for (auto it: m_Streams)
-				delete it.second;	
-			m_Streams.clear ();
-		}	
 		if (m_Pool)
 			i2p::tunnel::tunnels.StopTunnelPool (m_Pool);
 		m_IsRunning = false;
@@ -120,102 +113,127 @@ namespace stream
 		delete m_Service; m_Service = nullptr;
 	}	
 
-	void StreamingDestination::SendTunnelDataMsgs (const std::vector<i2p::tunnel::TunnelMessageBlock>& msgs)
+	const i2p::data::LeaseSet * ClientDestination::FindLeaseSet (const i2p::data::IdentHash& ident)
 	{
-		m_CurrentOutboundTunnel = m_Pool->GetNextOutboundTunnel (m_CurrentOutboundTunnel);
+		auto it = m_RemoteLeaseSets.find (ident);
-		if (m_CurrentOutboundTunnel)
+		if (it != m_RemoteLeaseSets.end ())
-			m_CurrentOutboundTunnel->SendTunnelDataMsg (msgs);
+		{	
+			if (it->second->HasNonExpiredLeases ())
+				return it->second;
 			else
 			{
-			LogPrint ("No outbound tunnels in the pool");
+				LogPrint ("All leases of remote LeaseSet expired. Request it");
-			for (auto it: msgs)
+				i2p::data::netdb.RequestDestination (ident, true, m_Pool);
-				DeleteI2NPMessage (it.data);
 			}	
 		}	
+		else
+		{	
+			auto ls = i2p::data::netdb.FindLeaseSet (ident);
+			if (ls)
+			{
+				ls = new i2p::data::LeaseSet (*ls);
+				m_RemoteLeaseSets[ident] = ls;			
+				return ls;
+			}	
+		}
+		return nullptr;
+	}	
 
-	void StreamingDestination::HandleNextPacket (Packet * packet)
+	const i2p::data::LeaseSet * ClientDestination::GetLeaseSet ()
 	{
-		uint32_t sendStreamID = packet->GetSendStreamID ();
+		if (!m_Pool) return nullptr;
-		if (sendStreamID)
+		if (!m_LeaseSet)
+			UpdateLeaseSet ();
+		return m_LeaseSet;
+	}	
+
+	void ClientDestination::UpdateLeaseSet ()
 	{
-			auto it = m_Streams.find (sendStreamID);
+		auto newLeaseSet = new i2p::data::LeaseSet (*m_Pool);
-			if (it != m_Streams.end ())
+		if (!m_LeaseSet)
-				it->second->HandleNextPacket (packet);
+			m_LeaseSet = newLeaseSet;
 		else
 		{	
-				LogPrint ("Unknown stream ", sendStreamID);
+			// TODO: implement it better
-				delete packet;
+			*m_LeaseSet = *newLeaseSet;
+			delete newLeaseSet;
 		}	
 	}	
-		else // new incoming stream
+
+	void ClientDestination::SendTunnelDataMsgs (const std::vector<i2p::tunnel::TunnelMessageBlock>& msgs)
 	{
-			auto incomingStream = CreateNewIncomingStream ();
+		m_CurrentOutboundTunnel = m_Pool->GetNextOutboundTunnel (m_CurrentOutboundTunnel);
-			incomingStream->HandleNextPacket (packet);
+		if (m_CurrentOutboundTunnel)
-			if (m_Acceptor != nullptr)
+			m_CurrentOutboundTunnel->SendTunnelDataMsg (msgs);
-				m_Acceptor (incomingStream);
 		else
 		{
-				LogPrint ("Acceptor for incoming stream is not set");
+			LogPrint ("No outbound tunnels in the pool");
-				DeleteStream (incomingStream);
+			for (auto it: msgs)
+				DeleteI2NPMessage (it.data);
 		}
 	}
+
+	void ClientDestination::ProcessGarlicMessage (I2NPMessage * msg)
+	{
+		m_Service->post (boost::bind (&ClientDestination::HandleGarlicMessage, this, msg)); 
 	}
 
-	Stream * StreamingDestination::CreateNewOutgoingStream (const i2p::data::LeaseSet& remote)
+	void ClientDestination::ProcessDeliveryStatusMessage (I2NPMessage * msg)
 	{
-		Stream * s = new Stream (*m_Service, *this, remote);
+		m_Service->post (boost::bind (&ClientDestination::HandleDeliveryStatusMessage, this, msg)); 
-		std::unique_lock<std::mutex> l(m_StreamsMutex);
-		m_Streams[s->GetRecvStreamID ()] = s;
-		return s;
 	}
 
-	Stream * StreamingDestination::CreateNewIncomingStream ()
+	void ClientDestination::HandleI2NPMessage (const uint8_t * buf, size_t len, i2p::tunnel::InboundTunnel * from)
 	{
-		Stream * s = new Stream (*m_Service, *this);
+		I2NPHeader * header = (I2NPHeader *)buf;
-		std::unique_lock<std::mutex> l(m_StreamsMutex);
+		switch (header->typeID)
-		m_Streams[s->GetRecvStreamID ()] = s;
+		{	
-		return s;
+			case eI2NPData:
+				HandleDataMessage (buf + sizeof (I2NPHeader), be16toh (header->size));
+			break;
+			case eI2NPDatabaseStore:
+				HandleDatabaseStoreMessage (buf + sizeof (I2NPHeader), be16toh (header->size));
+				i2p::HandleI2NPMessage (CreateI2NPMessage (buf, GetI2NPMessageLength (buf), from)); // TODO: remove
+			break;	
+			default:
+				i2p::HandleI2NPMessage (CreateI2NPMessage (buf, GetI2NPMessageLength (buf), from));
+		}		
 	}	
 
-	void StreamingDestination::DeleteStream (Stream * stream)
+	void ClientDestination::HandleDatabaseStoreMessage (const uint8_t * buf, size_t len)
 	{
-		if (stream)
+		I2NPDatabaseStoreMsg * msg = (I2NPDatabaseStoreMsg *)buf;
+		size_t offset = sizeof (I2NPDatabaseStoreMsg);
+		if (msg->replyToken) // TODO:
+			offset += 36;
+		if (msg->type == 1) // LeaseSet
 		{
-			std::unique_lock<std::mutex> l(m_StreamsMutex);
+			LogPrint ("Remote LeaseSet");
-			auto it = m_Streams.find (stream->GetRecvStreamID ());
+			auto it = m_RemoteLeaseSets.find (msg->key);
-			if (it != m_Streams.end ())
+			if (it != m_RemoteLeaseSets.end ())
 			{
-				m_Streams.erase (it);
+				it->second->Update (buf + offset, len - offset); 
-				if (m_Service)
+				LogPrint ("Remote LeaseSet updated");
-					m_Service->post ([stream](void) { delete stream; }); 
+			}
 			else
-					delete stream;
+			{	
+				LogPrint ("New remote LeaseSet added");
+				m_RemoteLeaseSets[msg->key] = new i2p::data::LeaseSet (buf + offset, len - offset);
 			}	
 		}	
+		else
+			LogPrint ("Unexpected client's DatabaseStore type ", msg->type, ". Dropped");
 	}	
 
-	const i2p::data::LeaseSet * StreamingDestination::GetLeaseSet ()
+	void ClientDestination::SetLeaseSetUpdated ()
 	{
-		if (!m_Pool) return nullptr;
+		i2p::garlic::GarlicDestination::SetLeaseSetUpdated ();	
-		if (!m_LeaseSet)
 		UpdateLeaseSet ();
-		return m_LeaseSet;
+		if (m_IsPublic)
-	}	
+			i2p::data::netdb.PublishLeaseSet (m_LeaseSet, m_Pool);
-
-	void StreamingDestination::UpdateLeaseSet ()
-	{
-		auto newLeaseSet = new i2p::data::LeaseSet (*m_Pool);
-		if (!m_LeaseSet)
-			m_LeaseSet = newLeaseSet;
-		else
-		{	
-			// TODO: implement it better
-			*m_LeaseSet = *newLeaseSet;
-			delete newLeaseSet;
-		}	
 	}
 
-	void StreamingDestination::HandleDataMessage (const uint8_t * buf, size_t len)
+	void ClientDestination::HandleDataMessage (const uint8_t * buf, size_t len)
 	{
 		uint32_t length = be32toh (*(uint32_t *)buf);
 		buf += 4;
@@ -226,10 +244,10 @@ namespace stream
 			CryptoPP::Gunzip decompressor;
 			decompressor.Put (buf, length);
 			decompressor.MessageEnd();
-			Packet * uncompressed = new Packet;
+			i2p::stream::Packet * uncompressed = new i2p::stream::Packet;
 			uncompressed->offset = 0;
 			uncompressed->len = decompressor.MaxRetrievable ();
-			if (uncompressed->len <= MAX_PACKET_SIZE)
+			if (uncompressed->len <= i2p::stream::MAX_PACKET_SIZE)
 			{
 				decompressor.Get (uncompressed->buf, uncompressed->len);
 				HandleNextPacket (uncompressed); 
@@ -245,11 +263,11 @@ namespace stream
 			LogPrint ("Data: unexpected protocol ", buf[9]);
 	}	
 	
-	I2NPMessage * StreamingDestination::CreateDataMessage (const uint8_t * payload, size_t len)
+	I2NPMessage * ClientDestination::CreateDataMessage (const uint8_t * payload, size_t len)
 	{
 		I2NPMessage * msg = NewI2NPShortMessage ();
 		CryptoPP::Gzip compressor;
-		if (len <= COMPRESSION_THRESHOLD_SIZE)
+		if (len <= i2p::stream::COMPRESSION_THRESHOLD_SIZE)
 			compressor.SetDeflateLevel (CryptoPP::Gzip::MIN_DEFLATE_LEVEL);
 		else
 			compressor.SetDeflateLevel (CryptoPP::Gzip::DEFAULT_DEFLATE_LEVEL);
@@ -267,91 +285,88 @@ namespace stream
 		
 		return msg;
 	}				
+}
 
-	void StreamingDestination::SetLeaseSetUpdated ()
+namespace stream
-	{
+{
-		i2p::garlic::GarlicDestination::SetLeaseSetUpdated ();	
-		UpdateLeaseSet ();
-		if (m_IsPublic)
-			i2p::data::netdb.PublishLeaseSet (m_LeaseSet, m_Pool);
-	}	
-
-	void StreamingDestination::ProcessGarlicMessage (I2NPMessage * msg)
-	{
-		m_Service->post (boost::bind (&StreamingDestination::HandleGarlicMessage, this, msg)); 
-	}
 
-	void StreamingDestination::ProcessDeliveryStatusMessage (I2NPMessage * msg)
+	void StreamingDestination::Start ()
 	{	
-		m_Service->post (boost::bind (&StreamingDestination::HandleDeliveryStatusMessage, this, msg)); 
+		ClientDestination::Start ();
 	}
 		
-	void StreamingDestination::HandleI2NPMessage (const uint8_t * buf, size_t len, i2p::tunnel::InboundTunnel * from)
+	void StreamingDestination::Stop ()
 	{	
-		I2NPHeader * header = (I2NPHeader *)buf;
+		ResetAcceptor ();
-		switch (header->typeID)
 		{
-			case eI2NPData:
+			std::unique_lock<std::mutex> l(m_StreamsMutex);
-				HandleDataMessage (buf + sizeof (I2NPHeader), be16toh (header->size));
+			for (auto it: m_Streams)
-			break;
+				delete it.second;	
-			case eI2NPDatabaseStore:
+			m_Streams.clear ();
-				HandleDatabaseStoreMessage (buf + sizeof (I2NPHeader), be16toh (header->size));
-				i2p::HandleI2NPMessage (CreateI2NPMessage (buf, GetI2NPMessageLength (buf), from)); // TODO: remove
-			break;	
-			default:
-				i2p::HandleI2NPMessage (CreateI2NPMessage (buf, GetI2NPMessageLength (buf), from));
 		}	
+		ClientDestination::Stop ();		
 	}	
 
-	void StreamingDestination::HandleDatabaseStoreMessage (const uint8_t * buf, size_t len)
+
+	void StreamingDestination::HandleNextPacket (Packet * packet)
 	{
-		I2NPDatabaseStoreMsg * msg = (I2NPDatabaseStoreMsg *)buf;
+		uint32_t sendStreamID = packet->GetSendStreamID ();
-		size_t offset = sizeof (I2NPDatabaseStoreMsg);
+		if (sendStreamID)
-		if (msg->replyToken) // TODO:
-			offset += 36;
-		if (msg->type == 1) // LeaseSet
 		{	
-			LogPrint ("Remote LeaseSet");
+			auto it = m_Streams.find (sendStreamID);
-			auto it = m_RemoteLeaseSets.find (msg->key);
+			if (it != m_Streams.end ())
-			if (it != m_RemoteLeaseSets.end ())
+				it->second->HandleNextPacket (packet);
+			else
 			{	
-				it->second->Update (buf + offset, len - offset); 
+				LogPrint ("Unknown stream ", sendStreamID);
-				LogPrint ("Remote LeaseSet updated");
+				delete packet;
+			}
 		}	
+		else // new incoming stream
+		{
+			auto incomingStream = CreateNewIncomingStream ();
+			incomingStream->HandleNextPacket (packet);
+			if (m_Acceptor != nullptr)
+				m_Acceptor (incomingStream);
 			else
 			{
-				LogPrint ("New remote LeaseSet added");
+				LogPrint ("Acceptor for incoming stream is not set");
-				m_RemoteLeaseSets[msg->key] = new i2p::data::LeaseSet (buf + offset, len - offset);
+				DeleteStream (incomingStream);
 			}
 		}	
-		else
-			LogPrint ("Unexpected client's DatabaseStore type ", msg->type, ". Dropped");
 	}	
 
-	const i2p::data::LeaseSet * StreamingDestination::FindLeaseSet (const i2p::data::IdentHash& ident)
+	Stream * StreamingDestination::CreateNewOutgoingStream (const i2p::data::LeaseSet& remote)
-	{
-		auto it = m_RemoteLeaseSets.find (ident);
-		if (it != m_RemoteLeaseSets.end ())
-		{	
-			if (it->second->HasNonExpiredLeases ())
-				return it->second;
-			else
 	{
-				LogPrint ("All leases of remote LeaseSet expired. Request it");
+		Stream * s = new Stream (*GetService (), *this, remote);
-				i2p::data::netdb.RequestDestination (ident, true, m_Pool);
+		std::unique_lock<std::mutex> l(m_StreamsMutex);
+		m_Streams[s->GetRecvStreamID ()] = s;
+		return s;
 	}	
+
+	Stream * StreamingDestination::CreateNewIncomingStream ()
+	{
+		Stream * s = new Stream (*GetService (), *this);
+		std::unique_lock<std::mutex> l(m_StreamsMutex);
+		m_Streams[s->GetRecvStreamID ()] = s;
+		return s;
 	}
-		else
+
+	void StreamingDestination::DeleteStream (Stream * stream)
 	{
-			auto ls = i2p::data::netdb.FindLeaseSet (ident);
+		if (stream)
-			if (ls)
 		{	
-				ls = new i2p::data::LeaseSet (*ls);
+			std::unique_lock<std::mutex> l(m_StreamsMutex);
-				m_RemoteLeaseSets[ident] = ls;			
+			auto it = m_Streams.find (stream->GetRecvStreamID ());
-				return ls;
+			if (it != m_Streams.end ())
+			{	
+				m_Streams.erase (it);
+				if (GetService ())
+					GetService ()->post ([stream](void) { delete stream; }); 
+				else
+					delete stream;
 			}	
 		}	
-		return nullptr;
 	}	
 }		
 }

Destination.h

@@ -12,35 +12,26 @@
 
 namespace i2p
 {
-namespace stream
+namespace client
 {
-	class StreamingDestination: public i2p::garlic::GarlicDestination 
+	class ClientDestination: public i2p::garlic::GarlicDestination
 	{
 		public:
 
-			StreamingDestination (bool isPublic, i2p::data::SigningKeyType sigType);
+			ClientDestination (bool isPublic, i2p::data::SigningKeyType sigType);
-			StreamingDestination (const std::string& fullPath, bool isPublic);
+			ClientDestination (const std::string& fullPath, bool isPublic);
-			StreamingDestination (const i2p::data::PrivateKeys& keys, bool isPublic);
+			ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic);
-			~StreamingDestination ();	
+			~ClientDestination ();	
 
-			void Start ();
+			virtual void Start ();
-			void Stop ();
+			virtual void Stop ();
 			bool IsRunning () const { return m_IsRunning; };
+			boost::asio::io_service * GetService () { return m_Service; };
+			i2p::tunnel::TunnelPool * GetTunnelPool () { return m_Pool; }; 
 
-			i2p::tunnel::TunnelPool * GetTunnelPool () const  { return m_Pool; };			
-
-			Stream * CreateNewOutgoingStream (const i2p::data::LeaseSet& remote);
-			void DeleteStream (Stream * stream);			
-			void SetAcceptor (const std::function<void (Stream *)>& acceptor) { m_Acceptor = acceptor; };
-			void ResetAcceptor () { m_Acceptor = nullptr; };
-			bool IsAcceptorSet () const { return m_Acceptor != nullptr; };	
-			void HandleNextPacket (Packet * packet);
-			void SendTunnelDataMsgs (const std::vector<i2p::tunnel::TunnelMessageBlock>& msgs);
 			void ResetCurrentOutboundTunnel () { m_CurrentOutboundTunnel = nullptr; };
 			const i2p::data::LeaseSet * FindLeaseSet (const i2p::data::IdentHash& ident);
-			// I2CP
+			void SendTunnelDataMsgs (const std::vector<i2p::tunnel::TunnelMessageBlock>& msgs);
-			void HandleDataMessage (const uint8_t * buf, size_t len);
-			I2NPMessage * CreateDataMessage (const uint8_t * payload, size_t len);
 
 			// implements LocalDestination
 			const i2p::data::PrivateKeys& GetPrivateKeys () const { return m_Keys; };
@@ -56,10 +47,17 @@ namespace stream
 			void ProcessDeliveryStatusMessage (I2NPMessage * msg);	
 			void SetLeaseSetUpdated ();
 
+			// I2CP
+			void HandleDataMessage (const uint8_t * buf, size_t len);
+			I2NPMessage * CreateDataMessage (const uint8_t * payload, size_t len);
+
+		protected:
+
+			virtual void HandleNextPacket (i2p::stream::Packet * packet) = 0; // TODO	
+
 		private:
 				
 			void Run ();			
-			Stream * CreateNewIncomingStream ();
 			void UpdateLeaseSet ();
 			void HandleDatabaseStoreMessage (const uint8_t * buf, size_t len);			
 
@@ -69,24 +67,61 @@ namespace stream
 			std::thread * m_Thread;	
 			boost::asio::io_service * m_Service;
 			boost::asio::io_service::work * m_Work;
-			std::map<i2p::data::IdentHash, i2p::data::LeaseSet *> m_RemoteLeaseSets;
-
-			std::mutex m_StreamsMutex;
-			std::map<uint32_t, Stream *> m_Streams;
 			i2p::data::PrivateKeys m_Keys;
 			uint8_t m_EncryptionPublicKey[256], m_EncryptionPrivateKey[256];
+			std::map<i2p::data::IdentHash, i2p::data::LeaseSet *> m_RemoteLeaseSets;
 
 			i2p::tunnel::TunnelPool * m_Pool;
 			i2p::tunnel::OutboundTunnel * m_CurrentOutboundTunnel;
 			i2p::data::LeaseSet * m_LeaseSet;
 			bool m_IsPublic;
 		
+		public:
+			
+			// for HTTP only
+			int GetNumRemoteLeaseSets () const { return m_RemoteLeaseSets.size (); };
+	};	
+}
+
+namespace stream
+{
+	class StreamingDestination: public i2p::client::ClientDestination 
+	{
+		public:
+
+			StreamingDestination (bool isPublic, i2p::data::SigningKeyType sigType):
+				ClientDestination (isPublic, sigType) {};
+			StreamingDestination (const std::string& fullPath, bool isPublic):
+				ClientDestination (fullPath, isPublic) {};
+			StreamingDestination (const i2p::data::PrivateKeys& keys, bool isPublic):
+				ClientDestination (keys, isPublic) {};
+			~StreamingDestination () {};	
+
+			void Start ();
+			void Stop ();
+
+			Stream * CreateNewOutgoingStream (const i2p::data::LeaseSet& remote);
+			void DeleteStream (Stream * stream);			
+			void SetAcceptor (const std::function<void (Stream *)>& acceptor) { m_Acceptor = acceptor; };
+			void ResetAcceptor () { m_Acceptor = nullptr; };
+			bool IsAcceptorSet () const { return m_Acceptor != nullptr; };	
+
+			// ClientDestination
+			void HandleNextPacket (Packet * packet);
+
+		private:		
+	
+			Stream * CreateNewIncomingStream ();
+
+		private:
+
+			std::mutex m_StreamsMutex;
+			std::map<uint32_t, Stream *> m_Streams;
 			std::function<void (Stream *)> m_Acceptor;
 			
 		public:
 
 			// for HTTP only
-			int GetNumRemoteLeaseSets () const { return m_RemoteLeaseSets.size (); };
 			const decltype(m_Streams)& GetStreams () const { return m_Streams; };
 	};		
 }