PurpleI2P/i2pd
1
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2025-01-22 12:24:19 +00:00
Code Issues Releases Activity

send queue for incoming I2CP messages

Browse Source
This commit is contained in:
orignal 2020-11-22 17:36:00 -05:00
parent c875ff923a
commit 771480e368
2 changed files with 82 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
libi2pd_client/I2CP.cpp
View File

@ -227,12 +227,17 @@ namespace client
I2CPSession::I2CPSession (I2CPServer& owner, std::shared_ptr<proto::socket> socket): I2CPSession::I2CPSession (I2CPServer& owner, std::shared_ptr<proto::socket> socket):
m_Owner (owner), m_Socket (socket), m_SessionID (0xFFFF), m_Owner (owner), m_Socket (socket), m_SessionID (0xFFFF),
m_MessageID (0), m_IsSendAccepted (true) m_MessageID (0), m_IsSendAccepted (true), m_IsSending (false)
{ {
} }
I2CPSession::~I2CPSession () I2CPSession::~I2CPSession ()
{ {
if (m_SendQueue)
{
for (auto& it: *m_SendQueue)
delete[] boost::asio::buffer_cast<const uint8_t *>(it);
}
} }
void I2CPSession::Start () void I2CPSession::Start ()
@ -358,25 +363,64 @@ namespace client
if (socket) if (socket)
{ {
auto l = len + I2CP_HEADER_SIZE; auto l = len + I2CP_HEADER_SIZE;
uint8_t * buf = new uint8_t[l]; uint8_t * buf = m_IsSending ? new uint8_t[l] : m_SendBuffer;
htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len); htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len);
buf[I2CP_HEADER_TYPE_OFFSET] = type; buf[I2CP_HEADER_TYPE_OFFSET] = type;
memcpy (buf + I2CP_HEADER_SIZE, payload, len); memcpy (buf + I2CP_HEADER_SIZE, payload, len);
boost::asio::async_write (*socket, boost::asio::buffer (buf, l), boost::asio::transfer_all (), if (m_IsSending)
std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (), {
std::placeholders::_1, std::placeholders::_2, buf)); if (!m_SendQueue)
m_SendQueue = std::make_shared<SendQueue::element_type> ();
if (m_SendQueue->size () > I2CP_MAX_SEND_QUEUE_SIZE)
{
LogPrint (eLogError, "I2CP: Queue size exceeds ", I2CP_MAX_SEND_QUEUE_SIZE);
return;
}
m_SendQueue->push_back ({buf, l});
}
else
{
m_IsSending = true;
boost::asio::async_write (*socket, boost::asio::buffer (m_SendBuffer, l), boost::asio::transfer_all (),
std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (),
std::placeholders::_1, std::placeholders::_2));
}
} }
else else
LogPrint (eLogError, "I2CP: Can't write to the socket"); LogPrint (eLogError, "I2CP: Can't write to the socket");
} }
void I2CPSession::HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred, const uint8_t * buf) void I2CPSession::HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred)
{ {
delete[] buf; if (ecode)
if (ecode && ecode != boost::asio::error::operation_aborted) {
Terminate (); if (ecode != boost::asio::error::operation_aborted)
Terminate ();
}
else if (m_SendQueue)
{
auto socket = m_Socket;
if (socket)
{
auto queue = m_SendQueue;
m_SendQueue = nullptr;
boost::asio::async_write (*socket, *queue, boost::asio::transfer_all (),
std::bind(&I2CPSession::HandleI2CPMessageSentQueue, shared_from_this (),
std::placeholders::_1, std::placeholders::_2, queue));
}
}
else
m_IsSending = false;
} }
void I2CPSession::HandleI2CPMessageSentQueue (const boost::system::error_code& ecode, std::size_t bytes_transferred, SendQueue queue)
{
for (auto& it: *queue)
delete[] boost::asio::buffer_cast<const uint8_t *>(it);;
HandleI2CPMessageSent (ecode, bytes_transferred);
}
std::string I2CPSession::ExtractString (const uint8_t * buf, size_t len) std::string I2CPSession::ExtractString (const uint8_t * buf, size_t len)
{ {
uint8_t l = buf[0]; uint8_t l = buf[0];
@ -810,16 +854,31 @@ namespace client
if (socket) if (socket)
{ {
auto l = len + 10 + I2CP_HEADER_SIZE; auto l = len + 10 + I2CP_HEADER_SIZE;
uint8_t * buf = new uint8_t[l]; uint8_t * buf = m_IsSending ? new uint8_t[l] : m_SendBuffer;
htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len + 10); htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len + 10);
buf[I2CP_HEADER_TYPE_OFFSET] = I2CP_MESSAGE_PAYLOAD_MESSAGE; buf[I2CP_HEADER_TYPE_OFFSET] = I2CP_MESSAGE_PAYLOAD_MESSAGE;
htobe16buf (buf + I2CP_HEADER_SIZE, m_SessionID); htobe16buf (buf + I2CP_HEADER_SIZE, m_SessionID);
htobe32buf (buf + I2CP_HEADER_SIZE + 2, m_MessageID++); htobe32buf (buf + I2CP_HEADER_SIZE + 2, m_MessageID++);
htobe32buf (buf + I2CP_HEADER_SIZE + 6, len); htobe32buf (buf + I2CP_HEADER_SIZE + 6, len);
memcpy (buf + I2CP_HEADER_SIZE + 10, payload, len); memcpy (buf + I2CP_HEADER_SIZE + 10, payload, len);
boost::asio::async_write (*socket, boost::asio::buffer (buf, l), boost::asio::transfer_all (), if (m_IsSending)
std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (), {
std::placeholders::_1, std::placeholders::_2, buf)); if (!m_SendQueue)
m_SendQueue = std::make_shared<SendQueue::element_type> ();
if (m_SendQueue->size () > I2CP_MAX_SEND_QUEUE_SIZE)
{
LogPrint (eLogError, "I2CP: Queue size exceeds ", I2CP_MAX_SEND_QUEUE_SIZE);
return;
}
m_SendQueue->push_back ({buf, l});
}
else
{
m_IsSending = true;
boost::asio::async_write (*socket, boost::asio::buffer (m_SendBuffer, l), boost::asio::transfer_all (),
std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (),
std::placeholders::_1, std::placeholders::_2));
}
} }
else else
LogPrint (eLogError, "I2CP: Can't write to the socket"); LogPrint (eLogError, "I2CP: Can't write to the socket");

