From db701c7928b7e20eaedbf51ba49bdfbaf51e156a Mon Sep 17 00:00:00 2001 From: orignal Date: Wed, 22 Jul 2026 14:33:26 -0400 Subject: [PATCH] moved BoostAsyncStream to BoostStream.h --- libi2pd/Streaming.h | 57 ----------------------- libi2pd_client/BoostStream.h | 89 ++++++++++++++++++++++++++++++++++++ libi2pd_client/Torrents.cpp | 4 +- libi2pd_client/Torrents.h | 3 +- 4 files changed, 93 insertions(+), 60 deletions(-) create mode 100644 libi2pd_client/BoostStream.h diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index f6e83623..8a3914aa 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -438,63 +438,6 @@ namespace stream } } } - -//------------------------------------------------- - - class BoostAsyncStream // for boost::beast and boost::asio - { - public: - - using executor_type = boost::asio::any_io_executor; - - BoostAsyncStream (std::shared_ptr stream): m_Stream (stream) {} - - // AsyncStream - executor_type get_executor() noexcept { return m_Stream->GetService ().get_executor(); } - - // AsyncReadStream - template - void async_read_some(const MutableBufferSequence& bufs, ReadHandler&& handler) - { - size_t received = 0; - for (auto it = boost::asio::buffer_sequence_begin (bufs); it != boost::asio::buffer_sequence_end (bufs); it++) - { - auto len = m_Stream->ReadSome ((uint8_t *)it->data (), it->size ()); - received += len; - if (received < it->size ()) break; - } - if (received > 0) // we have some data - handler (boost::system::error_code (), received); - else if (bufs.size () > 0) // wait for incoming data - m_Stream->AsyncReceive (*boost::asio::buffer_sequence_begin (bufs), - [handler = std::move (handler)](const boost::system::error_code& ecode, size_t bytes_transferred) mutable - { - handler (boost::system::error_code (), bytes_transferred); - }); - else - handler (boost::system::error_code (), 0); - } - - // AsyncWriteStream - template - void async_write_some(const ConstBufferSequence& bufs, WriteHandler&& handler) - { - 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 - } - handler (boost::system::error_code (), sent); - } - - private: - - std::shared_ptr m_Stream; - }; } } - #endif diff --git a/libi2pd_client/BoostStream.h b/libi2pd_client/BoostStream.h new file mode 100644 index 00000000..f5e51729 --- /dev/null +++ b/libi2pd_client/BoostStream.h @@ -0,0 +1,89 @@ +/* +* Copyright (c) 2026, The PurpleI2P Project +* +* This file is part of Purple i2pd project and licensed under BSD3 +* +* See full license text in LICENSE file at top of project tree +*/ + +#ifndef BOOST_STREAM_H_ +#define BOOST_STREAM_H_ + +#include +#include +#include "Streaming.h" + +namespace i2p +{ +namespace client +{ + class BoostAsyncStream // for boost::beast and boost::asio + { + public: + + using executor_type = boost::asio::any_io_executor; + + BoostAsyncStream (std::shared_ptr stream): m_Stream (stream) {} + + std::shared_ptr GetStream () const { return m_Stream; } + + // AsyncStream + executor_type get_executor() noexcept { return m_Stream->GetService ().get_executor(); } + + // AsyncReadStream + template + void async_read_some(const MutableBufferSequence& bufs, ReadHandler&& handler) + { + size_t received = 0; + for (auto it = boost::asio::buffer_sequence_begin (bufs); it != boost::asio::buffer_sequence_end (bufs); it++) + { + auto len = m_Stream->ReadSome ((uint8_t *)it->data (), it->size ()); + received += len; + if (received < it->size ()) break; + } + if (received > 0) // we have some data + handler (boost::system::error_code (), received); + else if (bufs.size () > 0) // wait for incoming data + m_Stream->AsyncReceive (*boost::asio::buffer_sequence_begin (bufs), + [handler = std::move (handler)](const boost::system::error_code& ecode, size_t bytes_transferred) mutable + { + handler (boost::system::error_code (), bytes_transferred); + }); + else + handler (boost::system::error_code (), 0); + } + + // AsyncWriteStream + template + void async_write_some(const ConstBufferSequence& bufs, WriteHandler&& handler) + { + 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 + } + handler (boost::system::error_code (), sent); + } + + private: + + std::shared_ptr m_Stream; + }; +} +} + +namespace boost::beast +{ + // for websockets over BoostAsyncStream + template + void async_teardown (boost::beast::role_type role, i2p::client::BoostAsyncStream& stream, TeardownHandler&& handler) + { + auto s = stream.GetStream (); + if (s) s->Close (); + handler (boost::system::error_code ()); // TODO: check stream's status + } +} +#endif diff --git a/libi2pd_client/Torrents.cpp b/libi2pd_client/Torrents.cpp index 761fbc28..a9e6cdb6 100644 --- a/libi2pd_client/Torrents.cpp +++ b/libi2pd_client/Torrents.cpp @@ -854,7 +854,7 @@ namespace torrents { if (stream) { - auto httpStream = std::make_shared(stream); + auto httpStream = std::make_shared(stream); boost::beast::http::async_write (*httpStream, *req, std::bind (&TorrentsTunnel::TrackerRequestSent, this, std::placeholders::_1, std::placeholders::_2, httpStream, torrent, req)); } @@ -862,7 +862,7 @@ namespace torrents } void TorrentsTunnel::TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred, - std::shared_ptr httpStream, std::shared_ptr torrent, + std::shared_ptr httpStream, std::shared_ptr torrent, std::shared_ptr > req) { if (!ecode) diff --git a/libi2pd_client/Torrents.h b/libi2pd_client/Torrents.h index 22d7304f..f39491f4 100644 --- a/libi2pd_client/Torrents.h +++ b/libi2pd_client/Torrents.h @@ -26,6 +26,7 @@ #include "HTTP.h" #include "I2PService.h" #include "AddressBook.h" +#include "BoostStream.h" namespace i2p { @@ -201,7 +202,7 @@ namespace torrents void RequestTracker (std::shared_ptr torrent); void TrackerRequestSent (const boost::beast::error_code& ecode, size_t bytes_transferred, - std::shared_ptr httpStream, std::shared_ptr torrent, + std::shared_ptr httpStream, std::shared_ptr torrent, std::shared_ptr > req); void ScheduleTrackerRequestsCheck ();