mirror of
https://github.com/PurpleI2P/i2pd.git
synced 2025-04-27 11:17:49 +02:00
Merge remote-tracking branch 'purple/openssl'
This commit is contained in:
Jeff Becker
commit
be0c1c0912
16 changed files with 154 additions and 149 deletions
128
SAM.cpp
128
SAM.cpp
|
|
@ -54,7 +54,11 @@ namespace client
|
|||
case eSAMSocketTypeAcceptor:
|
||||
{
|
||||
if (m_Session)
|
||||
{
|
||||
m_Session->DelSocket (shared_from_this ());
|
||||
if (m_Session->localDestination)
|
||||
m_Session->localDestination->StopAcceptingStreams ();
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
|
|
@ -104,10 +108,10 @@ namespace client
|
|||
separator++;
|
||||
std::map<std::string, std::string> params;
|
||||
ExtractParams (separator, params);
|
||||
auto it = params.find (SAM_PARAM_MAX);
|
||||
//auto it = params.find (SAM_PARAM_MAX);
|
||||
// TODO: check MIN as well
|
||||
if (it != params.end ())
|
||||
version = it->second;
|
||||
//if (it != params.end ())
|
||||
// version = it->second;
|
||||
}
|
||||
if (version[0] == '3') // we support v3 (3.0 and 3.1) only
|
||||
{
|
||||
|
|
@ -285,11 +289,6 @@ namespace client
|
|||
dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, shared_from_this (),
|
||||
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5));
|
||||
}
|
||||
else
|
||||
{
|
||||
// start accepting streams because we're not a datagram session
|
||||
m_Session->localDestination->AcceptStreams (std::bind (&SAMSession::AcceptI2P, m_Session, std::placeholders::_1));
|
||||
}
|
||||
|
||||
if (m_Session->localDestination->IsReady ())
|
||||
SendSessionCreateReplyOk ();
|
||||
|
|
@ -401,27 +400,17 @@ namespace client
|
|||
m_ID = id;
|
||||
m_Session = m_Owner.FindSession (id);
|
||||
if (m_Session)
|
||||
{
|
||||
{
|
||||
m_SocketType = eSAMSocketTypeAcceptor;
|
||||
m_Session->AddSocket (shared_from_this ());
|
||||
if (!m_Session->localDestination->IsAcceptingStreams ())
|
||||
m_Session->localDestination->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, shared_from_this (), std::placeholders::_1));
|
||||
SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false);
|
||||
}
|
||||
else
|
||||
SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true);
|
||||
}
|
||||
|
||||
void SAMSocket::Accept(std::shared_ptr<i2p::stream::Stream> stream)
|
||||
{
|
||||
if(stream) {
|
||||
m_SocketType = eSAMSocketTypeStream;
|
||||
HandleI2PAccept(stream);
|
||||
} else {
|
||||
SendMessageReply (SAM_STREAM_STATUS_I2P_ERROR, strlen(SAM_STREAM_STATUS_I2P_ERROR), true);
|
||||
auto s = shared_from_this ();
|
||||
m_Owner.GetService ().post ([s] { s->Terminate (); });
|
||||
|
||||
}
|
||||
}
|
||||
size_t SAMSocket::ProcessDatagramSend (char * buf, size_t len, const char * data)
|
||||
{
|
||||
LogPrint (eLogDebug, "SAM: datagram send: ", buf, " ", len);
|
||||
|
|
@ -475,21 +464,21 @@ namespace client
|
|||
std::string& name = params[SAM_PARAM_NAME];
|
||||
std::shared_ptr<const i2p::data::IdentityEx> identity;
|
||||
i2p::data::IdentHash ident;
|
||||
auto dest = m_Session == nullptr ? context.GetSharedLocalDestination() : m_Session->localDestination;
|
||||
if (name == "ME")
|
||||
SendNamingLookupReply (m_Session->localDestination->GetIdentity ());
|
||||
SendNamingLookupReply (dest->GetIdentity ());
|
||||
else if ((identity = context.GetAddressBook ().GetAddress (name)) != nullptr)
|
||||
SendNamingLookupReply (identity);
|
||||
else if (m_Session && m_Session->localDestination &&
|
||||
context.GetAddressBook ().GetIdentHash (name, ident))
|
||||
else if (context.GetAddressBook ().GetIdentHash (name, ident))
|
||||
{
|
||||
auto leaseSet = m_Session->localDestination->FindLeaseSet (ident);
|
||||
auto leaseSet = dest->FindLeaseSet (ident);
|
||||
if (leaseSet)
|
||||
SendNamingLookupReply (leaseSet->GetIdentity ());
|
||||
else
|
||||
m_Session->localDestination->RequestDestination (ident,
|
||||
dest->RequestDestination (ident,
|
||||
std::bind (&SAMSocket::HandleNamingLookupLeaseSetRequestComplete,
|
||||
shared_from_this (), std::placeholders::_1, ident));
|
||||
}
|
||||
shared_from_this (), std::placeholders::_1, ident));
|
||||
}
|
||||
else
|
||||
{
|
||||
LogPrint (eLogError, "SAM: naming failed, unknown address ", name);
|
||||
|
|
@ -664,8 +653,20 @@ namespace client
|
|||
if (stream)
|
||||
{
|
||||
LogPrint (eLogDebug, "SAM: incoming I2P connection for session ", m_ID);
|
||||
m_SocketType = eSAMSocketTypeStream;
|
||||
m_Stream = stream;
|
||||
context.GetAddressBook ().InsertAddress (stream->GetRemoteIdentity ());
|
||||
auto session = m_Owner.FindSession (m_ID);
|
||||
if (session)
|
||||
{
|
||||
// find more pending acceptors
|
||||
for (auto it: session->ListSockets ())
|
||||
if (it->m_SocketType == eSAMSocketTypeAcceptor)
|
||||
{
|
||||
session->localDestination->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, it, std::placeholders::_1));
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!m_IsSilent)
|
||||
{
|
||||
// get remote peer address
|
||||
|
|
@ -707,76 +708,26 @@ namespace client
|
|||
}
|
||||
|
||||
SAMSession::SAMSession (std::shared_ptr<ClientDestination> dest):
|
||||
localDestination (dest),
|
||||
m_BacklogPumper(dest->GetService())
|
||||
localDestination (dest)
|
||||
{
|
||||
PumpBacklog();
|
||||
}
|
||||
|
||||
void SAMSession::AcceptI2P(std::shared_ptr<i2p::stream::Stream> stream)
|
||||
|
||||
SAMSession::~SAMSession ()
|
||||
{
|
||||
if(!stream) return; // fail
|
||||
std::unique_lock<std::mutex> lock(m_SocketsMutex);
|
||||
if(m_Backlog.size() > SAM_MAX_ACCEPT_BACKLOG) {
|
||||
stream->Close();
|
||||
return;
|
||||
}
|
||||
m_Backlog.push_back(stream);
|
||||
}
|
||||
|
||||
void SAMSession::PumpBacklog()
|
||||
{
|
||||
// pump backlog every 100ms
|
||||
boost::posix_time::milliseconds dlt(100);
|
||||
m_BacklogPumper.expires_from_now(dlt);
|
||||
m_BacklogPumper.async_wait(std::bind(&SAMSession::HandlePumpBacklog, this, std::placeholders::_1));
|
||||
}
|
||||
|
||||
std::shared_ptr<SAMSocket> SAMSession::FindAcceptor()
|
||||
{
|
||||
for (auto & sock : m_Sockets) {
|
||||
auto t = sock->GetSocketType();
|
||||
if(t == eSAMSocketTypeAcceptor) {
|
||||
return sock;
|
||||
}
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
void SAMSession::HandlePumpBacklog(const boost::system::error_code & ec)
|
||||
{
|
||||
if(ec) return;
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(m_SocketsMutex);
|
||||
auto itr = m_Backlog.begin();
|
||||
while(itr != m_Backlog.end()) {
|
||||
auto sock = FindAcceptor();
|
||||
if (sock) {
|
||||
sock->Accept(*itr);
|
||||
itr = m_Backlog.erase(itr);
|
||||
} else {
|
||||
++itr;
|
||||
}
|
||||
}
|
||||
}
|
||||
PumpBacklog();
|
||||
CloseStreams();
|
||||
i2p::client::context.DeleteLocalDestination (localDestination);
|
||||
}
|
||||
|
||||
void SAMSession::CloseStreams ()
|
||||
{
|
||||
m_BacklogPumper.cancel();
|
||||
localDestination->GetService().post([&] () {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(m_SocketsMutex);
|
||||
for (auto& sock : m_Sockets) {
|
||||
sock->CloseStream();
|
||||
}
|
||||
for(auto & stream : m_Backlog) {
|
||||
stream->Close();
|
||||
}
|
||||
m_Sockets.clear();
|
||||
m_Backlog.clear();
|
||||
i2p::client::context.DeleteLocalDestination (localDestination);
|
||||
});
|
||||
}
|
||||
// XXX: should this be done inside locked parts?
|
||||
m_Sockets.clear();
|
||||
}
|
||||
|
||||
SAMBridge::SAMBridge (const std::string& address, int port):
|
||||
|
|
@ -887,9 +838,8 @@ namespace client
|
|||
auto session = std::make_shared<SAMSession>(localDestination);
|
||||
std::unique_lock<std::mutex> l(m_SessionsMutex);
|
||||
auto ret = m_Sessions.insert (std::make_pair(id, session));
|
||||
if (!ret.second) {
|
||||
if (!ret.second)
|
||||
LogPrint (eLogWarning, "SAM: Session ", id, " already exists");
|
||||
}
|
||||
return ret.first->second;
|
||||
}
|
||||
return nullptr;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue