PurpleI2P/i2pd
1
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2025-01-27 09:04:15 +00:00
Code Issues Releases Activity

read header and payload separately

Browse Source
This commit is contained in:
orignal 2016-06-13 13:20:21 -04:00
parent 68482d712b
commit fb94d6ae2b
2 changed files with 53 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

137
I2CP.cpp
View File

@ -115,15 +115,14 @@ namespace client
} }
I2CPSession::I2CPSession (I2CPServer& owner, std::shared_ptr<boost::asio::ip::tcp::socket> socket): I2CPSession::I2CPSession (I2CPServer& owner, std::shared_ptr<boost::asio::ip::tcp::socket> socket):
m_Owner (owner), m_Socket (socket), m_Owner (owner), m_Socket (socket), m_Payload (nullptr),
m_NextMessage (nullptr), m_NextMessageLen (0), m_NextMessageOffset (0),
m_SessionID (0xFFFF), m_MessageID (0), m_IsSendAccepted (true) m_SessionID (0xFFFF), m_MessageID (0), m_IsSendAccepted (true)
{ {
} }
I2CPSession::~I2CPSession () I2CPSession::~I2CPSession ()
{ {
delete[] m_NextMessage; delete[] m_Payload;
} }
void I2CPSession::Start () void I2CPSession::Start ()
@ -141,117 +140,72 @@ namespace client
if (m_Socket) if (m_Socket)
{ {
auto s = shared_from_this (); auto s = shared_from_this ();
m_Socket->async_read_some (boost::asio::buffer (m_Buffer, 1), m_Socket->async_read_some (boost::asio::buffer (m_Header, 1),
[s](const boost::system::error_code& ecode, std::size_t bytes_transferred) [s](const boost::system::error_code& ecode, std::size_t bytes_transferred)
{ {
if (!ecode && bytes_transferred > 0 && s->m_Buffer[0] == I2CP_PROTOCOL_BYTE) if (!ecode && bytes_transferred > 0 && s->m_Header[0] == I2CP_PROTOCOL_BYTE)
s->Receive (); s->ReceiveHeader ();
else else
s->Terminate (); s->Terminate ();
}); });
} }
} }
void I2CPSession::Receive () void I2CPSession::ReceiveHeader ()
{ {
m_Socket->async_read_some (boost::asio::buffer (m_Buffer, I2CP_SESSION_BUFFER_SIZE), boost::asio::async_read (*m_Socket, boost::asio::buffer (m_Header, I2CP_HEADER_SIZE),
std::bind (&I2CPSession::HandleReceived, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); boost::asio::transfer_all (),
std::bind (&I2CPSession::HandleReceivedHeader, shared_from_this (), std::placeholders::_1, std::placeholders::_2));
} }
void I2CPSession::HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) void I2CPSession::HandleReceivedHeader (const boost::system::error_code& ecode, std::size_t bytes_transferred)
{ {
if (ecode) if (ecode)
Terminate (); Terminate ();
else else
{ {
size_t offset = 0; // from m_Buffer m_PayloadLen = bufbe32toh (m_Header + I2CP_HEADER_LENGTH_OFFSET);
if (m_NextMessage) if (m_PayloadLen > 0)
{ {
if (!m_NextMessageLen) // we didn't receive header yet m_Payload = new uint8_t[m_PayloadLen];
{ ReceivePayload ();
if (m_NextMessageOffset + bytes_transferred < I2CP_HEADER_SIZE) }
{ else // no following payload
// still no complete header
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer, bytes_transferred);
m_NextMessageOffset += bytes_transferred;
offset = bytes_transferred;
}
else
{
// we know message length now
offset = I2CP_HEADER_SIZE - m_NextMessageOffset;
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer, offset);
m_NextMessageLen = bufbe32toh (m_NextMessage + I2CP_HEADER_LENGTH_OFFSET) + I2CP_HEADER_SIZE;
m_NextMessageOffset = I2CP_HEADER_SIZE;
}
}
if (offset < bytes_transferred)
{
auto msgRemainingLen = m_NextMessageLen - m_NextMessageOffset;
auto bufRemainingLen = bytes_transferred - offset;
if (bufRemainingLen < msgRemainingLen)
{
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer + offset, bufRemainingLen);
m_NextMessageOffset += bufRemainingLen;
offset += bufRemainingLen;
}
else
{
// m_NextMessage complete
offset += msgRemainingLen;
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer + offset, msgRemainingLen);
HandleNextMessage (m_NextMessage);
delete[] m_NextMessage;
m_NextMessage = nullptr;
}
}
}
// process the rest
while (offset < bytes_transferred)
{ {
if (bytes_transferred - offset < I2CP_HEADER_SIZE) HandleMessage ();
{ ReceiveHeader (); // next message
// we don't have message header yet }
m_NextMessage = new uint8_t[0xFFFF]; // allocate 64K
m_NextMessageLen = 0; // we must set message length later
m_NextMessageOffset = bytes_transferred - offset;
memcpy (m_NextMessage, m_Buffer + offset, m_NextMessageOffset); // just copy it
break;
}
auto msgLen = bufbe32toh (m_Buffer + offset + I2CP_HEADER_LENGTH_OFFSET) + I2CP_HEADER_SIZE;
if (msgLen > 0xFFFF) // 64K
{
LogPrint (eLogError, "I2CP: message length ", msgLen, " exceeds 64K. Terminated");
Terminate ();
return;
}
if (msgLen <= bytes_transferred - offset)
{
HandleNextMessage (m_Buffer + offset);
offset += msgLen;
}
else
{
m_NextMessageLen = msgLen;
m_NextMessageOffset = bytes_transferred - offset;
m_NextMessage = new uint8_t[m_NextMessageLen];
memcpy (m_NextMessage, m_Buffer + offset, m_NextMessageOffset);
offset = bytes_transferred;
}
}
Receive ();
} }
} }
void I2CPSession::HandleNextMessage (const uint8_t * buf) void I2CPSession::ReceivePayload ()
{ {
auto handler = m_Owner.GetMessagesHandlers ()[buf[I2CP_HEADER_TYPE_OFFSET]]; boost::asio::async_read (*m_Socket, boost::asio::buffer (m_Payload, m_PayloadLen),
if (handler) boost::asio::transfer_all (),
(this->*handler)(buf + I2CP_HEADER_SIZE, bufbe32toh (buf + I2CP_HEADER_LENGTH_OFFSET)); std::bind (&I2CPSession::HandleReceivedPayload, shared_from_this (), std::placeholders::_1, std::placeholders::_2));
}
void I2CPSession::HandleReceivedPayload (const boost::system::error_code& ecode, std::size_t bytes_transferred)
{
if (ecode)
Terminate ();
else else
LogPrint (eLogError, "I2CP: Unknown I2CP messsage ", (int)buf[I2CP_HEADER_TYPE_OFFSET]); {
HandleMessage ();
delete[] m_Payload;
m_Payload = nullptr;
m_PayloadLen = 0;
ReceiveHeader (); // next message
}
}
void I2CPSession::HandleMessage ()
{
auto handler = m_Owner.GetMessagesHandlers ()[m_Header[I2CP_HEADER_TYPE_OFFSET]];
if (handler)
(this->*handler)(m_Payload, m_PayloadLen);
else
LogPrint (eLogError, "I2CP: Unknown I2CP messsage ", (int)m_Header[I2CP_HEADER_TYPE_OFFSET]);
} }
void I2CPSession::Terminate () void I2CPSession::Terminate ()
@ -267,6 +221,7 @@ namespace client
m_Socket = nullptr; m_Socket = nullptr;
} }
m_Owner.RemoveSession (GetSessionID ()); m_Owner.RemoveSession (GetSessionID ());
LogPrint (eLogDebug, "I2CP: session ", m_SessionID, " terminated");
} }
void I2CPSession::SendI2CPMessage (uint8_t type, const uint8_t * payload, size_t len) void I2CPSession::SendI2CPMessage (uint8_t type, const uint8_t * payload, size_t len)

