Merge pull request #940 from majestrate/defer-connect

Defer connect

libi2pd_client/ClientContext.cpp

@@ -551,6 +551,13 @@ namespace client
 							clientTunnel = new I2PClientTunnel (name, dest, address, port, localDestination, destinationPort);
 							clientEndpoint = ((I2PClientTunnel*)clientTunnel)->GetLocalEndpoint ();
 						}
+						uint32_t timeout = section.second.get<uint32_t>(I2P_CLIENT_TUNNEL_CONNECT_TIMEOUT, 0);
+						if(timeout)
+						{
+							clientTunnel->SetConnectTimeout(timeout);
+							LogPrint(eLogInfo, "Clients: I2P Client tunnel connect timeout set to ", timeout);
+						}
+
 						auto ins = m_ClientTunnels.insert (std::make_pair (clientEndpoint,	std::unique_ptr<I2PService>(clientTunnel)));
 						if (ins.second)
 						{

libi2pd_client/ClientContext.h

@@ -36,6 +36,7 @@ namespace client
 	const char I2P_CLIENT_TUNNEL_SIGNATURE_TYPE[] = "signaturetype";
 	const char I2P_CLIENT_TUNNEL_DESTINATION_PORT[] = "destinationport";
 	const char I2P_CLIENT_TUNNEL_MATCH_TUNNELS[] = "matchtunnels";
+  const char I2P_CLIENT_TUNNEL_CONNECT_TIMEOUT[] = "connecttimeout";
 	const char I2P_SERVER_TUNNEL_HOST[] = "host";
 	const char I2P_SERVER_TUNNEL_HOST_OVERRIDE[] = "hostoverride";
 	const char I2P_SERVER_TUNNEL_PORT[] = "port";

libi2pd_client/I2PService.cpp

@@ -2,6 +2,7 @@
 #include "Identity.h"
 #include "ClientContext.h"
 #include "I2PService.h"
+#include <boost/asio/error.hpp>
 
 namespace i2p
 {
@@ -11,13 +12,18 @@ namespace client
 
 	I2PService::I2PService (std::shared_ptr<ClientDestination> localDestination):
 		m_LocalDestination (localDestination ? localDestination :
-					i2p::client::context.CreateNewLocalDestination (false, I2P_SERVICE_DEFAULT_KEY_TYPE)), isUpdated (true)
+			i2p::client::context.CreateNewLocalDestination (false, I2P_SERVICE_DEFAULT_KEY_TYPE)),
+    m_ReadyTimer(m_LocalDestination->GetService()),
+		m_ConnectTimeout(0),
+			isUpdated (true)
 	{
 		m_LocalDestination->Acquire ();
 	}
 
 	I2PService::I2PService (i2p::data::SigningKeyType kt):
 		m_LocalDestination (i2p::client::context.CreateNewLocalDestination (false, kt)),
+		m_ReadyTimer(m_LocalDestination->GetService()),
+		m_ConnectTimeout(0),
 		isUpdated (true)
 	{
 		m_LocalDestination->Acquire ();
@@ -31,17 +37,76 @@ namespace client
 
 	void I2PService::ClearHandlers ()
 	{
+		if(m_ConnectTimeout)
+			m_ReadyTimer.cancel();
 		std::unique_lock<std::mutex> l(m_HandlersMutex);
 		for (auto it: m_Handlers)
 			it->Terminate ();
 		m_Handlers.clear();
 	}
 
+	void I2PService::SetConnectTimeout(uint32_t timeout)
+	{
+		if(timeout && !m_ConnectTimeout)
+		{
+				TriggerReadyCheckTimer();
+		}
+		else if (m_ConnectTimeout && !timeout)
+		{
+			m_ReadyTimer.cancel();
+		}
+		m_ConnectTimeout = timeout;
+	}
+
+	void I2PService::AddReadyCallback(ReadyCallback cb)
+	{
+		uint32_t now = i2p::util::GetSecondsSinceEpoch();
+		uint32_t tm = now + m_ConnectTimeout;
+		LogPrint(eLogDebug, "I2PService::AddReadyCallback() ", tm, " ", now);
+		m_ReadyCallbacks.push_back({cb, tm});
+	}
+
+	void I2PService::TriggerReadyCheckTimer()
+	{
+		m_ReadyTimer.expires_from_now(boost::posix_time::seconds (1));
+		m_ReadyTimer.async_wait(std::bind(&I2PService::HandleReadyCheckTimer, this, std::placeholders::_1));
+	}
+
+	void I2PService::HandleReadyCheckTimer(const boost::system::error_code &ec)
+	{
+		if(ec || m_LocalDestination->IsReady())
+		{
+			for(auto & itr : m_ReadyCallbacks)
+				itr.first(ec);
+			m_ReadyCallbacks.clear();
+		}
+		else if(!m_LocalDestination->IsReady())
+		{
+			// expire timed out requests
+			uint32_t now = i2p::util::GetSecondsSinceEpoch ();
+			auto itr = m_ReadyCallbacks.begin();
+			while(itr != m_ReadyCallbacks.end())
+			{
+				if(itr->second >= now)
+				{
+					itr->first(boost::asio::error::timed_out);
+					itr = m_ReadyCallbacks.erase(itr);
+				}
+				else
+					++itr;
+			}
+		}
+		if(!ec)
+			TriggerReadyCheckTimer();
+	}
+
 	void I2PService::CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port) {
 		assert(streamRequestComplete);
 		i2p::data::IdentHash identHash;
 		if (i2p::client::context.GetAddressBook ().GetIdentHash (dest, identHash))
-			m_LocalDestination->CreateStream (streamRequestComplete, identHash, port);
+		{
+			CreateStream(streamRequestComplete, identHash, port);
+		}
 		else
 		{
 			LogPrint (eLogWarning, "I2PService: Remote destination not found: ", dest);
@@ -49,6 +114,29 @@ namespace client
 		}
 	}
 
+	void I2PService::CreateStream(StreamRequestComplete streamRequestComplete, const i2p::data::IdentHash & identHash, int port)
+	{
+		if(m_ConnectTimeout)
+		{
+			if(m_LocalDestination->IsReady())
+				m_LocalDestination->CreateStream (streamRequestComplete, identHash, port);
+			else
+			{
+				AddReadyCallback([this, streamRequestComplete, identHash, port] (const boost::system::error_code & ec) {
+						if(ec)
+						{
+							LogPrint(eLogWarning, "I2PService::CeateStream() ", ec.message());
+							streamRequestComplete(nullptr);
+						}
+						else
+							this->m_LocalDestination->CreateStream(streamRequestComplete, identHash, port);
+					});
+			}
+		}
+		else
+			m_LocalDestination->CreateStream(streamRequestComplete, identHash, port);
+	}
+
 	TCPIPPipe::TCPIPPipe(I2PService * owner, std::shared_ptr<boost::asio::ip::tcp::socket> upstream, std::shared_ptr<boost::asio::ip::tcp::socket> downstream) : I2PServiceHandler(owner), m_up(upstream), m_down(downstream)
 	{
 		boost::asio::socket_base::receive_buffer_size option(TCP_IP_PIPE_BUFFER_SIZE);

libi2pd_client/I2PService.h

@@ -14,8 +14,10 @@ namespace i2p
 namespace client
 {
 	class I2PServiceHandler;
-	class I2PService
+	class I2PService : std::enable_shared_from_this<I2PService>
 	{
+  public:
+    typedef std::function<void(const boost::system::error_code &)> ReadyCallback;
 		public:
 			I2PService (std::shared_ptr<ClientDestination> localDestination  = nullptr);
 			I2PService (i2p::data::SigningKeyType kt);
@@ -33,22 +35,34 @@ namespace client
 			}
 			void ClearHandlers ();
 
+    void SetConnectTimeout(uint32_t timeout);
+
+    void AddReadyCallback(ReadyCallback cb);
+
 			inline std::shared_ptr<ClientDestination> GetLocalDestination () { return m_LocalDestination; }
 			inline std::shared_ptr<const ClientDestination> GetLocalDestination () const  { return m_LocalDestination; }
 			inline void SetLocalDestination (std::shared_ptr<ClientDestination> dest) { m_LocalDestination = dest; }
 			void CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port = 0);
-
+      void CreateStream(StreamRequestComplete complete, const i2p::data::IdentHash & ident, int port);
 			inline boost::asio::io_service& GetService () { return m_LocalDestination->GetService (); }
 
 			virtual void Start () = 0;
 			virtual void Stop () = 0;
 
 			virtual const char* GetName() { return "Generic I2P Service"; }
+
+  private:
+    void TriggerReadyCheckTimer();
+    void HandleReadyCheckTimer(const boost::system::error_code & ec);
+
 		private:
 
 			std::shared_ptr<ClientDestination> m_LocalDestination;
 			std::unordered_set<std::shared_ptr<I2PServiceHandler> > m_Handlers;
 			std::mutex m_HandlersMutex;
+    std::vector<std::pair<ReadyCallback, uint32_t> > m_ReadyCallbacks;
+    boost::asio::deadline_timer m_ReadyTimer;
+    uint32_t m_ConnectTimeout;
 
 		public:
 			bool isUpdated; // transient, used during reload only

libi2pd_client/I2PTunnel.cpp

@@ -402,7 +402,7 @@ namespace client
 
 	void I2PClientTunnelHandler::Handle()
 	{
-		GetOwner()->GetLocalDestination ()->CreateStream ( 
+		GetOwner()->CreateStream (
 			std::bind (&I2PClientTunnelHandler::HandleStreamRequestComplete, shared_from_this(), std::placeholders::_1),
 			m_DestinationIdentHash, m_DestinationPort);
 	}