Merge pull request #146 from klondi/i2pservice

I2pservice

HTTPProxy.cpp

@@ -231,41 +231,9 @@ namespace proxy
 		}
 	}
 
-	void HTTPProxyServer::Start ()
+	std::shared_ptr<i2p::client::I2PServiceHandler> HTTPProxyServer::CreateHandler(boost::asio::ip::tcp::socket * socket)
 	{
-		m_Acceptor.listen ();
-		Accept ();
-	}
-
-	void HTTPProxyServer::Stop ()
-	{
-		m_Acceptor.close();
-		m_Timer.cancel ();
-		ClearHandlers();
-	}
-
-	void HTTPProxyServer::Accept ()
-	{
-		auto newSocket = new boost::asio::ip::tcp::socket (GetService ());
-		m_Acceptor.async_accept (*newSocket, std::bind (&HTTPProxyServer::HandleAccept, this,
-			std::placeholders::_1, newSocket));
-	}
-
-	void HTTPProxyServer::HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket)
-	{
-		if (!ecode)
-		{
-			LogPrint(eLogDebug,"--- HTTP Proxy accepted");
-			auto handler = std::make_shared<HTTPProxyHandler> (this, socket);
-			AddHandler(handler);
-			handler->Handle();
-			Accept();
-		}
-		else
-		{
-			LogPrint (eLogError,"--- HTTP Proxy Closing socket on accept because: ", ecode.message ());
-			delete socket;
-		}
+		return std::make_shared<HTTPProxyHandler> (this, socket);
 	}
 
 }

HTTPProxy.h

@@ -11,28 +11,16 @@ namespace i2p
 {
 namespace proxy
 {
-	class HTTPProxyHandler;
-	class HTTPProxyServer: public i2p::client::I2PService
+	class HTTPProxyServer: public i2p::client::TCPIPAcceptor
 	{
-		private:
-			std::set<std::shared_ptr<HTTPProxyHandler> > m_Handlers;
-			boost::asio::ip::tcp::acceptor m_Acceptor;
-			boost::asio::deadline_timer m_Timer;
-			std::mutex m_HandlersMutex;
-
-		private:
-
-			void Accept();
-			void HandleAccept(const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket);
+		protected:
+			// Implements TCPIPAcceptor
+			std::shared_ptr<i2p::client::I2PServiceHandler> CreateHandler(boost::asio::ip::tcp::socket * socket);
+			const char* GetName() { return "HTTP Proxy"; }
 
 		public:
-			HTTPProxyServer(int port) : I2PService(i2p::data::SIGNING_KEY_TYPE_DSA_SHA1),
-				m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
-				m_Timer (GetService ()) {};
-			~HTTPProxyServer() { Stop(); }
-
-			void Start ();
-			void Stop ();
+			HTTPProxyServer(int port) : TCPIPAcceptor(port, i2p::data::SIGNING_KEY_TYPE_DSA_SHA1) {}
+			~HTTPProxyServer() {}
 	};
 
 	typedef HTTPProxyServer HTTPProxy;

I2PService.cpp

@@ -32,5 +32,49 @@ namespace client
 			streamRequestComplete (nullptr);
 		}
 	}
+
+	void TCPIPAcceptor::Start ()
+	{
+		m_Acceptor.listen ();
+		Accept ();
+	}
+
+	void TCPIPAcceptor::Stop ()
+	{
+		m_Acceptor.close();
+		m_Timer.cancel ();
+		ClearHandlers();
+	}
+
+	void TCPIPAcceptor::Accept ()
+	{
+		auto newSocket = new boost::asio::ip::tcp::socket (GetService ());
+		m_Acceptor.async_accept (*newSocket, std::bind (&TCPIPAcceptor::HandleAccept, this,
+			std::placeholders::_1, newSocket));
+	}
+
+	void TCPIPAcceptor::HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket)
+	{
+		if (!ecode)
+		{
+			LogPrint(eLogDebug,"--- ",GetName()," accepted");
+			auto handler = CreateHandler(socket);
+			if (handler) {
+				AddHandler(handler);
+				handler->Handle();
+			} else {
+				socket->close();
+				delete socket;
+			}
+			Accept();
+		}
+		else
+		{
+			if (ecode != boost::asio::error::operation_aborted)
+				LogPrint (eLogError,"--- ",GetName()," Closing socket on accept because: ", ecode.message ());
+			delete socket;
+		}
+	}
+
 }
 }