11
libi2pd_client/I2CP.h
View File

@ -25,6 +25,7 @@ namespace client
const uint8_t I2CP_PROTOCOL_BYTE = 0x2A; const uint8_t I2CP_PROTOCOL_BYTE = 0x2A;
const size_t I2CP_SESSION_BUFFER_SIZE = 4096; const size_t I2CP_SESSION_BUFFER_SIZE = 4096;
const size_t I2CP_MAX_MESSAGE_LENGTH = 65535; const size_t I2CP_MAX_MESSAGE_LENGTH = 65535;
const size_t I2CP_MAX_SEND_QUEUE_SIZE = 256;
const size_t I2CP_HEADER_LENGTH_OFFSET = 0; const size_t I2CP_HEADER_LENGTH_OFFSET = 0;
const size_t I2CP_HEADER_TYPE_OFFSET = I2CP_HEADER_LENGTH_OFFSET + 4; const size_t I2CP_HEADER_TYPE_OFFSET = I2CP_HEADER_LENGTH_OFFSET + 4;
@ -122,6 +123,8 @@ namespace client
class I2CPServer; class I2CPServer;
class I2CPSession: public std::enable_shared_from_this<I2CPSession> class I2CPSession: public std::enable_shared_from_this<I2CPSession>
{ {
typedef std::shared_ptr<std::vector<boost::asio::const_buffer> > SendQueue;
public: public:
#ifdef ANDROID #ifdef ANDROID
@ -167,7 +170,8 @@ namespace client
void HandleMessage (); 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);
void HandleI2CPMessageSentQueue (const boost::system::error_code& ecode, std::size_t bytes_transferred, SendQueue queue);
std::string ExtractString (const uint8_t * buf, size_t len); std::string ExtractString (const uint8_t * buf, size_t len);
size_t PutString (uint8_t * buf, size_t len, const std::string& str); size_t PutString (uint8_t * buf, size_t len, const std::string& str);
void ExtractMapping (const uint8_t * buf, size_t len, std::map<std::string, std::string>& mapping); void ExtractMapping (const uint8_t * buf, size_t len, std::map<std::string, std::string>& mapping);
@ -186,6 +190,11 @@ namespace client
uint16_t m_SessionID; uint16_t m_SessionID;
uint32_t m_MessageID; uint32_t m_MessageID;
bool m_IsSendAccepted; bool m_IsSendAccepted;
// to client
bool m_IsSending;
uint8_t m_SendBuffer[I2CP_MAX_MESSAGE_LENGTH];
SendQueue m_SendQueue;
}; };
typedef void (I2CPSession::*I2CPMessageHandler)(const uint8_t * buf, size_t len); typedef void (I2CPSession::*I2CPMessageHandler)(const uint8_t * buf, size_t len);