12
I2CP.h
View File

@ -126,9 +126,11 @@ namespace client
private: private:
void ReadProtocolByte (); void ReadProtocolByte ();
void Receive (); void ReceiveHeader ();
void HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleReceivedHeader (const boost::system::error_code& ecode, std::size_t bytes_transferred);
void HandleNextMessage (const uint8_t * buf); void ReceivePayload ();
void HandleReceivedPayload (const boost::system::error_code& ecode, std::size_t bytes_transferred);
void HandleMessage ();
void Terminate (); void Terminate ();
void HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred, const uint8_t * buf); void HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred, const uint8_t * buf);
@ -143,8 +145,8 @@ namespace client
I2CPServer& m_Owner; I2CPServer& m_Owner;
std::shared_ptr<boost::asio::ip::tcp::socket> m_Socket; std::shared_ptr<boost::asio::ip::tcp::socket> m_Socket;
uint8_t m_Buffer[I2CP_SESSION_BUFFER_SIZE], * m_NextMessage; uint8_t m_Header[I2CP_HEADER_SIZE], * m_Payload;
size_t m_NextMessageLen, m_NextMessageOffset; size_t m_PayloadLen;
std::shared_ptr<I2CPDestination> m_Destination; std::shared_ptr<I2CPDestination> m_Destination;
uint16_t m_SessionID; uint16_t m_SessionID;