Browse Source

Add support for multiple udp server tunnels on one destionation

pull/2010/head
weko 4 months ago
parent
commit
5b93558bd0
  1. 68
      libi2pd/Datagram.cpp
  2. 26
      libi2pd/Datagram.h
  3. 2
      libi2pd_client/ClientContext.cpp
  4. 53
      libi2pd_client/SAM.cpp
  5. 46
      libi2pd_client/UDPTunnel.cpp
  6. 3
      libi2pd_client/UDPTunnel.h

68
libi2pd/Datagram.cpp

@ -19,7 +19,7 @@ namespace i2p
namespace datagram namespace datagram
{ {
DatagramDestination::DatagramDestination (std::shared_ptr<i2p::client::ClientDestination> owner, bool gzip): DatagramDestination::DatagramDestination (std::shared_ptr<i2p::client::ClientDestination> owner, bool gzip):
m_Owner (owner), m_Receiver (nullptr), m_RawReceiver (nullptr), m_Gzip (gzip) m_Owner (owner), m_DefaultReceiver (nullptr), m_DefaultRawReceiver (nullptr), m_Gzip (gzip)
{ {
if (m_Gzip) if (m_Gzip)
m_Deflator.reset (new i2p::data::GzipDeflator); m_Deflator.reset (new i2p::data::GzipDeflator);
@ -119,19 +119,79 @@ namespace datagram
void DatagramDestination::HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) void DatagramDestination::HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len)
{ {
if (m_RawReceiver) auto r = FindRawReceiver(toPort);
m_RawReceiver (fromPort, toPort, buf, len);
if (r)
r (fromPort, toPort, buf, len);
else else
LogPrint (eLogWarning, "DatagramDestination: no receiver for raw datagram"); LogPrint (eLogWarning, "DatagramDestination: no receiver for raw datagram");
} }
void DatagramDestination::SetReceiver (const Receiver& receiver, uint16_t port)
{
std::lock_guard<std::mutex> lock(m_ReceiversMutex);
m_ReceiversByPorts[port] = receiver;
if (!m_DefaultReceiver) {
m_DefaultReceiver = receiver;
m_DefaultReceiverPort = port;
}
}
void DatagramDestination::ResetReceiver (uint16_t port)
{
std::lock_guard<std::mutex> lock(m_ReceiversMutex);
m_ReceiversByPorts.erase (port);
if (m_DefaultReceiverPort == port) {
m_DefaultReceiver = nullptr;
m_DefaultReceiverPort = 0;
}
}
void DatagramDestination::SetRawReceiver (const RawReceiver& receiver, uint16_t port)
{
std::lock_guard<std::mutex> lock(m_RawReceiversMutex);
m_RawReceiversByPorts[port] = receiver;
if (!m_DefaultRawReceiver) {
m_DefaultRawReceiver = receiver;
m_DefaultRawReceiverPort = port;
}
};
void DatagramDestination::ResetRawReceiver (uint16_t port)
{
std::lock_guard<std::mutex> lock(m_RawReceiversMutex);
m_RawReceiversByPorts.erase (port);
if (m_DefaultRawReceiverPort == port) {
m_DefaultRawReceiver = nullptr;
m_DefaultRawReceiverPort = 0;
}
}
DatagramDestination::Receiver DatagramDestination::FindReceiver(uint16_t port) DatagramDestination::Receiver DatagramDestination::FindReceiver(uint16_t port)
{ {
std::lock_guard<std::mutex> lock(m_ReceiversMutex); std::lock_guard<std::mutex> lock(m_ReceiversMutex);
Receiver r = m_Receiver; Receiver r = nullptr;
auto itr = m_ReceiversByPorts.find(port); auto itr = m_ReceiversByPorts.find(port);
if (itr != m_ReceiversByPorts.end()) if (itr != m_ReceiversByPorts.end())
r = itr->second; r = itr->second;
else {
r = m_DefaultReceiver;
}
return r;
}
DatagramDestination::RawReceiver DatagramDestination::FindRawReceiver(uint16_t port)
{
std::lock_guard<std::mutex> lock(m_RawReceiversMutex);
RawReceiver r = nullptr;
auto itr = m_RawReceiversByPorts.find(port);
if (itr != m_RawReceiversByPorts.end())
r = itr->second;
else {
r = m_DefaultRawReceiver;
}
return r; return r;
} }

26
libi2pd/Datagram.h

@ -126,14 +126,12 @@ namespace datagram
void HandleDataMessagePayload (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len, bool isRaw = false); void HandleDataMessagePayload (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len, bool isRaw = false);
void SetReceiver (const Receiver& receiver) { m_Receiver = receiver; };
void ResetReceiver () { m_Receiver = nullptr; };
void SetReceiver (const Receiver& receiver, uint16_t port) { std::lock_guard<std::mutex> lock(m_ReceiversMutex); m_ReceiversByPorts[port] = receiver; }; void SetReceiver (const Receiver& receiver, uint16_t port);
void ResetReceiver (uint16_t port) { std::lock_guard<std::mutex> lock(m_ReceiversMutex); m_ReceiversByPorts.erase (port); }; void ResetReceiver (uint16_t port);
void SetRawReceiver (const RawReceiver& receiver) { m_RawReceiver = receiver; }; void SetRawReceiver (const RawReceiver& receiver, uint16_t port);
void ResetRawReceiver () { m_RawReceiver = nullptr; }; void ResetRawReceiver (uint16_t port);
std::shared_ptr<DatagramSession::Info> GetInfoForRemote(const i2p::data::IdentHash & remote); std::shared_ptr<DatagramSession::Info> GetInfoForRemote(const i2p::data::IdentHash & remote);
@ -150,20 +148,26 @@ namespace datagram
void HandleDatagram (uint16_t fromPort, uint16_t toPort, uint8_t *const& buf, size_t len); void HandleDatagram (uint16_t fromPort, uint16_t toPort, uint8_t *const& buf, size_t len);
void HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); void HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len);
/** find a receiver by port, if none by port is found try default receiver, otherwise returns nullptr */
Receiver FindReceiver(uint16_t port); Receiver FindReceiver(uint16_t port);
RawReceiver FindRawReceiver(uint16_t port);
private: private:
std::shared_ptr<i2p::client::ClientDestination> m_Owner; std::shared_ptr<i2p::client::ClientDestination> m_Owner;
Receiver m_Receiver; // default
RawReceiver m_RawReceiver; // default
bool m_Gzip; // gzip compression of data messages
std::mutex m_SessionsMutex; std::mutex m_SessionsMutex;
std::map<i2p::data::IdentHash, DatagramSession_ptr > m_Sessions; std::map<i2p::data::IdentHash, DatagramSession_ptr > m_Sessions;
Receiver m_DefaultReceiver;
RawReceiver m_DefaultRawReceiver;
uint16_t m_DefaultReceiverPort;
uint16_t m_DefaultRawReceiverPort;
std::mutex m_ReceiversMutex; std::mutex m_ReceiversMutex;
std::map<uint16_t, Receiver> m_ReceiversByPorts; std::mutex m_RawReceiversMutex;
std::unordered_map<uint16_t, Receiver> m_ReceiversByPorts;
std::unordered_map<uint16_t, RawReceiver> m_RawReceiversByPorts;
bool m_Gzip; // gzip compression of data messages
i2p::data::GzipInflator m_Inflator; i2p::data::GzipInflator m_Inflator;
std::unique_ptr<i2p::data::GzipDeflator> m_Deflator; std::unique_ptr<i2p::data::GzipDeflator> m_Deflator;
std::vector<uint8_t> m_From, m_Signature; std::vector<uint8_t> m_From, m_Signature;

