Browse Source

Merge pull request #129 from klondi/httpproxy

Asynchronous stream creation
pull/130/head
orignal 10 years ago
parent
commit
f1ae5817eb
  1. 43
      Destination.cpp
  2. 6
      Destination.h
  3. 2
      I2PTunnel.h

43
Destination.cpp

@ -1,5 +1,5 @@
#include <algorithm> #include <algorithm>
#include <mutex> #include <cassert>
#include <boost/lexical_cast.hpp> #include <boost/lexical_cast.hpp>
#include <cryptopp/dh.h> #include <cryptopp/dh.h>
#include "Log.h" #include "Log.h"
@ -384,46 +384,47 @@ namespace client
} }
} }
std::shared_ptr<i2p::stream::Stream> ClientDestination::CreateStream (const std::string& dest, int port) { void ClientDestination::CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port) {
assert(streamRequestComplete);
i2p::data::IdentHash identHash; i2p::data::IdentHash identHash;
if (i2p::client::context.GetAddressBook ().GetIdentHash (dest, identHash)) if (i2p::client::context.GetAddressBook ().GetIdentHash (dest, identHash))
return CreateStream (identHash, port); CreateStream (streamRequestComplete, identHash, port);
else else
{ {
LogPrint (eLogWarning, "Remote destination ", dest, " not found"); LogPrint (eLogWarning, "Remote destination ", dest, " not found");
return nullptr; streamRequestComplete (nullptr);
} }
} }
std::shared_ptr<i2p::stream::Stream> ClientDestination::CreateStream (const i2p::data::IdentHash& dest, int port) { void ClientDestination::CreateStream (StreamRequestComplete streamRequestComplete, const i2p::data::IdentHash& dest, int port) {
assert(streamRequestComplete);
const i2p::data::LeaseSet * leaseSet = FindLeaseSet (dest); const i2p::data::LeaseSet * leaseSet = FindLeaseSet (dest);
if (!leaseSet) if (leaseSet)
streamRequestComplete(CreateStream (*leaseSet, port));
else
{ {
bool found = false;
std::condition_variable newDataReceived;
std::mutex newDataReceivedMutex;
std::unique_lock<std::mutex> l(newDataReceivedMutex);
RequestDestination (dest, RequestDestination (dest,
[&newDataReceived, &found](bool success) [this, streamRequestComplete, dest, port](bool success)
{ {
found = success; if (!success)
newDataReceived.notify_all (); streamRequestComplete (nullptr);
}); else
if (newDataReceived.wait_for (l, std::chrono::seconds (STREAM_REQUEST_TIMEOUT)) == std::cv_status::timeout) {
LogPrint (eLogError, "Subscription LeseseSet request timeout expired"); const i2p::data::LeaseSet * leaseSet = FindLeaseSet (dest);
if (found)
leaseSet = FindLeaseSet (dest);
}
if (leaseSet) if (leaseSet)
return CreateStream (*leaseSet, port); streamRequestComplete(CreateStream (*leaseSet, port));
else else
return nullptr; streamRequestComplete (nullptr);
}
});
}
} }
std::shared_ptr<i2p::stream::Stream> ClientDestination::CreateStream (const i2p::data::LeaseSet& remote, int port) std::shared_ptr<i2p::stream::Stream> ClientDestination::CreateStream (const i2p::data::LeaseSet& remote, int port)
{ {
if (m_StreamingDestination) if (m_StreamingDestination)
return m_StreamingDestination->CreateNewOutgoingStream (remote, port); return m_StreamingDestination->CreateNewOutgoingStream (remote, port);
else
return nullptr; return nullptr;
} }

6
Destination.h

@ -49,6 +49,8 @@ namespace client
RequestComplete requestComplete; RequestComplete requestComplete;
}; };
typedef std::function<void (std::shared_ptr<i2p::stream::Stream> stream)> StreamRequestComplete;
public: public:
ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic, const std::map<std::string, std::string> * params = nullptr); ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic, const std::map<std::string, std::string> * params = nullptr);
@ -65,8 +67,8 @@ namespace client
// streaming // streaming
i2p::stream::StreamingDestination * GetStreamingDestination () const { return m_StreamingDestination; }; i2p::stream::StreamingDestination * GetStreamingDestination () const { return m_StreamingDestination; };
std::shared_ptr<i2p::stream::Stream> CreateStream (const std::string& dest, int port = 0); void CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port = 0);
std::shared_ptr<i2p::stream::Stream> CreateStream (const i2p::data::IdentHash& dest, int port = 0); void CreateStream (StreamRequestComplete streamRequestComplete, const i2p::data::IdentHash& dest, int port = 0);
std::shared_ptr<i2p::stream::Stream> CreateStream (const i2p::data::LeaseSet& remote, int port = 0); std::shared_ptr<i2p::stream::Stream> CreateStream (const i2p::data::LeaseSet& remote, int port = 0);
void AcceptStreams (const i2p::stream::StreamingDestination::Acceptor& acceptor); void AcceptStreams (const i2p::stream::StreamingDestination::Acceptor& acceptor);
void StopAcceptingStreams (); void StopAcceptingStreams ();

2
I2PTunnel.h

@ -17,7 +17,7 @@ namespace client
const size_t I2P_TUNNEL_CONNECTION_BUFFER_SIZE = 8192; const size_t I2P_TUNNEL_CONNECTION_BUFFER_SIZE = 8192;
const int I2P_TUNNEL_CONNECTION_MAX_IDLE = 3600; // in seconds const int I2P_TUNNEL_CONNECTION_MAX_IDLE = 3600; // in seconds
const int I2P_TUNNEL_DESTINATION_REQUEST_TIMEOUT = 10; // in seconds const int I2P_TUNNEL_DESTINATION_REQUEST_TIMEOUT = 10; // in seconds
const uint16_t I2P_TUNNEL_DEFAULT_KEY_TYPE = i2p::data::SIGNING_KEY_TYPE_ECDSA_SHA256_P256; const i2p::data::SigningKeyType I2P_TUNNEL_DEFAULT_KEY_TYPE = i2p::data::SIGNING_KEY_TYPE_ECDSA_SHA256_P256;
class I2PTunnel; class I2PTunnel;
class I2PTunnelConnection: public std::enable_shared_from_this<I2PTunnelConnection> class I2PTunnelConnection: public std::enable_shared_from_this<I2PTunnelConnection>

Loading…
Cancel
Save