I2PService.h

@@ -46,6 +46,7 @@ namespace client
 			virtual void Start () = 0;
 			virtual void Stop () = 0;
 
+			virtual const char* GetName() { return "Generic I2P Service"; }
 		private:
 
 			ClientDestination * m_LocalDestination;
@@ -59,19 +60,49 @@ namespace client
 		public:
 			I2PServiceHandler(I2PService * parent) : m_Service(parent), m_Dead(false) { }
 			virtual ~I2PServiceHandler() { }
+			//If you override this make sure you call it from the children
+			virtual void Handle() {}; //Start handling the socket
 		protected:
 			// Call when terminating or handing over to avoid race conditions
-			inline bool Kill() { return m_Dead.exchange(true); }
+			inline bool Kill () { return m_Dead.exchange(true); }
 			// Call to know if the handler is dead
-			inline bool Dead() { return m_Dead; }
+			inline bool Dead () { return m_Dead; }
 			// Call when done to clean up (make sure Kill is called first)
-			inline void Done(std::shared_ptr<I2PServiceHandler> me) { if(m_Service) m_Service->RemoveHandler(me); }
+			inline void Done (std::shared_ptr<I2PServiceHandler> me) { if(m_Service) m_Service->RemoveHandler(me); }
 			// Call to talk with the owner
 			inline I2PService * GetOwner() { return m_Service; }
 		private:
 			I2PService *m_Service;
 			std::atomic<bool> m_Dead; //To avoid cleaning up multiple times
 	};
+
+	/* TODO: support IPv6 too */
+	//This is a service that listens for connections on the IP network and interacts with I2P
+	class TCPIPAcceptor: public I2PService
+	{
+		public:
+			TCPIPAcceptor (int port, ClientDestination * localDestination = nullptr) :
+				I2PService(localDestination),
+				m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
+				m_Timer (GetService ()) {}
+			TCPIPAcceptor (int port, i2p::data::SigningKeyType kt) :
+				I2PService(kt),
+				m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
+				m_Timer (GetService ()) {}
+			virtual ~TCPIPAcceptor () { TCPIPAcceptor::Stop(); }
+			//If you override this make sure you call it from the children
+			void Start ();
+			//If you override this make sure you call it from the children
+			void Stop ();
+		protected:
+			virtual std::shared_ptr<I2PServiceHandler> CreateHandler(boost::asio::ip::tcp::socket * socket) = 0;
+			virtual const char* GetName() { return "Generic TCP/IP accepting daemon"; }
+		private:
+			void Accept();
+			void HandleAccept(const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket);
+			boost::asio::ip::tcp::acceptor m_Acceptor;
+			boost::asio::deadline_timer m_Timer;
+	};
 }
 }
 

I2PTunnel.cpp

@@ -151,30 +151,69 @@ namespace client
 		}
 	}
 
-	I2PClientTunnel::I2PClientTunnel (const std::string& destination, int port, ClientDestination * localDestination): 
-		I2PService (localDestination),
-		m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
-		m_Timer (GetService ()), m_Destination (destination), m_DestinationIdentHash (nullptr)
+	/* This handler tries to stablish a connection with the desired server and dies if it fails to do so */
+	class I2PClientTunnelHandler: public I2PServiceHandler, public std::enable_shared_from_this<I2PClientTunnelHandler>
 	{
-	}	
+		public:
+			I2PClientTunnelHandler (I2PClientTunnel * parent, i2p::data::IdentHash destination,
+						boost::asio::ip::tcp::socket * socket):
+				I2PServiceHandler(parent), m_DestinationIdentHash(destination), m_Socket(socket) {}
+			void Handle();
+			void Terminate();
+		private:
+			void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream);
+			i2p::data::IdentHash m_DestinationIdentHash;
+			boost::asio::ip::tcp::socket * m_Socket;
+	};
 