2
libi2pd_client/ClientContext.cpp

@ -775,7 +775,7 @@ namespace client
address = "127.0.0.1"; address = "127.0.0.1";
} }
auto localAddress = boost::asio::ip::address::from_string(address); auto localAddress = boost::asio::ip::address::from_string(address);
auto serverTunnel = std::make_shared<I2PUDPServerTunnel>(name, localDestination, localAddress, endpoint, port, gzip); auto serverTunnel = std::make_shared<I2PUDPServerTunnel>(name, localDestination, localAddress, endpoint, inPort, gzip);
if(!isUniqueLocal) if(!isUniqueLocal)
{ {
LogPrint(eLogInfo, "Clients: Disabling loopback address mapping"); LogPrint(eLogInfo, "Clients: Disabling loopback address mapping");

53
libi2pd_client/SAM.cpp

@ -415,12 +415,17 @@ namespace client
{ {
session->UDPEndpoint = forward; session->UDPEndpoint = forward;
auto dest = session->GetLocalDestination ()->CreateDatagramDestination (); auto dest = session->GetLocalDestination ()->CreateDatagramDestination ();
auto port = std::stoi(params[SAM_PARAM_PORT]);
if (type == eSAMSessionTypeDatagram) if (type == eSAMSessionTypeDatagram)
dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, shared_from_this (), dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, shared_from_this (),
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5)); std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5),
port
);
else // raw else // raw
dest->SetRawReceiver (std::bind (&SAMSocket::HandleI2PRawDatagramReceive, shared_from_this (), dest->SetRawReceiver (std::bind (&SAMSocket::HandleI2PRawDatagramReceive, shared_from_this (),
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4),
port
);
} }
if (session->GetLocalDestination ()->IsReady ()) if (session->GetLocalDestination ()->IsReady ())
@ -524,7 +529,7 @@ namespace client
if (addr->IsIdentHash ()) if (addr->IsIdentHash ())
{ {
if (session->GetLocalDestination ()->GetIdentHash () != addr->identHash) if (session->GetLocalDestination ()->GetIdentHash () != addr->identHash)
{ {
auto leaseSet = session->GetLocalDestination ()->FindLeaseSet(addr->identHash); auto leaseSet = session->GetLocalDestination ()->FindLeaseSet(addr->identHash);
if (leaseSet) if (leaseSet)
Connect(leaseSet, session); Connect(leaseSet, session);
@ -556,7 +561,7 @@ namespace client
if (session) if (session)
{ {
if (session->GetLocalDestination ()->SupportsEncryptionType (remote->GetEncryptionType ())) if (session->GetLocalDestination ()->SupportsEncryptionType (remote->GetEncryptionType ()))
{ {
m_SocketType = eSAMSocketTypeStream; m_SocketType = eSAMSocketTypeStream;
m_Stream = session->GetLocalDestination ()->CreateStream (remote); m_Stream = session->GetLocalDestination ()->CreateStream (remote);
if (m_Stream) if (m_Stream)
@ -570,7 +575,7 @@ namespace client
SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true);
} }
else else
SendStreamCantReachPeer ("Incompatible crypto"); SendStreamCantReachPeer ("Incompatible crypto");
} }
else else
SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true);
@ -583,7 +588,7 @@ namespace client
else else
{ {
LogPrint (eLogError, "SAM: Destination to connect not found"); LogPrint (eLogError, "SAM: Destination to connect not found");
SendStreamCantReachPeer ("LeaseSet not found"); SendStreamCantReachPeer ("LeaseSet not found");
} }
} }
@ -612,27 +617,27 @@ namespace client
session->GetLocalDestination ()->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, shared_from_this (), std::placeholders::_1)); session->GetLocalDestination ()->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, shared_from_this (), std::placeholders::_1));
} }
else else
{ {
auto ts = i2p::util::GetSecondsSinceEpoch (); auto ts = i2p::util::GetSecondsSinceEpoch ();
while (!session->acceptQueue.empty () && session->acceptQueue.front ().second + SAM_SESSION_MAX_ACCEPT_INTERVAL > ts) while (!session->acceptQueue.empty () && session->acceptQueue.front ().second + SAM_SESSION_MAX_ACCEPT_INTERVAL > ts)
{ {
auto socket = session->acceptQueue.front ().first; auto socket = session->acceptQueue.front ().first;
session->acceptQueue.pop_front (); session->acceptQueue.pop_front ();
if (socket) if (socket)
m_Owner.GetService ().post (std::bind(&SAMSocket::TerminateClose, socket)); m_Owner.GetService ().post (std::bind(&SAMSocket::TerminateClose, socket));
} }
if (session->acceptQueue.size () < SAM_SESSION_MAX_ACCEPT_QUEUE_SIZE) if (session->acceptQueue.size () < SAM_SESSION_MAX_ACCEPT_QUEUE_SIZE)
{ {
// already accepting, queue up // already accepting, queue up
SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false); SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false);
session->acceptQueue.push_back (std::make_pair(shared_from_this(), ts)); session->acceptQueue.push_back (std::make_pair(shared_from_this(), ts));
} }
else else
{ {
LogPrint (eLogInfo, "SAM: Session ", m_ID, " accept queue is full ", session->acceptQueue.size ()); LogPrint (eLogInfo, "SAM: Session ", m_ID, " accept queue is full ", session->acceptQueue.size ());
SendStreamI2PError ("Already accepting"); SendStreamI2PError ("Already accepting");
} }
} }
} }
else else
SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true);
@ -875,8 +880,8 @@ namespace client
size_t len = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, reply, msg.c_str()); size_t len = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, reply, msg.c_str());
#endif #endif
SendMessageReply (m_Buffer, len, true); SendMessageReply (m_Buffer, len, true);
} }
void SAMSocket::SendSessionI2PError(const std::string & msg) void SAMSocket::SendSessionI2PError(const std::string & msg)
{ {
LogPrint (eLogError, "SAM: Session I2P error: ", msg); LogPrint (eLogError, "SAM: Session I2P error: ", msg);
@ -886,14 +891,14 @@ namespace client
void SAMSocket::SendStreamI2PError(const std::string & msg) void SAMSocket::SendStreamI2PError(const std::string & msg)
{ {
LogPrint (eLogError, "SAM: Stream I2P error: ", msg); LogPrint (eLogError, "SAM: Stream I2P error: ", msg);
SendReplyWithMessage (SAM_STREAM_STATUS_I2P_ERROR, msg); SendReplyWithMessage (SAM_STREAM_STATUS_I2P_ERROR, msg);
} }
void SAMSocket::SendStreamCantReachPeer(const std::string & msg) void SAMSocket::SendStreamCantReachPeer(const std::string & msg)
{ {
SendReplyWithMessage (SAM_STREAM_STATUS_CANT_REACH_PEER, msg); SendReplyWithMessage (SAM_STREAM_STATUS_CANT_REACH_PEER, msg);
} }
void SAMSocket::HandleNamingLookupLeaseSetRequestComplete (std::shared_ptr<i2p::data::LeaseSet> leaseSet, std::string name) void SAMSocket::HandleNamingLookupLeaseSetRequestComplete (std::shared_ptr<i2p::data::LeaseSet> leaseSet, std::string name)
{ {
if (leaseSet) if (leaseSet)
@ -1093,14 +1098,14 @@ namespace client
// pending acceptors // pending acceptors
auto ts = i2p::util::GetSecondsSinceEpoch (); auto ts = i2p::util::GetSecondsSinceEpoch ();
while (!session->acceptQueue.empty () && session->acceptQueue.front ().second + SAM_SESSION_MAX_ACCEPT_INTERVAL > ts) while (!session->acceptQueue.empty () && session->acceptQueue.front ().second + SAM_SESSION_MAX_ACCEPT_INTERVAL > ts)
{ {
auto socket = session->acceptQueue.front ().first; auto socket = session->acceptQueue.front ().first;
session->acceptQueue.pop_front (); session->acceptQueue.pop_front ();
if (socket) if (socket)
m_Owner.GetService ().post (std::bind(&SAMSocket::TerminateClose, socket)); m_Owner.GetService ().post (std::bind(&SAMSocket::TerminateClose, socket));
} }
if (!session->acceptQueue.empty ()) if (!session->acceptQueue.empty ())
{ {
auto socket = session->acceptQueue.front ().first; auto socket = session->acceptQueue.front ().first;
session->acceptQueue.pop_front (); session->acceptQueue.pop_front ();
if (socket && socket->GetSocketType () == eSAMSocketTypeAcceptor) if (socket && socket->GetSocketType () == eSAMSocketTypeAcceptor)
@ -1108,7 +1113,7 @@ namespace client
socket->m_IsAccepting = true; socket->m_IsAccepting = true;
session->GetLocalDestination ()->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, socket, std::placeholders::_1)); session->GetLocalDestination ()->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, socket, std::placeholders::_1));
} }
} }
} }
if (!m_IsSilent) if (!m_IsSilent)
{ {

46
libi2pd_client/UDPTunnel.cpp

@ -32,7 +32,7 @@ namespace client
auto it = m_Sessions.find (GetSessionIndex (fromPort, toPort)); auto it = m_Sessions.find (GetSessionIndex (fromPort, toPort));
if (it != m_Sessions.end ()) if (it != m_Sessions.end ())
m_LastSession = it->second; m_LastSession = it->second;
else else
m_LastSession = nullptr; m_LastSession = nullptr;
} }
if (m_LastSession) if (m_LastSession)
@ -47,7 +47,7 @@ namespace client
std::lock_guard<std::mutex> lock(m_SessionsMutex); std::lock_guard<std::mutex> lock(m_SessionsMutex);
uint64_t now = i2p::util::GetMillisecondsSinceEpoch(); uint64_t now = i2p::util::GetMillisecondsSinceEpoch();
auto itr = m_Sessions.begin(); auto itr = m_Sessions.begin();
while(itr != m_Sessions.end()) while(itr != m_Sessions.end())
{ {
if(now - itr->second->LastActivity >= delta ) if(now - itr->second->LastActivity >= delta )
itr = m_Sessions.erase(itr); itr = m_Sessions.erase(itr);
@ -88,10 +88,10 @@ namespace client
{ {
LogPrint(eLogWarning, "UDPServer: Session with from ", remotePort, " and to ", localPort, " ports already exists. But from differend address. Removed"); LogPrint(eLogWarning, "UDPServer: Session with from ", remotePort, " and to ", localPort, " ports already exists. But from differend address. Removed");
m_Sessions.erase (it); m_Sessions.erase (it);
} }
} }
} }
boost::asio::ip::address addr; boost::asio::ip::address addr;
/** create new udp session */ /** create new udp session */
if(m_IsUniqueLocal && m_LocalAddress.is_loopback()) if(m_IsUniqueLocal && m_LocalAddress.is_loopback())
@ -102,7 +102,7 @@ namespace client
else else
addr = m_LocalAddress; addr = m_LocalAddress;
auto s = std::make_shared<UDPSession>(boost::asio::ip::udp::endpoint(addr, 0), auto s = std::make_shared<UDPSession>(boost::asio::ip::udp::endpoint(addr, 0),
m_LocalDest, m_RemoteEndpoint, ih, localPort, remotePort); m_LocalDest, m_RemoteEndpoint, ih, localPort, remotePort);
std::lock_guard<std::mutex> lock(m_SessionsMutex); std::lock_guard<std::mutex> lock(m_SessionsMutex);
m_Sessions.emplace (idx, s); m_Sessions.emplace (idx, s);
@ -163,9 +163,9 @@ namespace client
} }
I2PUDPServerTunnel::I2PUDPServerTunnel (const std::string & name, std::shared_ptr<i2p::client::ClientDestination> localDestination, I2PUDPServerTunnel::I2PUDPServerTunnel (const std::string & name, std::shared_ptr<i2p::client::ClientDestination> localDestination,
const boost::asio::ip::address& localAddress, const boost::asio::ip::udp::endpoint& forwardTo, uint16_t port, bool gzip) : const boost::asio::ip::address& localAddress, const boost::asio::ip::udp::endpoint& forwardTo, uint16_t inPort, bool gzip) :
m_IsUniqueLocal (true), m_Name (name), m_LocalAddress (localAddress), m_IsUniqueLocal (true), m_Name (name), m_LocalAddress (localAddress),
m_RemoteEndpoint (forwardTo), m_LocalDest (localDestination), m_Gzip (gzip) m_RemoteEndpoint (forwardTo), m_LocalDest (localDestination), m_inPort(inPort), m_Gzip (gzip)
{ {
} }
@ -179,14 +179,23 @@ namespace client
m_LocalDest->Start (); m_LocalDest->Start ();
auto dgram = m_LocalDest->CreateDatagramDestination (m_Gzip); auto dgram = m_LocalDest->CreateDatagramDestination (m_Gzip);
dgram->SetReceiver (std::bind (&I2PUDPServerTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5)); dgram->SetReceiver (
dgram->SetRawReceiver (std::bind (&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); std::bind (&I2PUDPServerTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5),
m_inPort
);
dgram->SetRawReceiver (
std::bind (&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4),
m_inPort
);
} }
void I2PUDPServerTunnel::Stop () void I2PUDPServerTunnel::Stop ()
{ {
auto dgram = m_LocalDest->GetDatagramDestination (); auto dgram = m_LocalDest->GetDatagramDestination ();
if (dgram) dgram->ResetReceiver (); if (dgram) {
dgram->ResetReceiver (m_inPort);
dgram->ResetRawReceiver (m_inPort);
}
} }
std::vector<std::shared_ptr<DatagramSessionInfo> > I2PUDPServerTunnel::GetSessions () std::vector<std::shared_ptr<DatagramSessionInfo> > I2PUDPServerTunnel::GetSessions ()
@ -240,9 +249,13 @@ namespace client
dgram->SetReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2P, this, dgram->SetReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2P, this,
std::placeholders::_1, std::placeholders::_2, std::placeholders::_1, std::placeholders::_2,
std::placeholders::_3, std::placeholders::_4, std::placeholders::_3, std::placeholders::_4,
std::placeholders::_5)); std::placeholders::_5),
RemotePort
);
dgram->SetRawReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2PRaw, this, dgram->SetRawReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2PRaw, this,
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4),
RemotePort
);
m_LocalDest->Start (); m_LocalDest->Start ();
if (m_ResolveThread == nullptr) if (m_ResolveThread == nullptr)
@ -253,7 +266,10 @@ namespace client
void I2PUDPClientTunnel::Stop () void I2PUDPClientTunnel::Stop ()
{ {
auto dgram = m_LocalDest->GetDatagramDestination (); auto dgram = m_LocalDest->GetDatagramDestination ();
if (dgram) dgram->ResetReceiver (); if (dgram) {
dgram->ResetReceiver (RemotePort);
dgram->ResetRawReceiver (RemotePort);
}
m_cancel_resolve = true; m_cancel_resolve = true;
m_Sessions.clear(); m_Sessions.clear();

3
libi2pd_client/UDPTunnel.h

@ -105,7 +105,7 @@ namespace client
void HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); void HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len);
UDPSessionPtr ObtainUDPSession (const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort); UDPSessionPtr ObtainUDPSession (const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort);
uint32_t GetSessionIndex (uint16_t fromPort, uint16_t toPort) const { return ((uint32_t)fromPort << 16) + toPort; } uint32_t GetSessionIndex (uint16_t fromPort, uint16_t toPort) const { return ((uint32_t)fromPort << 16) + toPort; }
private: private:
bool m_IsUniqueLocal; bool m_IsUniqueLocal;
@ -116,6 +116,7 @@ namespace client
std::unordered_map<uint32_t, UDPSessionPtr> m_Sessions; // (from port, to port)->session std::unordered_map<uint32_t, UDPSessionPtr> m_Sessions; // (from port, to port)->session
std::shared_ptr<i2p::client::ClientDestination> m_LocalDest; std::shared_ptr<i2p::client::ClientDestination> m_LocalDest;
UDPSessionPtr m_LastSession; UDPSessionPtr m_LastSession;
uint16_t m_inPort;
bool m_Gzip; bool m_Gzip;
public: public:

Loading…
Cancel
Save