diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b37defc9b..a269ae9fe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -578,6 +578,17 @@ jobs: platform: "x64" version: "vs2022" tests: "*" + - image: windows-2025-vs2026 + configuration: "StaticRelease" + platform: "x64" + version: "vs2026" + tests: "*" + + - image: windows-2025-vs2026 + configuration: "StaticDebug" + platform: "x64" + version: "vs2026" + tests: "*" runs-on: ${{ matrix.image }} diff --git a/Makefile.am b/Makefile.am index 1bbe3d5f1..c5beae820 100644 --- a/Makefile.am +++ b/Makefile.am @@ -108,12 +108,12 @@ src_libbitcoin_network_la_SOURCES = \ src/net/proxy_actions.cpp \ src/net/proxy_queue.cpp \ src/net/socket.cpp \ + src/net/socket_body.cpp \ src/net/socket_connect.cpp \ src/net/socket_http.cpp \ - src/net/socket_p2p.cpp \ - src/net/socket_raw.cpp \ src/net/socket_rpc.cpp \ src/net/socket_stop.cpp \ + src/net/socket_tcp.cpp \ src/net/socket_wait.cpp \ src/net/socket_ws.cpp \ src/protocols/protocol.cpp \ diff --git a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj index f1eeba864..18f16a7ac 100644 --- a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj +++ b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj @@ -202,12 +202,12 @@ + - - + diff --git a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters index b428ac415..e66de30ca 100644 --- a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters +++ b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters @@ -387,16 +387,13 @@ src\net - - src\net - - + src\net - + src\net - + src\net @@ -405,6 +402,9 @@ src\net + + src\net + src\net diff --git a/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj b/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj index 592a59386..9b3ba599d 100644 --- a/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj +++ b/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj @@ -134,7 +134,6 @@ - $(IntDir)test_config_address.obj diff --git a/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters b/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters index 264dbeca4..1d5ff0a3c 100644 --- a/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters +++ b/builds/msvc/vs2026/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters @@ -153,9 +153,6 @@ src\channels - - src\channels - src\config diff --git a/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj b/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj index 0b2b524c1..5bcdaee0e 100644 --- a/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj +++ b/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj @@ -128,7 +128,6 @@ - $(IntDir)src_config_address.obj @@ -203,6 +202,7 @@ + @@ -311,7 +311,6 @@ - @@ -426,7 +425,6 @@ - diff --git a/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj.filters b/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj.filters index 96ef47f7c..e66de30ca 100644 --- a/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj.filters +++ b/builds/msvc/vs2026/libbitcoin-network/libbitcoin-network.vcxproj.filters @@ -189,9 +189,6 @@ src\channels - - src\channels - src\config @@ -390,6 +387,9 @@ src\net + + src\net + src\net @@ -698,9 +698,6 @@ include\bitcoin\network\channels - - include\bitcoin\network\channels - include\bitcoin\network\channels @@ -1043,9 +1040,6 @@ include\bitcoin\network\protocols - - include\bitcoin\network\protocols - include\bitcoin\network\protocols diff --git a/include/bitcoin/network/channels/channel_rpc.hpp b/include/bitcoin/network/channels/channel_rpc.hpp index a7f55724e..3aaf697c9 100644 --- a/include/bitcoin/network/channels/channel_rpc.hpp +++ b/include/bitcoin/network/channels/channel_rpc.hpp @@ -91,8 +91,7 @@ class channel_rpc /// Handle send completion, invokes receive() for non-notifications. template inline void handle_send(const code& ec, size_t bytes, - const rpc::message_cptr& message, - const result_handler& handler) NOEXCEPT; + const std::string& method, const result_handler& handler) NOEXCEPT; /// Stranded handler invoked from stop(). inline void stopping(const code& ec) NOEXCEPT override; diff --git a/include/bitcoin/network/error.hpp b/include/bitcoin/network/error.hpp index 6c3d9c736..b8514061d 100644 --- a/include/bitcoin/network/error.hpp +++ b/include/bitcoin/network/error.hpp @@ -313,7 +313,6 @@ enum error_t : uint8_t jsonrpc_v1_requires_array_params, jsonrpc_v1_requires_id, jsonrpc_params_not_collection, - jsonrpc_reader_bad_buffer, jsonrpc_reader_stall, jsonrpc_reader_exception, jsonrpc_writer_exception diff --git a/include/bitcoin/network/impl/channels/channel_rpc.ipp b/include/bitcoin/network/impl/channels/channel_rpc.ipp index e5fa484f0..b523f0f65 100644 --- a/include/bitcoin/network/impl/channels/channel_rpc.ipp +++ b/include/bitcoin/network/impl/channels/channel_rpc.ipp @@ -55,8 +55,8 @@ inline void CLASS::resume() NOEXCEPT // Read cycle. // ---------------------------------------------------------------------------- +// Failure to call receive() after successful message handling stalls channel. -// Failure to call after successful message handling causes stalled channel. TEMPLATE inline void CLASS::receive() NOEXCEPT { @@ -69,19 +69,8 @@ inline void CLASS::receive() NOEXCEPT return; reading_ = true; - const auto in = emplace_shared - ( - // default json model, unused size_hint, unused serialization buffer. - json::json_value{}, + const auto in = to_shared(); - // default incoming rpc message. - rpc::request_t{}, - - // !strict allows params singleton to be accepted as array (Electrum). - false - ); - - // Post handle_read to strand upon stop, error, or buffer full. read(request_buffer(), *in, std::bind(&channel_rpc::handle_receive, shared_from_base(), _1, _2, in)); @@ -149,6 +138,15 @@ inline http::flat_buffer& CLASS::request_buffer() NOEXCEPT // Send. // ---------------------------------------------------------------------------- +template +std::string extract_method(const Message& message) NOEXCEPT +{ + if constexpr (is_same_type) + return "response: " + message.error.value_or(rpc::result_t{}).message; + else + return "request: " + message.method; +} + // protected TEMPLATE template @@ -157,53 +155,39 @@ inline void CLASS::send(Message&& message, size_t size_hint, { BC_ASSERT(stranded()); using namespace std::placeholders; - using namespace system; - - // Templated message due to notification sending request_t. - const auto out = emplace_shared> - ( - // default json model, buffer size_hint, default serialization buffer. - json::json_value{ {}, size_hint, {} }, - // outgoing rpc message (request_t or response_t). - std::forward(message), + auto complete = std::bind(&CLASS::handle_send, + shared_from_base(), _1, _2, extract_method(message), + std::move(handler)); - // unused strict json-rpc. - true - ); - - // Write message to socket, capture its pointer for lifetime. - write(*out, std::bind(&CLASS::handle_send, - shared_from_base(), _1, _2, out, std::move(handler))); + // Write message (response or notification) to socket. + write( + { + json::json_value{ .size_hint = size_hint }, + std::forward(message) + }, std::move(complete)); } // protected TEMPLATE template inline void CLASS::handle_send(const code& ec, size_t bytes, - const rpc::message_cptr& message, - const result_handler& handler) NOEXCEPT + const std::string& method, const result_handler& handler) NOEXCEPT { BC_ASSERT(stranded()); - if (ec) stop(ec); + if (ec) + stop(ec); + + LOGA("Rpc " << method << " (" << bytes << ") bytes [" << endpoint() << "] "); // Typically a noop, but handshake may pause channel here. handler(ec); - // Only invoke continuation for a request response (not notification). + // Restart the listener (following response to request only). if constexpr (is_same_type) { - LOGA("Rpc response: (" << bytes << ") bytes [" << endpoint() << "] " - << message->message.error.value_or(rpc::result_t{}).message); - - // Continue the read loop (does not unpause or restart). receive(); } - else - { - LOGA("Rpc notification: (" << bytes << ") bytes [" << endpoint() << "] " - << message->message.method); - } } TEMPLATE diff --git a/include/bitcoin/network/impl/messages/json_body.ipp b/include/bitcoin/network/impl/messages/json_body.ipp index 8b5ef4d2b..e33b0bfe9 100644 --- a/include/bitcoin/network/impl/messages/json_body.ipp +++ b/include/bitcoin/network/impl/messages/json_body.ipp @@ -61,6 +61,19 @@ size_t CLASS::reader::put(const buffer_type& buffer, boost_code& ec) NOEXCEPT using namespace system; using namespace network::error; + const auto size = buffer.size(); + if (is_zero(size)) + { + ec.clear(); + return {}; + } + + if (is_null(buffer.data())) + { + ec = to_http_code(http_error_t::bad_alloc); + return {}; + } + try { const auto data = pointer_cast(buffer.data()); @@ -89,21 +102,32 @@ size_t CLASS::reader::put(const buffer_type& buffer, boost_code& ec) NOEXCEPT TEMPLATE void CLASS::reader::finish(boost_code& ec) NOEXCEPT { - ec.clear(); + // Finishing can be very confusing. The derived rpc body calls this base + // method, and then the virtual done() is tested to determine whether the + // logical object is fully read (including optionally required terminator). + // Any error code here signals the beast reader (and any reader that + // terminates based on the end of the framed data, such as websockets) that + // the parse has failed (terminal error). However for custom stream readers + // that may not be aware of byte termination, the `need_more` implies that + // the parse has not failed and that more bytes may be parsed. In either + // case, when this is called and the parse is complete, then the parsed + // json object is moved to the model and the parser is released. In the + // case of the derived json-rpc reader, the json is also then converted to + // the rpc model if valid, otherwise returning a failure code. In no case + // is the underlying parser_.finish(ec) ever called, as that would preclude + // use in the unbounded scenario. - // Calling parser.finish() when parser.done() results in error::incomplete. - if (!parser_.done()) + using namespace network::error; + if (!done()) { - BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) - parser_.finish(ec); - BC_POP_WARNING() - - if (ec) return; + ec = to_http_code(http_error_t::need_more); + return; } try { - value_.model = parser_.release(); + ec.clear(); + value_.model = parser_.release(); } catch (const boost::system::system_error& e) { @@ -113,7 +137,6 @@ void CLASS::reader::finish(boost_code& ec) NOEXCEPT catch (...) { // As a catch-all we blame alloc. - using namespace network::error; ec = to_http_code(http_error_t::bad_alloc); } @@ -158,12 +181,18 @@ CLASS::writer::out_buffer CLASS::writer::get(boost_code& ec) NOEXCEPT { using namespace network::error; - if (serializer_.done()) + if (done()) { ec = to_http_code(http_error_t::end_of_stream); return {}; } + if (!value_.buffer) + { + ec = to_http_code(http_error_t::bad_alloc); + return {}; + } + const auto size = value_.buffer->max_size(); if (is_zero(size)) { diff --git a/include/bitcoin/network/messages/http_body.hpp b/include/bitcoin/network/messages/http_body.hpp index a4833d963..507f379b5 100644 --- a/include/bitcoin/network/messages/http_body.hpp +++ b/include/bitcoin/network/messages/http_body.hpp @@ -70,7 +70,8 @@ using body_writer = std::variant buffer_writer, // 16 bytes string_writer, // 8 bytes json_writer, // 136 bytes! - rpc::writer // 144 bytes! + rpc::writer, // 144 bytes! + rpc::notifier // 144 bytes! >; using empty_value = http::empty_body::value_type; @@ -177,25 +178,29 @@ struct BCT_API body /// Select reader based on content-type header. inline void assign_reader(header& header, value_type& value) NOEXCEPT { - switch (http::content_media_type(header)) + // Allows for preselection of reader for non-http reads. + if (!value_.has_value()) { - case http::media_type::application_json: - if (value_.plain_json) - value = json_value{}; - else - value = rpc::request{}; - break; - case http::media_type::text_plain: - value = string_value{}; - break; - case http::media_type::application_octet_stream: - if (http::has_attachment(header)) - value = file_value{}; - else - value = data_value{}; - break; - default: - value = empty_value{}; + switch (http::content_media_type(header)) + { + case http::media_type::application_json: + if (value_.plain_json) + value = json_value{}; + else + value = rpc::request{}; + break; + case http::media_type::text_plain: + value = string_value{}; + break; + case http::media_type::application_octet_stream: + if (http::has_attachment(header)) + value = file_value{}; + else + value = data_value{}; + break; + default: + value = empty_value{}; + } } std::visit(overload @@ -302,14 +307,17 @@ struct BCT_API body }, [&](rpc::response& value) NOEXCEPT { - // json_writer is not movable (by contained serializer). + // rpc::writer is not movable (by contained serializer). // So requires in-place construction for variant populate. return body_writer{ std::in_place_type, header, value }; }, - [&](rpc::request&) NOEXCEPT + [&](rpc::request& value) NOEXCEPT { - return body_writer{ std::monostate{} }; + // rpc::writer is not movable (by contained serializer). + // So requires in-place construction for variant populate. + return body_writer{ std::in_place_type, + header, value }; } }, value.value()); } diff --git a/include/bitcoin/network/messages/json_body.hpp b/include/bitcoin/network/messages/json_body.hpp index 89453e29d..0cc81acad 100644 --- a/include/bitcoin/network/messages/json_body.hpp +++ b/include/bitcoin/network/messages/json_body.hpp @@ -76,11 +76,11 @@ struct body protected: value_type& value_; - boost::json::stream_parser parser_{}; private: size_t total_{}; http::length_type expected_{}; + boost::json::stream_parser parser_{}; }; class writer diff --git a/include/bitcoin/network/messages/rpc/body.hpp b/include/bitcoin/network/messages/rpc/body.hpp index 09391ac4a..3f7eb8c96 100644 --- a/include/bitcoin/network/messages/rpc/body.hpp +++ b/include/bitcoin/network/messages/rpc/body.hpp @@ -33,7 +33,7 @@ struct message_type : public json::json_value { Type message{}; - bool strict{ true }; + bool strict{}; }; /// Derived boost::beast::http body for JSON-RPC messages. @@ -115,6 +115,8 @@ using response_cptr = std::shared_ptr; using response_ptr = std::shared_ptr; using writer = response_body::writer; +using notifier = rpc::request_body::writer; + // Allows request to be sent (for notifications). template using message_value = typename body::value_type; diff --git a/include/bitcoin/network/net/proxy.hpp b/include/bitcoin/network/net/proxy.hpp index 0fbd56db5..fd18eb1c1 100644 --- a/include/bitcoin/network/net/proxy.hpp +++ b/include/bitcoin/network/net/proxy.hpp @@ -122,7 +122,7 @@ class BCT_API proxy /// Cancel wait or any asynchronous read/write operation, handlers posted. virtual void cancel(result_handler&& handler) NOEXCEPT; - /// RAW (generic, variable size). + /// WS (generic, framed). /// ----------------------------------------------------------------------- /// Read complete logical message for websockets (not for tcp). @@ -135,7 +135,7 @@ class BCT_API proxy virtual void write(const asio::const_buffer& in, bool binary, count_handler&& handler) NOEXCEPT; - /// P2P (generic, fixed size, WS: always binary). + /// TCP (generic, fixed size). /// ----------------------------------------------------------------------- /// Read fixed-size TCP message from the remote endpoint into buffer. @@ -154,11 +154,11 @@ class BCT_API proxy count_handler&& handler) NOEXCEPT; /// Write rpc response to the socket (json buffer in body). - virtual void write(rpc::response& response, + virtual void write(rpc::response&& response, count_handler&& handler) NOEXCEPT; /// Write rpc notification (request) to the socket (json buffer in body). - virtual void write(rpc::request& notification, + virtual void write(rpc::request&& notification, count_handler&& handler) NOEXCEPT; /// HTTP (generic/rpc). @@ -177,13 +177,13 @@ class BCT_API proxy typedef std::deque queue; // For write buffering. - void do_raw_write(const asio::const_buffer& payload, bool binary, + void do_ws_write(const asio::const_buffer& payload, bool binary, const count_handler& handler) NOEXCEPT; - void do_p2p_write(const asio::const_buffer& payload, + void do_tcp_write(const asio::const_buffer& payload, const count_handler& handler) NOEXCEPT; - void do_response_write(const ref& response, + void do_response_write(const rpc::response_ptr& response, const count_handler& handler) NOEXCEPT; - void do_notification_write(const ref& notification, + void do_notification_write(const rpc::request_ptr& notification, const count_handler& handler) NOEXCEPT; void do_subscribe_stop(const result_handler& handler, const result_handler& complete) NOEXCEPT; diff --git a/include/bitcoin/network/net/socket.hpp b/include/bitcoin/network/net/socket.hpp index 72c3f854e..e69bfa077 100644 --- a/include/bitcoin/network/net/socket.hpp +++ b/include/bitcoin/network/net/socket.hpp @@ -110,28 +110,26 @@ class BCT_API socket virtual void connect(const asio::endpoints& range, result_handler&& handler) NOEXCEPT; - /// RAW (generic, variable size). + /// WS (generic, framed). /// ----------------------------------------------------------------------- - /// Read complete logical message for websockets (not for tcp). - /// Read available buffer from the socket, handler posted to socket strand. - virtual void raw_read(http::flat_buffer& out, + /// Read framed message from websocket, handler posted to socket strand. + virtual void ws_read(http::flat_buffer& out, count_handler&& handler) NOEXCEPT; - /// Binary or text mode applies to websockets (no-op for tcp). - /// Write the provided buffer to socket, handler posted to socket strand. - virtual void raw_write(const asio::const_buffer& in, bool binary, + /// Write buffer as frame to websocket, handler posted to socket strand. + virtual void ws_write(const asio::const_buffer& in, bool binary, count_handler&& handler) NOEXCEPT; - /// P2P (generic, fixed size). + /// TCP (generic, fixed size). /// ----------------------------------------------------------------------- - /// Read fixed buffer from the socket, handler posted to socket strand. - virtual void p2p_read(const asio::mutable_buffer& out, + /// Read fixed buffer from tcp socket, handler posted to socket strand. + virtual void tcp_read(const asio::mutable_buffer& out, count_handler&& handler) NOEXCEPT; - /// Write the provided buffer to socket, handler posted to socket strand. - virtual void p2p_write(const asio::const_buffer& in, + /// Write fixed buffer to socket, handler posted to socket strand. + virtual void tcp_write(const asio::const_buffer& in, count_handler&& handler) NOEXCEPT; /// RPC (TCP: electrum/stratum_v1, WS: btcd). @@ -142,11 +140,26 @@ class BCT_API socket count_handler&& handler) NOEXCEPT; /// Write rpc response to the socket, handler posted to socket strand. - virtual void rpc_write(rpc::response& response, + virtual void rpc_write(rpc::response&& response, count_handler&& handler) NOEXCEPT; /// Write rpc notification to the socket, handler posted to socket strand. - virtual void rpc_notify(rpc::request& notification, + virtual void rpc_notify(rpc::request&& notification, + count_handler&& handler) NOEXCEPT; + + /// BODY (non-http body read/write). + /// ----------------------------------------------------------------------- + + /// Read body request from the socket, handler posted to socket strand. + virtual void body_read(http::flat_buffer& buffer, http::request& request, + count_handler&& handler) NOEXCEPT; + + /// Write body response to the socket, handler posted to socket strand. + virtual void body_write(http::response&& response, + count_handler&& handler) NOEXCEPT; + + /// Write body notification to the socket, handler posted to socket strand. + virtual void body_notify(http::request&& notification, count_handler&& handler) NOEXCEPT; /// HTTP (generic/rpc). @@ -224,8 +237,8 @@ class BCT_API socket /// Variant (ws vs. tcp) helpers (protected by strand). /// ----------------------------------------------------------------------- - void async_read(http::flat_buffer& buffer, const count_handler& handler, - size_t size=rpc::writer::default_buffer) NOEXCEPT; + void async_read(http::flat_buffer& buffer, + const count_handler& handler) NOEXCEPT; void async_read(const asio::mutable_buffer& buffer, const count_handler& handler) NOEXCEPT; void async_read_some(const asio::mutable_buffer& buffer, @@ -237,46 +250,49 @@ class BCT_API socket using http_parser = boost::beast::http::request_parser; using http_parser_ptr = std::shared_ptr; - struct read_rpc + struct read_state { - typedef std::shared_ptr ptr; + typedef std::shared_ptr ptr; - read_rpc(rpc::request& request, http::flat_buffer& buffer) NOEXCEPT - : value{ request }, reader{ value }, buffer{ buffer } + read_state(http::request& request, http::flat_buffer& buffer) NOEXCEPT + : value{ request }, reader{ value.base(), value.body() }, + buffer{ buffer } { } - rpc::request& value; - rpc::request_body::reader reader; + http::request& value; + http::body::reader reader; http::flat_buffer& buffer; }; - struct write_rpc + struct write_state { - typedef std::shared_ptr ptr; - using out_buffer = rpc::writer::out_buffer; + typedef std::shared_ptr ptr; + using out_buffer = http::body::writer::out_buffer; - write_rpc(rpc::response& response) NOEXCEPT - : value{ response }, writer{ value } + write_state(http::response&& response) NOEXCEPT + : value{ std::move(response) }, writer{ value.base(), value.body() } { } - rpc::response& value; - rpc::response_body::writer writer; + bool more{ true }; + http::response value; + http::body::writer writer; }; - struct notify_rpc + struct notify_state { - typedef std::shared_ptr ptr; - using out_buffer = rpc::writer::out_buffer; + typedef std::shared_ptr ptr; + using out_buffer = http::body::writer::out_buffer; - notify_rpc(rpc::request& request) NOEXCEPT - : value{ request }, writer{ value } + notify_state(http::request&& request) NOEXCEPT + : value{ std::move(request) }, writer{ value.base(), value.body() } { } - rpc::request& value; - rpc::request_body::writer writer; + bool more{ true }; + http::request value; + http::body::writer writer; }; // do @@ -299,23 +315,25 @@ class BCT_API socket const result_handler& handler) NOEXCEPT; void do_handshake(const result_handler& handler) NOEXCEPT; - // raw (tcp/ws) - void do_raw_read(ref out, + // ws (framed) + void do_ws_read(ref out, const count_handler& handler) NOEXCEPT; - void do_raw_write(const asio::const_buffer& in, bool binary, + void do_ws_write(const asio::const_buffer& in, bool binary, const count_handler& handler) NOEXCEPT; - // p2p (tcp/ws) - void do_p2p_read(const asio::mutable_buffer& out, + // tcp (fixed) + void do_tcp_read(const asio::mutable_buffer& out, const count_handler& handler) NOEXCEPT; - // rpc (tcp/ws) - void do_rpc_read(boost_code ec, size_t total, - const read_rpc::ptr& in, const count_handler& handler) NOEXCEPT; - void do_rpc_write(boost_code ec, size_t total, - const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT; - void do_rpc_notify(boost_code ec, size_t total, - const notify_rpc::ptr& out, const count_handler& handler) NOEXCEPT; + // body + void do_body_read(boost_code ec, size_t total, + const read_state::ptr& in, const count_handler& handler) NOEXCEPT; + void handle_body_read(boost_code ec, size_t size, size_t total, + const read_state::ptr& in, const count_handler& handler) NOEXCEPT; + void do_body_write(boost_code ec, size_t total, + const write_state::ptr& out, const count_handler& handler) NOEXCEPT; + void do_body_notify(boost_code ec, size_t total, + const notify_state::ptr& out, const count_handler& handler) NOEXCEPT; // http void do_http_read(ref buffer, @@ -347,20 +365,23 @@ class BCT_API socket void handle_handshake(const boost_code& ec, const result_handler& handler) NOEXCEPT; - // raw/p2p/read/write (tcp/ws) + // read/write (tcp/ws) void handle_async(const boost_code& ec, size_t size, const count_handler& handler,const std::string& operation) NOEXCEPT; void handle_async_read(const boost_code& ec, size_t size, const asio::mutable_buffer& buffer, const http::flat_buffer_ptr& flat, const count_handler& handler) NOEXCEPT; - // rpc (tcp/ws) - void handle_rpc_read(boost_code ec, size_t size, size_t total, - const read_rpc::ptr& in, const count_handler& handler) NOEXCEPT; - void handle_rpc_write(boost_code ec, size_t size, size_t total, - const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT; - void handle_rpc_notify(boost_code ec, size_t size, size_t total, - const notify_rpc::ptr& out, const count_handler& handler) NOEXCEPT; + // rpc + void handle_rpc_read(const code& ec, size_t bytes, + const ref& out, const http::request_ptr& in, + const count_handler& handler) NOEXCEPT; + + // body + void handle_body_write(boost_code ec, size_t size, size_t total, + const write_state::ptr& out, const count_handler& handler) NOEXCEPT; + void handle_body_notify(boost_code ec, size_t size, size_t total, + const notify_state::ptr& out, const count_handler& handler) NOEXCEPT; // http (generic/rpc) void handle_http_read(const boost_code& ec, size_t size, diff --git a/src/channels/channel_http.cpp b/src/channels/channel_http.cpp index 5d9d6427b..bf03585e5 100644 --- a/src/channels/channel_http.cpp +++ b/src/channels/channel_http.cpp @@ -165,18 +165,18 @@ void channel_http::send(response&& response, result_handler&& handler) NOEXCEPT BC_ASSERT(stranded()); assign_json_buffer(response); - const auto ptr = system::move_shared(std::move(response)); + const auto out = system::move_shared(std::move(response)); count_handler complete = std::bind(&channel_http::handle_send, - shared_from_base(), _1, _2, ptr, std::move(handler)); + shared_from_base(), _1, _2, out, std::move(handler)); - if (!ptr) + if (!out) { complete(error::bad_alloc, {}); return; } - // response has been moved to ptr. - write(*ptr, std::move(complete)); + // response has been moved to out. + write(*out, std::move(complete)); } void channel_http::handle_send(const code& ec, size_t bytes, diff --git a/src/error.cpp b/src/error.cpp index 4493b0ea6..ea857ec64 100644 --- a/src/error.cpp +++ b/src/error.cpp @@ -300,7 +300,6 @@ DEFINE_ERROR_T_MESSAGE_MAP(error) { jsonrpc_v1_requires_array_params, "jsonrpc v1 requires array params" }, { jsonrpc_v1_requires_id, "jsonrpc v1 requires id" }, { jsonrpc_params_not_collection, "jsonrpc params not collection" }, - { jsonrpc_reader_bad_buffer, "jsonrpc reader bad buffer" }, { jsonrpc_reader_stall, "jsonrpc reader stall" }, { jsonrpc_reader_exception, "jsonrpc reader exception" }, { jsonrpc_writer_exception, "jsonrpc writer exception" } diff --git a/src/messages/rpc/body.cpp b/src/messages/rpc/body.cpp index a98ea7022..26b2ac993 100644 --- a/src/messages/rpc/body.cpp +++ b/src/messages/rpc/body.cpp @@ -35,6 +35,18 @@ BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) BC_PUSH_WARNING(NO_UNGUARDED_POINTERS) BC_PUSH_WARNING(NO_POINTER_ARITHMETIC) +constexpr bool is_rpc_terminator(char character) NOEXCEPT +{ + // Batched rpc messages are array-terminated. + return character == '}' || character == ']'; +} + +constexpr bool is_electrum_terminator(char character) NOEXCEPT +{ + // Electrum rpc messages require a trailing newline. + return character == '\n'; +} + // rpc::body::reader // ---------------------------------------------------------------------------- @@ -42,21 +54,11 @@ template <> size_t body::reader:: put(const buffer_type& buffer, boost_code& ec) NOEXCEPT { - const auto size = buffer.size(); - if (is_zero(size)) - { - ec.clear(); - return {}; - } - - if (is_null(buffer.data())) - { - ec = code{ error::jsonrpc_reader_bad_buffer }; - return {}; - } - + // Null and empty guarded in base reader. const auto parsed = base::reader::put(buffer, ec); - if (ec || !parser_.done()) + + // Not done until terminated and can't search for it until parser is done. + if (ec || !base::reader::done()) return parsed; // http json does not use termination. @@ -71,20 +73,23 @@ put(const buffer_type& buffer, boost_code& ec) NOEXCEPT return parsed; } - // boost::json consumes whitespace, and leaves any subsequent chars - // unparsed, so terminator must be in the parsed segment of the buffer. const auto data = pointer_cast(buffer.data()); - for (auto index = parsed; !is_zero(index);) + + // boost::json consumes whitespace, and leaves any subsequent chars + // unparsed, so terminator must be in the parsed buffer segment. + for (auto index = parsed; !is_zero(index); --index) { - if (data[--index] == '\n') + const auto character = data[sub1(index)]; + if (is_electrum_terminator(character)) { - // There may be unparsed characters (ok, next message). has_terminator_ = true; - return parsed; + break; } + + if (is_rpc_terminator(character)) + break; } - // There is no terminator (yet). return parsed; } @@ -93,13 +98,14 @@ bool body::reader:: done() const NOEXCEPT { // Parser may be done but with terminator still outstanding. - return parser_.done() && (!terminated_ || has_terminator_); + return base::reader::done() && (!terminated_ || has_terminator_); } template <> void body::reader:: finish(boost_code& ec) NOEXCEPT { + // See notes in base reader. base::reader::finish(ec); if (ec) return; diff --git a/src/net/connector_socks.cpp b/src/net/connector_socks.cpp index f41e43e36..1e07479d8 100644 --- a/src/net/connector_socks.cpp +++ b/src/net/connector_socks.cpp @@ -177,7 +177,7 @@ void connector_socks::do_socks_greeting_write(const code& ec, // Start of socket strand sequence. // All socket operations are dispatched to its own strand, so this write // will be posted before invocation. This makes socket calls thread safe. - socket->p2p_write({ greeting->data(), greeting->size() }, + socket->tcp_write({ greeting->data(), greeting->size() }, std::bind(&connector_socks::handle_socks_greeting_write, shared_from_base(), _1, _2, socket, greeting)); @@ -202,7 +202,7 @@ void connector_socks::handle_socks_greeting_write(const code& ec, size_t size, const auto response = emplace_shared>(); - socket->p2p_read({ response->data(), response->size() }, + socket->tcp_read({ response->data(), response->size() }, std::bind(&connector_socks::handle_socks_method_read, shared_from_base(), _1, _2, socket, response)); @@ -273,7 +273,7 @@ void connector_socks::handle_socks_method_read(const code& ec, size_t size, *it++ = narrow_cast(password_length); it = std::copy(password.begin(), password.end(), it); - socket->p2p_write({ authenticator->data(), authenticator->size() }, + socket->tcp_write({ authenticator->data(), authenticator->size() }, std::bind(&connector_socks::handle_socks_authentication_write, shared_from_base(), _1, _2, socket, authenticator)); @@ -299,7 +299,7 @@ void connector_socks::handle_socks_authentication_write(const code& ec, const auto auth_res = emplace_shared>(); - socket->p2p_read({ auth_res->data(), auth_res->size() }, + socket->tcp_read({ auth_res->data(), auth_res->size() }, std::bind(&connector_socks::handle_socks_authentication_read, shared_from_base(), _1, _2, socket, auth_res)); @@ -404,7 +404,7 @@ void connector_socks::do_socks_connect_write( it = std::copy(port.begin(), port.end(), it); } - socket->p2p_write({ request->data(), request->size() }, + socket->tcp_write({ request->data(), request->size() }, std::bind(&connector_socks::handle_socks_connect_write, shared_from_base(), _1, _2, socket, request)); @@ -429,7 +429,7 @@ void connector_socks::handle_socks_connect_write(const code& ec, size_t size, const auto response = emplace_shared>(); - socket->p2p_read({ response->data(), response->size() }, + socket->tcp_read({ response->data(), response->size() }, std::bind(&connector_socks::handle_socks_response_read, shared_from_base(), _1, _2, socket, response)); @@ -473,7 +473,7 @@ void connector_socks::handle_socks_response_read(const code& ec, size_t size, // A version-4 IP address with length of 4 octets. const auto address = emplace_shared(4u + port_size); - socket->p2p_read({ address->data(), address->size() }, + socket->tcp_read({ address->data(), address->size() }, std::bind(&connector_socks::handle_socks_address_read, shared_from_base(), _1, _2, socket, address)); @@ -484,7 +484,7 @@ void connector_socks::handle_socks_response_read(const code& ec, size_t size, // A version-6 IP address with length of 16 octets. const auto address = emplace_shared(16u + port_size); - socket->p2p_read({ address->data(), address->size() }, + socket->tcp_read({ address->data(), address->size() }, std::bind(&connector_socks::handle_socks_address_read, shared_from_base(), _1, _2, socket, address)); @@ -497,7 +497,7 @@ void connector_socks::handle_socks_response_read(const code& ec, size_t size, // of name that follow (and excludes two byte length of the port). const auto length = emplace_shared>(); - socket->p2p_read({ length->data(), sizeof(uint8_t) }, + socket->tcp_read({ length->data(), sizeof(uint8_t) }, std::bind(&connector_socks::handle_socks_length_read, shared_from_base(), _1, _2, socket, length)); @@ -532,7 +532,7 @@ void connector_socks::handle_socks_length_read(const code& ec, size_t size, const auto length = host_length->front() + port_size; const auto address = emplace_shared(length); - socket->p2p_read({ address->data(), address->size() }, + socket->tcp_read({ address->data(), address->size() }, std::bind(&connector_socks::handle_socks_address_read, shared_from_base(), _1, _2, socket, address)); diff --git a/src/net/proxy_actions.cpp b/src/net/proxy_actions.cpp index bd9403888..5e6022bd0 100644 --- a/src/net/proxy_actions.cpp +++ b/src/net/proxy_actions.cpp @@ -28,6 +28,7 @@ namespace network { BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) +using namespace system; using namespace std::placeholders; // Wait (all). @@ -43,19 +44,19 @@ void proxy::cancel(result_handler&& handler) NOEXCEPT socket_->cancel(std::move(handler)); } -// RAW (generic, variable size). +// WS (generic, framed). // ---------------------------------------------------------------------------- void proxy::read(http::flat_buffer& out, count_handler&& handler) NOEXCEPT { do_reading(); - socket_->raw_read(out, std::move(handler)); + socket_->ws_read(out, std::move(handler)); } void proxy::write(const asio::const_buffer& in, bool binary, count_handler&& handler) NOEXCEPT { - writer call = std::bind(&proxy::do_raw_write, + writer call = std::bind(&proxy::do_ws_write, shared_from_this(), in, binary, std::move(handler)); boost::asio::dispatch(strand(), @@ -64,28 +65,28 @@ void proxy::write(const asio::const_buffer& in, bool binary, } // private -void proxy::do_raw_write(const asio::const_buffer& payload, bool binary, +void proxy::do_ws_write(const asio::const_buffer& payload, bool binary, const count_handler& handler) NOEXCEPT { - socket_->raw_write({ payload.data(), payload.size() }, binary, + socket_->ws_write({ payload.data(), payload.size() }, binary, std::bind(&proxy::handle_write, shared_from_this(), _1, _2, handler)); } -// P2P (generic, fixed size). +// TCP (generic, fixed size). // ---------------------------------------------------------------------------- void proxy::read(const asio::mutable_buffer& out, count_handler&& handler) NOEXCEPT { do_reading(); - socket_->p2p_read(out, std::move(handler)); + socket_->tcp_read(out, std::move(handler)); } void proxy::write(const asio::const_buffer& in, count_handler&& handler) NOEXCEPT { - writer call = std::bind(&proxy::do_p2p_write, + writer call = std::bind(&proxy::do_tcp_write, shared_from_this(), in, std::move(handler)); boost::asio::dispatch(strand(), @@ -94,10 +95,10 @@ void proxy::write(const asio::const_buffer& in, } // private -void proxy::do_p2p_write(const asio::const_buffer& payload, +void proxy::do_tcp_write(const asio::const_buffer& payload, const count_handler& handler) NOEXCEPT { - socket_->p2p_write({ payload.data(), payload.size() }, + socket_->tcp_write({ payload.data(), payload.size() }, std::bind(&proxy::handle_write, shared_from_this(), _1, _2, handler)); } @@ -112,20 +113,24 @@ void proxy::read(http::flat_buffer& buffer, rpc::request& request, socket_->rpc_read(buffer, request, std::move(handler)); } -void proxy::write(rpc::response& response, count_handler&& handler) NOEXCEPT +void proxy::write(rpc::response&& response, count_handler&& handler) NOEXCEPT { + // Pointer ships moveable message through the send queue. + const auto out = move_shared(std::move(response)); writer call = std::bind(&proxy::do_response_write, - shared_from_this(), std::ref(response), std::move(handler)); + shared_from_this(), out, std::move(handler)); boost::asio::dispatch(strand(), std::bind(&proxy::do_write, shared_from_this(), std::move(call))); } -void proxy::write(rpc::request& notification, count_handler&& handler) NOEXCEPT +void proxy::write(rpc::request&& notification, count_handler&& handler) NOEXCEPT { + // Pointer ships moveable message through the send queue. + const auto out = move_shared(std::move(notification)); writer call = std::bind(&proxy::do_notification_write, - shared_from_this(), std::ref(notification), std::move(handler)); + shared_from_this(), out, std::move(handler)); boost::asio::dispatch(strand(), std::bind(&proxy::do_write, @@ -133,19 +138,19 @@ void proxy::write(rpc::request& notification, count_handler&& handler) NOEXCEPT } // private -void proxy::do_response_write(const ref& response, +void proxy::do_response_write(const rpc::response_ptr& response, const count_handler& handler) NOEXCEPT { - socket_->rpc_write(response.get(), + socket_->rpc_write(std::move(*response), std::bind(&proxy::handle_write, shared_from_this(), _1, _2, handler)); } // private -void proxy::do_notification_write(const ref& notification, +void proxy::do_notification_write(const rpc::request_ptr& notification, const count_handler& handler) NOEXCEPT { - socket_->rpc_notify(notification.get(), + socket_->rpc_notify(std::move(*notification), std::bind(&proxy::handle_write, shared_from_this(), _1, _2, handler)); } diff --git a/src/net/socket.cpp b/src/net/socket.cpp index 27213617a..1c0da390f 100644 --- a/src/net/socket.cpp +++ b/src/net/socket.cpp @@ -265,6 +265,7 @@ asio::ssl::socket& socket::get_ssl() NOEXCEPT // ---------------------------------------------------------------------------- // protected +// write framed (ws) or unframed (tcp) fixed size (const buffer size). void socket::async_write(const asio::const_buffer& buffer, bool binary, const count_handler& handler) NOEXCEPT { @@ -293,7 +294,7 @@ void socket::async_write(const asio::const_buffer& buffer, bool binary, } } - +// read some (up to mutable buffer capacity). void socket::async_read_some(const asio::mutable_buffer& buffer, const count_handler& handler) NOEXCEPT { @@ -319,35 +320,33 @@ void socket::async_read_some(const asio::mutable_buffer& buffer, } } -// raw +// read framed/ws (fails if beyond flat buffer capacity). void socket::async_read(http::flat_buffer& buffer, - const count_handler& handler, size_t size) NOEXCEPT + const count_handler& handler) NOEXCEPT { try { if (is_websocket()) { buffer.consume(buffer.size()); - - // Complete logical message semantics. VARIANT_DISPATCH_METHOD(get_ws(), async_read(buffer, std::bind(&socket::handle_async, - shared_from_this(), _1, _2, handler, "async_read_raw"))); + shared_from_this(), _1, _2, handler, "async_read"))); } else { - // Any available data semantics (ok). - async_read_some(buffer.prepare(size), handler); + // Use async_read_some() or async_read(mutable_buffer). + handler(error::operation_failed, {}); } } catch (const std::exception& e) { - LOGF("Exception @ async_read_raw: " << e.what()); + LOGF("Exception @ async_read: " << e.what()); handler(error::operation_failed, {}); } } -// fixed/p2p +// read fixed/p2p (waits until mutable buffer is filled). void socket::async_read(const asio::mutable_buffer& buffer, const count_handler& handler) NOEXCEPT { @@ -355,24 +354,20 @@ void socket::async_read(const asio::mutable_buffer& buffer, { if (is_websocket()) { - const auto flat = std::make_shared(); - count_handler complete = std::bind(&socket::handle_async_read, - shared_from_this(), _1, _2, buffer, flat, handler); - - // Websocket doesn't have fixed-size reads, so simulate it. - async_read(*flat, std::move(complete)); + // Websockets read frame not size, use async_read(flat_buffer). + handler(error::operation_failed, {}); } else { // Fixed-size semantics. VARIANT_DISPATCH_FUNCTION(boost::asio::async_read, get_tcp(), buffer, std::bind(&socket::handle_async, - shared_from_this(), _1, _2, handler, "async_read_fixed")); + shared_from_this(), _1, _2, handler, "async_read")); } } catch (const std::exception& e) { - LOGF("Exception @ async_read_fixed: " << e.what()); + LOGF("Exception @ async_read: " << e.what()); handler(error::operation_failed, {}); } } diff --git a/src/net/socket_body.cpp b/src/net/socket_body.cpp new file mode 100644 index 000000000..46f50c903 --- /dev/null +++ b/src/net/socket_body.cpp @@ -0,0 +1,267 @@ +/** + * Copyright (c) 2011-2026 libbitcoin developers (see AUTHORS) + * + * This file is part of libbitcoin. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +#include + +#include +#include +#include +#include + +namespace libbitcoin { +namespace network { + +using namespace system; +using namespace network::rpc; +using namespace std::placeholders; + +// Shared pointers required in handler parameters so closures control lifetime. +BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR) +BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED) +BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) + +// BODY (read). +// ---------------------------------------------------------------------------- +// The body_ methods accept any boost::beast-compliant body to read/write a tcp +// socket. For http read/write use the http_ methods. The rcp_ methods are just +// a specialization of these methods, passing the rpc::body<> types. For simple +// fixed-size tcp (p2p) use the tcp_ methods, and for simple framed ws use the +// ws_ methods. The body methods require fixed size or framed read/write. The +// json and json-rpc bodies are internally framed by json, so can read/write +// over a raw tcp socket (electrum) or a framed websocket (btcd). And again for +// http framing, use the http_ methods, as these incorporate header processing. + +void socket::body_read(http::flat_buffer& buffer, http::request& request, + count_handler&& handler) NOEXCEPT +{ + boost_code ec{}; + const auto in = emplace_shared(request, buffer); + in->reader.init({}, ec); + + boost::asio::dispatch(strand_, + std::bind(&socket::do_body_read, + shared_from_this(), ec, zero, in, std::move(handler))); +} + +// private +void socket::do_body_read(boost_code ec, size_t total, + const read_state::ptr& in, const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + constexpr auto size = rpc::writer::default_buffer; + + if (ec) + { + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("body-read", ec); + handler(code, total); + return; + } + + async_read_some(in->buffer.prepare(size), + std::bind(&socket::handle_body_read, + shared_from_this(), _1, _2, total, in, handler)); +} + +// private +void socket::handle_body_read(boost_code ec, size_t size, size_t total, + const read_state::ptr& in, const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + + total = ceilinged_add(total, size); + if (error::asio_is_canceled(ec)) + { + handler(error::channel_stopped, total); + return; + } + + if (total > maximum_) + { + handler(error::message_overflow, total); + return; + } + + if (!ec) + { + in->buffer.commit(size); + in->buffer.consume(in->reader.put(in->buffer.data(), ec)); + if (!ec) + { + // The json/json-rpc readers do not finalize on finish, instead + // they return need_more if not complete and success if complete. + in->reader.finish(ec); + + if (!ec) + { + handler(error::success, total); + return; + } + + if (ec == error::http_error_t::need_more) + ec.clear(); + } + } + + // Handle error condition or incomplete message. + do_body_read(ec, total, in, handler); +} + +// Body (write). +// ---------------------------------------------------------------------------- + +void socket::body_write(http::response&& response, + count_handler&& handler) NOEXCEPT +{ + boost_code ec{}; + const auto out = emplace_shared(std::move(response)); + out->writer.init(ec); + + boost::asio::dispatch(strand_, + std::bind(&socket::do_body_write, + shared_from_this(), ec, zero, out, std::move(handler))); +} + +// private +void socket::do_body_write(boost_code ec, size_t total, + const write_state::ptr& out, const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + + const auto buffer = ec ? write_state::out_buffer{} : out->writer.get(ec); + if (!buffer.has_value()) + { + handler(error::bad_stream, total); + return; + } + + if (ec) + { + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("body-write", ec); + handler(code, total); + return; + } + + out->more = buffer.value().second; + const auto& data = buffer.value().first; + + // TODO: derive websocket binary/text from body type mapping. + async_write(data, false, + std::bind(&socket::handle_body_write, + shared_from_this(), _1, _2, total, out, handler)); +} + +// private +void socket::handle_body_write(boost_code ec, size_t size, size_t total, + const write_state::ptr& out, const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + + total = ceilinged_add(total, size); + if (error::asio_is_canceled(ec)) + { + handler(error::channel_stopped, total); + return; + } + + if (!ec && !out->more) + { + handler(error::success, total); + return; + } + + // Handle error condition or incomplete message. + do_body_write(ec, total, out, handler); +} + +// Body (notify). +// ---------------------------------------------------------------------------- +// Identical to body_write() apart from request vs. response. + +void socket::body_notify(http::request&& notification, + count_handler&& handler) NOEXCEPT +{ + boost_code ec{}; + const auto out = emplace_shared(std::move(notification)); + out->writer.init(ec); + + boost::asio::dispatch(strand_, + std::bind(&socket::do_body_notify, + shared_from_this(), ec, zero, out, std::move(handler))); +} + +// private +void socket::do_body_notify(boost_code ec, size_t total, + const notify_state::ptr& out, const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + + const auto buffer = ec ? notify_state::out_buffer{} : out->writer.get(ec); + if (!buffer.has_value()) + { + handler(error::bad_stream, total); + return; + } + + if (ec) + { + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("body-notify", ec); + handler(code, total); + return; + } + + out->more = buffer.value().second; + const auto& data = buffer.value().first; + + // TODO: derive websocket binary/text from body type mapping. + async_write(data, false, + std::bind(&socket::handle_body_notify, + shared_from_this(), _1, _2, total, out, handler)); +} + +// private +void socket::handle_body_notify(boost_code ec, size_t size, size_t total, + const notify_state::ptr& out, const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + + total = ceilinged_add(total, size); + if (error::asio_is_canceled(ec)) + { + handler(error::channel_stopped, total); + return; + } + + if (!ec && !out->more) + { + handler(error::success, total); + return; + } + + // Handle error condition or incomplete message. + do_body_notify(ec, total, out, handler); +} + +BC_POP_WARNING() +BC_POP_WARNING() +BC_POP_WARNING() + +} // namespace network +} // namespace libbitcoin diff --git a/src/net/socket_http.cpp b/src/net/socket_http.cpp index 022ef88a5..2cd9a3a18 100644 --- a/src/net/socket_http.cpp +++ b/src/net/socket_http.cpp @@ -111,7 +111,7 @@ void socket::handle_http_read(const boost_code& ec, size_t size, } const auto code = error::http_to_error_code(ec); - if (code == error::unknown) logx("http", ec); + if (code == error::unknown) logx("http-read", ec); handler(code, size); } @@ -165,7 +165,7 @@ void socket::handle_http_write(const boost_code& ec, size_t size, } const auto code = error::http_to_error_code(ec); - if (code == error::unknown) logx("http", ec); + if (code == error::unknown) logx("http-write", ec); handler(code, size); } diff --git a/src/net/socket_raw.cpp b/src/net/socket_raw.cpp deleted file mode 100644 index 59145e936..000000000 --- a/src/net/socket_raw.cpp +++ /dev/null @@ -1,78 +0,0 @@ -/** - * Copyright (c) 2011-2026 libbitcoin developers (see AUTHORS) - * - * This file is part of libbitcoin. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - */ -#include - -#include -#include -#include - -namespace libbitcoin { -namespace network { - -using namespace std::placeholders; - -BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) - -// RAW (read). -// ---------------------------------------------------------------------------- - -// The flat_buffer must have sufficient capacity already allocated before the -// RAW read is initiated. If the incoming WebSocket message exceeds the current -// capacity, the read will fail or truncate. -void socket::raw_read(http::flat_buffer& out, - count_handler&& handler) NOEXCEPT -{ - boost::asio::dispatch(strand_, - std::bind(&socket::do_raw_read, - shared_from_this(), std::ref(out), std::move(handler))); -} - -// private -void socket::do_raw_read(ref out, - const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - async_read(out.get(), handler); -} - -// RAW (write). -// ---------------------------------------------------------------------------- - -// The const_buffer is fully allocated before write, so this is currently -// identical to p2p write. -void socket::raw_write(const asio::const_buffer& in, bool binary, - count_handler&& handler) NOEXCEPT -{ - boost::asio::dispatch(strand_, - std::bind(&socket::do_raw_write, - shared_from_this(), in, binary, std::move(handler))); -} - -// private -void socket::do_raw_write(const asio::const_buffer& in, bool binary, - const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - async_write(in, binary, handler); -} - -BC_POP_WARNING() - -} // namespace network -} // namespace libbitcoin diff --git a/src/net/socket_rpc.cpp b/src/net/socket_rpc.cpp index 564a92108..88582469c 100644 --- a/src/net/socket_rpc.cpp +++ b/src/net/socket_rpc.cpp @@ -18,16 +18,13 @@ */ #include -#include #include #include -#include namespace libbitcoin { namespace network { using namespace system; -using namespace network::rpc; using namespace std::placeholders; // Shared pointers required in handler parameters so closures control lifetime. @@ -35,207 +32,50 @@ BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR) BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED) BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) -// RPC (read). -// ---------------------------------------------------------------------------- - void socket::rpc_read(http::flat_buffer& buffer, rpc::request& request, count_handler&& handler) NOEXCEPT { - boost_code ec{}; - const auto in = emplace_shared(request, buffer); - in->reader.init({}, ec); - - boost::asio::dispatch(strand_, - std::bind(&socket::do_rpc_read, - shared_from_this(), ec, zero, in, std::move(handler))); -} - -// private -void socket::do_rpc_read(boost_code ec, size_t total, - const read_rpc::ptr& in, const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - constexpr auto size = rpc::writer::default_buffer; + // Create variant http request to capture read. + const auto in = to_shared(); - if (ec) - { - // Json parser emits rpc, http and json codes. - const auto code = error::rpc_to_error_code(ec); - if (code == error::unknown) logx("rpc-read", ec); - handler(code, total); - return; - } + // Preselect rpc::request body value type. + in->body() = rpc::request{}; - async_read_some(in->buffer.prepare(size), + // Capture body and move it back into request reference. + body_read(buffer, *in, std::bind(&socket::handle_rpc_read, - shared_from_this(), _1, _2, total, in, handler)); + shared_from_this(), _1, _2, std::ref(request), in, + std::move(handler))); } // private -void socket::handle_rpc_read(boost_code ec, size_t size, size_t total, - const read_rpc::ptr& in, const count_handler& handler) NOEXCEPT +void socket::handle_rpc_read(const code& ec, size_t bytes, + const ref& out, const http::request_ptr& in, + const count_handler& handler) NOEXCEPT { - BC_ASSERT(stranded()); - - total = ceilinged_add(total, size); - if (error::asio_is_canceled(ec)) - { - handler(error::channel_stopped, total); - return; - } - - if (total > maximum_) - { - handler(error::message_overflow, total); - return; - } - if (!ec) { - in->buffer.commit(size); - const auto data = in->buffer.data(); - const auto parsed = in->reader.put(data, ec); - if (!ec) - { - in->buffer.consume(parsed); - if (in->reader.done()) - { - in->reader.finish(ec); - if (!ec) - { - handler(error::success, total); - return; - } - } - } + // Move rpc::request from http body value to caller out param. + out.get() = std::move(std::get(in->body().value())); } - // Handle error condition or incomplete message. - do_rpc_read(ec, total, in, handler); + handler(ec, bytes); } -// RPC (write). -// ---------------------------------------------------------------------------- - -void socket::rpc_write(rpc::response& response, +void socket::rpc_write(rpc::response&& response, count_handler&& handler) NOEXCEPT { - boost_code ec{}; - const auto out = emplace_shared(response); - out->writer.init(ec); - - // Dispatch success or fail, for handler invoke on strand. - boost::asio::dispatch(strand_, - std::bind(&socket::do_rpc_write, - shared_from_this(), ec, zero, out, std::move(handler))); + http::response out{}; + out.body() = std::move(response); + body_write(std::move(out), std::move(handler)); } -// private -void socket::do_rpc_write(boost_code ec, size_t total, - const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - - const auto buffer = ec ? write_rpc::out_buffer{} : out->writer.get(ec); - if (ec) - { - // Json serializer emits rpc, http and json codes. - const auto code = error::rpc_to_error_code(ec); - if (code == error::unknown) logx("rpc-write", ec); - handler(code, total); - return; - } - - BC_ASSERT(buffer.has_value()); - - async_write(buffer.value().first, false, - std::bind(&socket::handle_rpc_write, - shared_from_this(), _1, _2, total, out, handler)); -} - -// private -void socket::handle_rpc_write(boost_code ec, size_t size, size_t total, - const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - - total = ceilinged_add(total, size); - if (error::asio_is_canceled(ec)) - { - handler(error::channel_stopped, total); - return; - } - - if (!ec && out->writer.done()) - { - handler(error::success, total); - return; - } - - // Handle error condition or incomplete message. - do_rpc_write(ec, total, out, handler); -} - -/// RPC (notify). -// ---------------------------------------------------------------------------- - -void socket::rpc_notify(rpc::request& notification, +void socket::rpc_notify(rpc::request&& notification, count_handler&& handler) NOEXCEPT { - boost_code ec{}; - const auto out = emplace_shared(notification); - out->writer.init(ec); - - // Dispatch success or fail, for handler invoke on strand. - boost::asio::dispatch(strand_, - std::bind(&socket::do_rpc_notify, - shared_from_this(), ec, zero, out, std::move(handler))); -} - -// private -void socket::do_rpc_notify(boost_code ec, size_t total, - const notify_rpc::ptr& out, const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - - const auto buffer = ec ? notify_rpc::out_buffer{} : out->writer.get(ec); - if (ec) - { - // Json serializer emits rpc, http and json codes. - const auto code = error::rpc_to_error_code(ec); - if (code == error::unknown) logx("rpc-notify", ec); - handler(code, total); - return; - } - - BC_ASSERT(buffer.has_value()); - - async_write(buffer.value().first, false, - std::bind(&socket::handle_rpc_notify, - shared_from_this(), _1, _2, total, out, handler)); -} - -// private -void socket::handle_rpc_notify(boost_code ec, size_t size, size_t total, - const notify_rpc::ptr& out, const count_handler& handler) NOEXCEPT -{ - BC_ASSERT(stranded()); - - total = ceilinged_add(total, size); - if (error::asio_is_canceled(ec)) - { - handler(error::channel_stopped, total); - return; - } - - if (!ec && out->writer.done()) - { - handler(error::success, total); - return; - } - - // Handle error condition or incomplete message. - do_rpc_notify(ec, total, out, handler); + http::request out{}; + out.body() = std::move(notification); + body_notify(std::move(out), std::move(handler)); } BC_POP_WARNING() diff --git a/src/net/socket_p2p.cpp b/src/net/socket_tcp.cpp similarity index 78% rename from src/net/socket_p2p.cpp rename to src/net/socket_tcp.cpp index 4a4412100..1689c7646 100644 --- a/src/net/socket_p2p.cpp +++ b/src/net/socket_tcp.cpp @@ -29,32 +29,34 @@ using namespace std::placeholders; BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) -// P2P (read). +// TCP (read). // ---------------------------------------------------------------------------- +// Fixed-size read that completes only when the mutable_buffer is filled. -void socket::p2p_read(const asio::mutable_buffer& out, +void socket::tcp_read(const asio::mutable_buffer& out, count_handler&& handler) NOEXCEPT { boost::asio::dispatch(strand_, - std::bind(&socket::do_p2p_read, + std::bind(&socket::do_tcp_read, shared_from_this(), out, std::move(handler))); } // private -void socket::do_p2p_read(const asio::mutable_buffer& out, +void socket::do_tcp_read(const asio::mutable_buffer& out, const count_handler& handler) NOEXCEPT { BC_ASSERT(stranded()); async_read(out, handler); } -// P2P (write). +// TCP (write). // ---------------------------------------------------------------------------- +// Buffer is fully allocated before write, identical to ws_write. -void socket::p2p_write(const asio::const_buffer& in, +void socket::tcp_write(const asio::const_buffer& in, count_handler&& handler) NOEXCEPT { - raw_write(in, true, std::move(handler)); + ws_write(in, true, std::move(handler)); } BC_POP_WARNING() diff --git a/src/net/socket_ws.cpp b/src/net/socket_ws.cpp index 619e6dbeb..168fd2ea6 100644 --- a/src/net/socket_ws.cpp +++ b/src/net/socket_ws.cpp @@ -32,6 +32,47 @@ using namespace std::placeholders; BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) +// WS (read). +// ---------------------------------------------------------------------------- +// The flat_buffer's max_size() must be large enough to hold the complete ws +// framed message. The read will fail if the message exceeds this limit. + +void socket::ws_read(http::flat_buffer& out, + count_handler&& handler) NOEXCEPT +{ + boost::asio::dispatch(strand_, + std::bind(&socket::do_ws_read, + shared_from_this(), std::ref(out), std::move(handler))); +} + +// private +void socket::do_ws_read(ref out, + const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + async_read(out.get(), handler); +} + +// WS (write). +// ---------------------------------------------------------------------------- +// Buffer is fully allocated before write, identical to tcp_write. + +void socket::ws_write(const asio::const_buffer& in, bool binary, + count_handler&& handler) NOEXCEPT +{ + boost::asio::dispatch(strand_, + std::bind(&socket::do_ws_write, + shared_from_this(), in, binary, std::move(handler))); +} + +// private +void socket::do_ws_write(const asio::const_buffer& in, bool binary, + const count_handler& handler) NOEXCEPT +{ + BC_ASSERT(stranded()); + async_write(in, binary, handler); +} + // WS (event). // ---------------------------------------------------------------------------- // This is a unique/internal aspect of websockets. diff --git a/test/config/authority.cpp b/test/config/authority.cpp index 2a993867a..eeb10b268 100644 --- a/test/config/authority.cpp +++ b/test/config/authority.cpp @@ -223,7 +223,7 @@ BOOST_AUTO_TEST_CASE(authority__ip_port_cidr__ip_address__expected) BOOST_AUTO_TEST_CASE(authority__ip_port_cidr__boost_address__expected) { constexpr uint16_t expected_port = 42; - const auto address = asio::address::from_string(BC_AUTHORITY_IPV6_COMPRESSED_ADDRESS); + const auto address = boost::asio::ip::make_address(BC_AUTHORITY_IPV6_COMPRESSED_ADDRESS); const authority host(address, expected_port); BOOST_REQUIRE_EQUAL(host.port(), expected_port); BOOST_REQUIRE_EQUAL(host.ip(), address); @@ -233,7 +233,7 @@ BOOST_AUTO_TEST_CASE(authority__ip_port_cidr__boost_address__expected) BOOST_AUTO_TEST_CASE(authority__port__boost_endpoint__expected) { constexpr uint16_t expected_port = 42; - const auto address = asio::address::from_string(BC_AUTHORITY_IPV6_COMPRESSED_ADDRESS); + const auto address = boost::asio::ip::make_address(BC_AUTHORITY_IPV6_COMPRESSED_ADDRESS); const asio::endpoint tcp_endpoint(address, expected_port); const authority host(tcp_endpoint); BOOST_REQUIRE_EQUAL(host.port(), expected_port); @@ -290,14 +290,14 @@ BOOST_AUTO_TEST_CASE(authority__to_ip_address__ip_address__expected) BOOST_AUTO_TEST_CASE(authority__to_ip_address__boost_address__expected) { - const auto address = asio::address::from_string(BC_AUTHORITY_IPV6_COMPRESSED_ADDRESS); + const auto address = boost::asio::ip::make_address(BC_AUTHORITY_IPV6_COMPRESSED_ADDRESS); const authority host(address, 42); BOOST_REQUIRE(ip_equal(host.to_ip_address(), test_ipv6_address)); } BOOST_AUTO_TEST_CASE(authority__to_ip_address__boost_endpoint__expected) { - const auto address = asio::address::from_string(BC_AUTHORITY_IPV4_ADDRESS); + const auto address = boost::asio::ip::make_address(BC_AUTHORITY_IPV4_ADDRESS); const asio::endpoint tcp_endpoint(address, 42); const authority host(tcp_endpoint); BOOST_REQUIRE(ip_equal(host.to_ip_address(), test_mapped_ip_address)); diff --git a/test/config/utilities.cpp b/test/config/utilities.cpp index 4ea6aa909..8a335caf8 100644 --- a/test/config/utilities.cpp +++ b/test/config/utilities.cpp @@ -382,7 +382,10 @@ BOOST_AUTO_TEST_CASE(utilities__from_address__v6_mapped_loopback__loopback_v4) // address.to_v4() is not the same as ipv6.to_v4(), the former is a getter // while the latter is a convertor. BOOST_REQUIRE(ip.to_v6().is_v4_mapped()); - BOOST_REQUIRE(ip.to_v6().to_v4().is_loopback()); + + ////BOOST_REQUIRE(ip.to_v6().to_v4().is_loopback()); + const auto loop = boost::asio::ip::make_address_v4(boost::asio::ip::v4_mapped, ip.to_v6()); + BOOST_REQUIRE(loop.is_loopback()); } BOOST_AUTO_TEST_CASE(utilities__from_address__loopback_v6__loopback_v6) diff --git a/test/error.cpp b/test/error.cpp index 903e6a634..4db825d09 100644 --- a/test/error.cpp +++ b/test/error.cpp @@ -2149,15 +2149,6 @@ BOOST_AUTO_TEST_CASE(error_t__code__jsonrpc_params_not_collection__true_expected BOOST_REQUIRE_EQUAL(ec.message(), "jsonrpc params not collection"); } -BOOST_AUTO_TEST_CASE(error_t__code__jsonrpc_reader_bad_buffer__true_expected_message) -{ - constexpr auto value = error::jsonrpc_reader_bad_buffer; - const auto ec = code(value); - BOOST_REQUIRE(ec); - BOOST_REQUIRE(ec == value); - BOOST_REQUIRE_EQUAL(ec.message(), "jsonrpc reader bad buffer"); -} - BOOST_AUTO_TEST_CASE(error_t__code__jsonrpc_reader_stall__true_expected_message) { constexpr auto value = error::jsonrpc_reader_stall; diff --git a/test/messages/http_body_reader.cpp b/test/messages/http_body_reader.cpp index 71c02fc1d..5e7ef0e6c 100644 --- a/test/messages/http_body_reader.cpp +++ b/test/messages/http_body_reader.cpp @@ -40,12 +40,11 @@ struct accessor BOOST_AUTO_TEST_SUITE(http_body_reader_tests) -BOOST_AUTO_TEST_CASE(http_body_reader__init__bogus__constructs_empty_reader) +BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_bogus__empty_reader) { message_header header{}; header.set(http::field::content_type, "bogus"); body::value_type value{}; - value = empty_body::value_type{}; accessor reader(header, value); length_type length{ max_size_t }; boost_code ec{}; @@ -53,12 +52,24 @@ BOOST_AUTO_TEST_CASE(http_body_reader__init__bogus__constructs_empty_reader) BOOST_REQUIRE(std::holds_alternative(reader.reader_)); } -BOOST_AUTO_TEST_CASE(http_body_reader__init__plain_json__constructs_json_reader) +BOOST_AUTO_TEST_CASE(http_body_reader__init__string_body_bogus__string_reader) +{ + message_header header{}; + header.set(http::field::content_type, "bogus"); + body::value_type value{}; + value = string_body::value_type{}; + accessor reader(header, value); + length_type length{ max_size_t }; + boost_code ec{}; + reader.init(length, ec); + BOOST_REQUIRE(std::holds_alternative(reader.reader_)); +} + +BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_plain_application_json__constructs_json_reader) { message_header header{}; header.set(http::field::content_type, "application/json"); body::value_type value{}; - value = json_body::value_type{}; value.plain_json = true; accessor reader(header, value); length_type length{ max_size_t }; @@ -67,12 +78,11 @@ BOOST_AUTO_TEST_CASE(http_body_reader__init__plain_json__constructs_json_reader) BOOST_REQUIRE(std::holds_alternative(reader.reader_)); } -BOOST_AUTO_TEST_CASE(http_body_reader__init__rpc_json__constructs_rpc_reader) +BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_not_plain_application_json__constructs_rpc_reader) { message_header header{}; header.set(http::field::content_type, "application/json"); body::value_type value{}; - value = json_body::value_type{}; value.plain_json = false; accessor reader(header, value); length_type length{ max_size_t }; @@ -81,13 +91,12 @@ BOOST_AUTO_TEST_CASE(http_body_reader__init__rpc_json__constructs_rpc_reader) BOOST_REQUIRE(std::holds_alternative(reader.reader_)); } -BOOST_AUTO_TEST_CASE(http_body_reader__init__application_octet_stream__constructs_data_reader) +BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_application_octet_stream__constructs_data_reader) { message_header header{}; header.set(http::field::content_type, "application/octet-stream"); header.set(http::field::content_disposition, "bogus"); body::value_type value{}; - value = chunk_body::value_type{}; accessor reader(header, value); length_type length{ max_size_t }; boost_code ec{}; @@ -96,13 +105,12 @@ BOOST_AUTO_TEST_CASE(http_body_reader__init__application_octet_stream__construct } // TODO: linux debug boost asserts on body_.file_.is_open(). -////BOOST_AUTO_TEST_CASE(http_body_reader__init__application_octet_stream_with_attachment__constructs_file_reader) +////BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_application_octet_stream_with_attachment__constructs_file_reader) ////{ //// message_header header{}; //// header.set(http::field::content_type, "application/octet-stream"); //// header.set(http::field::content_disposition, "filename=somenonsense.jpg"); //// body::value_type value{}; -//// value = file_body::value_type{}; //// accessor reader(header, value); //// length_type length{ max_size_t }; //// boost_code ec{}; @@ -111,13 +119,12 @@ BOOST_AUTO_TEST_CASE(http_body_reader__init__application_octet_stream__construct ////} // TODO: linux debug boost asserts on body_.file_.is_open(). -////BOOST_AUTO_TEST_CASE(http_body_reader__init__application_octet_stream_with_dirty_attachment__constructs_file_reader) +////BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_application_octet_stream_with_dirty_attachment__constructs_file_reader) ////{ //// message_header header{}; //// header.set(http::field::content_type, "application/octet-stream"); //// header.set(http::field::content_disposition, "dirty 42; filename* = somenonsense.jpg; some other nonsense"); //// body::value_type value{}; -//// value = file_body::value_type{}; //// accessor reader(header, value); //// length_type length{ max_size_t }; //// boost_code ec{}; @@ -125,12 +132,11 @@ BOOST_AUTO_TEST_CASE(http_body_reader__init__application_octet_stream__construct //// BOOST_REQUIRE(std::holds_alternative(reader.reader_)); ////} -BOOST_AUTO_TEST_CASE(http_body_reader__init__text_plain__constructs_string_reader) +BOOST_AUTO_TEST_CASE(http_body_reader__init__default_body_text_plain__constructs_string_reader) { message_header header{}; header.set(http::field::content_type, "text/plain"); body::value_type value{}; - value = string_body::value_type{}; accessor reader(header, value); length_type length{ max_size_t }; boost_code ec{}; diff --git a/test/messages/rpc/body_reader.cpp b/test/messages/rpc/body_reader.cpp index c09ae872e..688074a3b 100644 --- a/test/messages/rpc/body_reader.cpp +++ b/test/messages/rpc/body_reader.cpp @@ -149,7 +149,7 @@ BOOST_AUTO_TEST_CASE(rpc_body_reader__finish__simple_request_terminated_with_new BOOST_REQUIRE(body.model.is_null()); } -BOOST_AUTO_TEST_CASE(rpc_body_reader__finish__simple_request_terminated_without_newline__success_not_done) +BOOST_AUTO_TEST_CASE(rpc_body_reader__finish__simple_request_terminated_without_newline__need_more_not_done) { const std::string_view text{ R"({"jsonrpc":"2.0","id":1,"method":"test"})" }; const asio::const_buffer buffer{ text.data(), text.size() }; @@ -163,7 +163,7 @@ BOOST_AUTO_TEST_CASE(rpc_body_reader__finish__simple_request_terminated_without_ BOOST_REQUIRE(!ec); reader.finish(ec); - BOOST_REQUIRE(!ec); + BOOST_REQUIRE(ec == error::http_error_t::need_more); BOOST_REQUIRE(!reader.done()); } diff --git a/test/net/socket.cpp b/test/net/socket.cpp index f641ca9f5..de7b1c1fb 100644 --- a/test/net/socket.cpp +++ b/test/net/socket.cpp @@ -187,7 +187,7 @@ BOOST_AUTO_TEST_CASE(socket__read__disconnected__error) const auto instance = std::make_shared(log, pool.service(), std::move(params)); system::data_array<42> data; - instance->p2p_read({ data.data(), data.size() }, + instance->tcp_read({ data.data(), data.size() }, [instance](const code& ec, size_t size) { // 10009 (WSAEBADF, invalid file handle) gets mapped to bad_stream. @@ -213,7 +213,7 @@ BOOST_AUTO_TEST_CASE(socket__write__disconnected__bad_stream) const auto instance = std::make_shared(log, pool.service(), std::move(params)); system::data_array<42> data; - instance->p2p_write({ data.data(), data.size() }, + instance->tcp_write({ data.data(), data.size() }, [instance](const code& ec, size_t size) { // 10009 (WSAEBADF, invalid file handle) gets mapped to bad_stream.