-	I2PClientTunnel::~I2PClientTunnel ()
+	void I2PClientTunnelHandler::Handle()
 	{
-		Stop ();
+		GetOwner()->GetLocalDestination ()->CreateStream (std::bind (&I2PClientTunnelHandler::HandleStreamRequestComplete,
+								shared_from_this(), std::placeholders::_1), m_DestinationIdentHash);
 	}
-	
+
+	void I2PClientTunnelHandler::HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream)
+	{
+		if (stream)
+		{
+			if (Kill()) return;
+			LogPrint (eLogInfo,"New I2PTunnel connection");
+			auto connection = std::make_shared<I2PTunnelConnection>(GetOwner(), m_Socket, stream);
+			GetOwner()->AddHandler (connection);
+			connection->I2PConnect ();
+			Done(shared_from_this());
+		}
+		else
+		{
+			LogPrint (eLogError,"I2P Client Tunnel Issue when creating the stream, check the previous warnings for more info.");
+			Terminate();
+		}
+	}
+
+	void I2PClientTunnelHandler::Terminate()
+	{
+		if (Kill()) return;
+		if (m_Socket)
+		{
+			m_Socket->close();
+			delete m_Socket;
+			m_Socket = nullptr;
+		}
+		Done(shared_from_this());
+	}
+
+	I2PClientTunnel::I2PClientTunnel (const std::string& destination, int port, ClientDestination * localDestination): 
+		TCPIPAcceptor (port,localDestination), m_Destination (destination), m_DestinationIdentHash (nullptr)
+	{}	
+
 	void I2PClientTunnel::Start ()
 	{
 		GetIdentHash();
-		m_Acceptor.listen ();
-		Accept ();
 	}
 
 	void I2PClientTunnel::Stop ()
 	{
-		m_Acceptor.close();
-		m_Timer.cancel ();
-		ClearHandlers ();
+		TCPIPAcceptor::Stop();
 		auto *originalIdentHash = m_DestinationIdentHash;
 		m_DestinationIdentHash = nullptr;
 		delete originalIdentHash;
@@ -194,51 +233,13 @@ namespace client
 		return m_DestinationIdentHash;
 	}
 
-	
-	void I2PClientTunnel::Accept ()
+	std::shared_ptr<I2PServiceHandler> I2PClientTunnel::CreateHandler(boost::asio::ip::tcp::socket * socket)
 	{
-		auto newSocket = new boost::asio::ip::tcp::socket (GetService ());
-		m_Acceptor.async_accept (*newSocket, std::bind (&I2PClientTunnel::HandleAccept, this,
-			std::placeholders::_1, newSocket));
-	}	
-
-	void I2PClientTunnel::HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket)
-	{
-		if (!ecode)
-		{
-			const i2p::data::IdentHash *identHash = GetIdentHash();
-			if (identHash)
-				GetLocalDestination ()->CreateStream (
-					std::bind (&I2PClientTunnel::HandleStreamRequestComplete,
-					this, std::placeholders::_1, socket), *identHash);
-			else
-			{
-				LogPrint (eLogError,"Closing socket");
-				delete socket;
-			}
-			Accept ();
-		}
+		const i2p::data::IdentHash *identHash = GetIdentHash();
+		if (identHash)
+			return  std::make_shared<I2PClientTunnelHandler>(this, *identHash, socket);
 		else
-		{
-			LogPrint (eLogError,"Closing socket on accept because: ", ecode.message ());
-			delete socket;
-		}
-	}
-
-	void I2PClientTunnel::HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream, boost::asio::ip::tcp::socket * socket)
-	{
-		if (stream)
-		{
-			LogPrint (eLogInfo,"New I2PTunnel connection");
-			auto connection = std::make_shared<I2PTunnelConnection>(this, socket, stream);
-			AddHandler (connection);
-			connection->I2PConnect ();
-		}
-		else
-		{
-			LogPrint (eLogError,"Issue when creating the stream, check the previous warnings for more info.");
-			delete socket;
-		}
+			return nullptr;
 	}
 
 	I2PServerTunnel::I2PServerTunnel (const std::string& address, int port, ClientDestination * localDestination): 

I2PTunnel.h

@@ -55,12 +55,18 @@ namespace client
 			bool m_IsQuiet; // don't send destination
 	};
 
-	class I2PClientTunnel: public I2PService
+	class I2PClientTunnel: public TCPIPAcceptor
 	{
+		protected:
+
+			// Implements TCPIPAcceptor
+			std::shared_ptr<I2PServiceHandler> CreateHandler(boost::asio::ip::tcp::socket * socket);
+			const char* GetName() { return "I2P Client Tunnel"; }
+
 		public:
 
 			I2PClientTunnel (const std::string& destination, int port, ClientDestination * localDestination = nullptr);
-			~I2PClientTunnel ();				
+			~I2PClientTunnel () {}
 	
 			void Start ();
 			void Stop ();
@@ -68,14 +74,7 @@ namespace client
 		private:
 
 			const i2p::data::IdentHash * GetIdentHash ();
-			void Accept ();
-			void HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket);
-			void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream, boost::asio::ip::tcp::socket * socket);
 
-		private:
-
-			boost::asio::ip::tcp::acceptor m_Acceptor;
-			boost::asio::deadline_timer m_Timer;
 			std::string m_Destination;
 			const i2p::data::IdentHash * m_DestinationIdentHash;
 	};	

SOCKS.cpp

@@ -514,41 +514,9 @@ namespace proxy
 		}
 	}
 
-	void SOCKSServer::Start ()
+	std::shared_ptr<i2p::client::I2PServiceHandler> SOCKSServer::CreateHandler(boost::asio::ip::tcp::socket * socket)
 	{
-		m_Acceptor.listen ();
-		Accept ();
-	}
-
-	void SOCKSServer::Stop ()
-	{
-		m_Acceptor.close();
-		m_Timer.cancel ();
-		ClearHandlers();
-	}
-
-	void SOCKSServer::Accept ()
-	{
-		auto newSocket = new boost::asio::ip::tcp::socket (GetService ());
-		m_Acceptor.async_accept (*newSocket, std::bind (&SOCKSServer::HandleAccept, this,
-			std::placeholders::_1, newSocket));
-	}
-
-	void SOCKSServer::HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket)
-	{
-		if (!ecode)
-		{
-			LogPrint(eLogDebug,"--- SOCKS accepted");
-			auto handle = std::make_shared<SOCKSHandler> (this, socket);
-			AddHandler(handle);
-			handle->Handle();
-			Accept();
-		}
-		else
-		{
-			LogPrint (eLogError,"--- SOCKS Closing socket on accept because: ", ecode.message ());
-			delete socket;
-		}
+		return std::make_shared<SOCKSHandler> (this, socket);
 	}
 
 }

SOCKS.h

@@ -11,28 +11,16 @@ namespace i2p
 {
 namespace proxy
 {
-	class SOCKSHandler;
-	class SOCKSServer: public i2p::client::I2PService
+	class SOCKSServer: public i2p::client::TCPIPAcceptor
 	{
-		private:
-			std::set<std::shared_ptr<SOCKSHandler> > m_Handlers;
-			boost::asio::ip::tcp::acceptor m_Acceptor;
-			boost::asio::deadline_timer m_Timer;
-			std::mutex m_HandlersMutex;
-
-		private:
-
-			void Accept();
-			void HandleAccept(const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket);
+		protected:
+			// Implements TCPIPAcceptor
+			std::shared_ptr<i2p::client::I2PServiceHandler> CreateHandler(boost::asio::ip::tcp::socket * socket);
+			const char* GetName() { return "SOCKS"; }
 
 		public:
-			SOCKSServer(int port) : I2PService(nullptr),
-				m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
-				m_Timer (GetService ()) {};
-			~SOCKSServer() { Stop(); }
-
-			void Start ();
-			void Stop ();
+			SOCKSServer(int port) : TCPIPAcceptor(port) {}
+			~SOCKSServer() {}
 	};
 
 	typedef SOCKSServer SOCKSProxy;