diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index 1185ac5e..9eb5bed6 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -19,6 +19,26 @@ namespace i2p { namespace stream { + SendBuffer::SendBuffer (const std::vector >& bufs, size_t totalLen, SendHandler&& h): + len(totalLen), offset (0), handler(std::move (h)) + { + buf = new uint8_t[len]; + size_t offset1 = 0; + for (const auto& [b, l]: bufs) + { + if (offset1 + l <= totalLen) + { + memcpy (buf + offset1, b, l); + offset1 += l; + } + else + { + memcpy (buf + offset1, b, totalLen - offset1); + break; + } + } + } + void SendBufferQueue::Add (std::shared_ptr&& buf) { if (buf) @@ -939,13 +959,22 @@ namespace stream std::shared_ptr buffer; if (len > 0 && buf) buffer = std::make_shared(buf, len, std::move (handler)); - else if (handler) - handler(boost::system::error_code ()); - auto s = shared_from_this (); - boost::asio::post (m_Service, [s, buffer = std::move(buffer)]() mutable + else + { + if (handler) + handler(boost::system::error_code ()); + return; + } + Send (std::move (buffer)); + } + + void Stream::Send (std::shared_ptr&& buf) + { + if (!buf) return; + boost::asio::post (m_Service, [s = shared_from_this (), buf = std::move(buf)]() mutable { - if (buffer) - s->m_SendBuffer.Add (std::move(buffer)); + if (buf) + s->m_SendBuffer.Add (std::move(buf)); s->SendBuffer (); }); } diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index 8a3914aa..d7978991 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -144,6 +144,7 @@ namespace stream { buf = new uint8_t[len]; } + SendBuffer (const std::vector >& bufs, size_t totalLen, SendHandler&& h); ~SendBuffer () { delete[] buf; @@ -209,6 +210,7 @@ namespace stream void HandlePing (Packet * packet); size_t Send (const uint8_t * buf, size_t len); void AsyncSend (const uint8_t * buf, size_t len, SendHandler&& handler); + void Send (std::shared_ptr&& buf); void SendPing (); template diff --git a/libi2pd_client/BoostStream.h b/libi2pd_client/BoostStream.h index f5e51729..4a182683 100644 --- a/libi2pd_client/BoostStream.h +++ b/libi2pd_client/BoostStream.h @@ -57,15 +57,36 @@ namespace client template void async_write_some(const ConstBufferSequence& bufs, WriteHandler&& handler) { + std::vector > bufsToSend; size_t sent = 0; for (auto it = boost::asio::buffer_sequence_begin (bufs); it != boost::asio::buffer_sequence_end (bufs); it++) { const auto& buf = *it; sent += buf.size (); - m_Stream->Send ((const uint8_t *)buf.data (), buf.size ()); - // TODO: AsyncSend wiht callback for last buf, but not possible below C++23 + bufsToSend.push_back ({ (const uint8_t *)buf.data (), buf.size () }); + } - handler (boost::system::error_code (), sent); + if (sent) + { +#ifdef __cpp_lib_move_only_function // with C++23 + // we can save handler with SendBuffer + auto sendBuffer = std::make_shared(bufsToSend, sent, + [handler = std::move (handler), sent](const boost::system::error_code& ecode) mutable + { + handler (ecode, sent); + }); +#else + // we can't save handler with SendBuffer due to lack of std::move_only_function + auto sendBuffer = std::make_shared(bufsToSend, sent, nullptr); +#endif + m_Stream->Send (std::move (sendBuffer)); +#ifndef __cpp_lib_move_only_function // no std::move_only_function + // invoke handler right after send + handler (boost::system::error_code (), sent); +#endif + } + else + handler (boost::system::error_code (), 0); } private: