diff --git a/Makefile.am b/Makefile.am index 787bfbd1a..7f21cd408 100644 --- a/Makefile.am +++ b/Makefile.am @@ -45,6 +45,7 @@ src_libbitcoin_network_la_SOURCES = \ src/channels/channel.cpp \ src/channels/channel_http.cpp \ src/channels/channel_peer.cpp \ + src/channels/channel_rpc.cpp \ src/channels/channel_ws.cpp \ src/config/address.cpp \ src/config/authority.cpp \ @@ -160,6 +161,7 @@ test_libbitcoin_network_test_SOURCES = \ test/channels/channel.cpp \ test/channels/channel_http.cpp \ test/channels/channel_peer.cpp \ + test/channels/channel_rpc.cpp \ test/channels/channel_ws.cpp \ test/config/address.cpp \ test/config/authority.cpp \ @@ -302,6 +304,7 @@ include_bitcoin_network_channels_HEADERS = \ include/bitcoin/network/channels/channel.hpp \ include/bitcoin/network/channels/channel_http.hpp \ include/bitcoin/network/channels/channel_peer.hpp \ + include/bitcoin/network/channels/channel_rpc.hpp \ include/bitcoin/network/channels/channel_ws.hpp \ include/bitcoin/network/channels/channels.hpp @@ -450,6 +453,7 @@ include_bitcoin_network_protocols_HEADERS = \ include/bitcoin/network/protocols/protocol_ping_106.hpp \ include/bitcoin/network/protocols/protocol_ping_60001.hpp \ include/bitcoin/network/protocols/protocol_reject_70002.hpp \ + include/bitcoin/network/protocols/protocol_rpc.hpp \ include/bitcoin/network/protocols/protocol_seed_209.hpp \ include/bitcoin/network/protocols/protocol_version_106.hpp \ include/bitcoin/network/protocols/protocol_version_70001.hpp \ diff --git a/builds/cmake/CMakeLists.txt b/builds/cmake/CMakeLists.txt index c4c3899ab..f1f4f93f7 100644 --- a/builds/cmake/CMakeLists.txt +++ b/builds/cmake/CMakeLists.txt @@ -227,6 +227,7 @@ add_library( ${CANONICAL_LIB_NAME} "../../src/channels/channel.cpp" "../../src/channels/channel_http.cpp" "../../src/channels/channel_peer.cpp" + "../../src/channels/channel_rpc.cpp" "../../src/channels/channel_ws.cpp" "../../src/config/address.cpp" "../../src/config/authority.cpp" @@ -366,6 +367,7 @@ if (with-tests) "../../test/channels/channel.cpp" "../../test/channels/channel_http.cpp" "../../test/channels/channel_peer.cpp" + "../../test/channels/channel_rpc.cpp" "../../test/channels/channel_ws.cpp" "../../test/config/address.cpp" "../../test/config/authority.cpp" diff --git a/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj b/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj index 104574f37..dbf25e770 100644 --- a/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj +++ b/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj @@ -133,6 +133,7 @@ + $(IntDir)test_config_address.obj diff --git a/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters b/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters index 95405b704..2a5db43f3 100644 --- a/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters +++ b/builds/msvc/vs2022/libbitcoin-network-test/libbitcoin-network-test.vcxproj.filters @@ -102,6 +102,9 @@ src\channels + + src\channels + src\channels diff --git a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj index d9e8b5f87..56fbe5986 100644 --- a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj +++ b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj @@ -127,6 +127,7 @@ + $(IntDir)src_config_address.obj @@ -241,6 +242,7 @@ + @@ -336,6 +338,7 @@ + diff --git a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters index a414fa2f7..c43582d76 100644 --- a/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters +++ b/builds/msvc/vs2022/libbitcoin-network/libbitcoin-network.vcxproj.filters @@ -147,6 +147,9 @@ src\channels + + src\channels + src\channels @@ -473,6 +476,9 @@ include\bitcoin\network\channels + + include\bitcoin\network\channels + include\bitcoin\network\channels @@ -758,6 +764,9 @@ include\bitcoin\network\protocols + + include\bitcoin\network\protocols + include\bitcoin\network\protocols diff --git a/include/bitcoin/network.hpp b/include/bitcoin/network.hpp index dfe40bdcf..f6c7a15aa 100644 --- a/include/bitcoin/network.hpp +++ b/include/bitcoin/network.hpp @@ -42,6 +42,7 @@ #include #include #include +#include #include #include #include @@ -133,6 +134,7 @@ #include #include #include +#include #include #include #include diff --git a/include/bitcoin/network/channels/channel_rpc.hpp b/include/bitcoin/network/channels/channel_rpc.hpp new file mode 100644 index 000000000..8e78676d8 --- /dev/null +++ b/include/bitcoin/network/channels/channel_rpc.hpp @@ -0,0 +1,49 @@ +/** + * Copyright (c) 2011-2025 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 . + */ +#ifndef LIBBITCOIN_NETWORK_CHANNELS_CHANNEL_RPC_HPP +#define LIBBITCOIN_NETWORK_CHANNELS_CHANNEL_RPC_HPP + +#include +#include +#include +#include +#include + +namespace libbitcoin { +namespace network { + +/// rpc over tcp channel. +class BCT_API channel_rpc + : public channel +{ +public: + typedef std::shared_ptr ptr; + + inline channel_rpc(const logger& log, const socket::ptr& socket, + uint64_t identifier, const settings_t& settings, + const options_t& options) NOEXCEPT + : channel(log, socket, identifier, settings, options) + { + } +}; + +} // namespace network +} // namespace libbitcoin + +#endif diff --git a/include/bitcoin/network/channels/channels.hpp b/include/bitcoin/network/channels/channels.hpp index 691b50a0b..923966f13 100644 --- a/include/bitcoin/network/channels/channels.hpp +++ b/include/bitcoin/network/channels/channels.hpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #endif diff --git a/include/bitcoin/network/error.hpp b/include/bitcoin/network/error.hpp index 78f81dfb2..a0c1e0e6d 100644 --- a/include/bitcoin/network/error.hpp +++ b/include/bitcoin/network/error.hpp @@ -238,7 +238,7 @@ BCT_API bool asio_is_canceled(const boost_code& ec) NOEXCEPT; BCT_API code asio_to_error_code(const boost_code& ec) NOEXCEPT; /// 1:1 mapping of boost::beast:http::error to network (or error::unknown). -BCT_API code beast_to_error_code(const boost_code& ec) NOEXCEPT; +BCT_API code http_to_error_code(const boost_code& ec) NOEXCEPT; } // namespace error } // namespace network diff --git a/include/bitcoin/network/impl/messages/json_body.ipp b/include/bitcoin/network/impl/messages/json_body.ipp index 0c648be2a..7aa0b2f87 100644 --- a/include/bitcoin/network/impl/messages/json_body.ipp +++ b/include/bitcoin/network/impl/messages/json_body.ipp @@ -114,6 +114,12 @@ void CLASS::reader::finish(boost_code& ec) NOEXCEPT parser_.reset(); } +TEMPLATE +bool CLASS::reader::done() const NOEXCEPT +{ + return parser_.done(); +} + // json::body<>::writer // ---------------------------------------------------------------------------- diff --git a/include/bitcoin/network/messages/json_body.hpp b/include/bitcoin/network/messages/json_body.hpp index 19bf9d2e6..0b345b291 100644 --- a/include/bitcoin/network/messages/json_body.hpp +++ b/include/bitcoin/network/messages/json_body.hpp @@ -73,6 +73,7 @@ struct body virtual void init(const http::length_type& length, boost_code& ec) NOEXCEPT; virtual size_t put(const buffer_type& buffer, boost_code& ec) NOEXCEPT; virtual void finish(boost_code& ec) NOEXCEPT; + virtual bool done() const NOEXCEPT; protected: value_type& value_; @@ -86,6 +87,7 @@ struct body class writer { public: + static constexpr size_t default_buffer = 4096; using const_buffers_type = asio::const_buffer; using out_buffer = http::get_buffer; @@ -108,9 +110,6 @@ struct body protected: value_type& value_; boost::json::serializer serializer_; - - private: - static constexpr size_t default_buffer = 4096; }; }; diff --git a/include/bitcoin/network/net/proxy.hpp b/include/bitcoin/network/net/proxy.hpp index 711c3a51a..7bdd13c66 100644 --- a/include/bitcoin/network/net/proxy.hpp +++ b/include/bitcoin/network/net/proxy.hpp @@ -42,6 +42,8 @@ class BCT_API proxy public: typedef std::shared_ptr ptr; typedef subscriber<> stop_subscriber; + typedef rpc::request_body::value_type rpc_in_value; + typedef rpc::response_body::value_type rpc_out_value; DELETE_COPY_MOVE(proxy); @@ -126,19 +128,17 @@ class BCT_API proxy count_handler&& handler) NOEXCEPT; /// Send a complete TCP message to the remote endpoint. - virtual void write(const asio::const_buffer& payload, + virtual void write(const asio::const_buffer& buffer, count_handler&& handler) NOEXCEPT; /// TCP-RPC (e.g. electrum, stratum_v1). /// ----------------------------------------------------------------------- /// Read full rpc request from the socket, handler posted to socket strand. - virtual void read(rpc::response_t& out, - count_handler&& handler) NOEXCEPT; + virtual void read(rpc_in_value& out, count_handler&& handler) NOEXCEPT; /// Write full rpc response to the socket, handler posted to socket strand. - virtual void write(const rpc::response_t& in, - count_handler&& handler) NOEXCEPT; + virtual void write(rpc_out_value&& in, count_handler&& handler) NOEXCEPT; /// HTTP (generic). /// ----------------------------------------------------------------------- diff --git a/include/bitcoin/network/net/socket.hpp b/include/bitcoin/network/net/socket.hpp index c5b938228..eec1e7edc 100644 --- a/include/bitcoin/network/net/socket.hpp +++ b/include/bitcoin/network/net/socket.hpp @@ -41,6 +41,8 @@ class BCT_API socket { public: typedef std::shared_ptr ptr; + typedef rpc::request_body::value_type rpc_in_value; + typedef rpc::response_body::value_type rpc_out_value; DELETE_COPY_MOVE(socket); @@ -104,11 +106,11 @@ class BCT_API socket /// ----------------------------------------------------------------------- /// Read full rpc request from the socket, handler posted to socket strand. - virtual void rpc_read(rpc::response_t& out, + virtual void rpc_read(rpc_in_value& request, count_handler&& handler) NOEXCEPT; /// Write full rpc response to the socket, handler posted to socket strand. - virtual void rpc_write(const rpc::response_t& model, + virtual void rpc_write(rpc_out_value&& response, count_handler&& handler) NOEXCEPT; /// HTTP (generic). @@ -158,13 +160,46 @@ class BCT_API socket /// The socket was upgraded to a websocket (requires strand). virtual bool websocket() const NOEXCEPT; + /// Utility. + asio::socket& get_transport() NOEXCEPT; + void logx(const std::string& context, const boost_code& ec) const NOEXCEPT; + private: + struct read_rpc + { + typedef std::shared_ptr ptr; + using rpc_reader = rpc::request_body::reader; + + read_rpc(rpc_in_value& request_) NOEXCEPT + : value{}, reader{ value } + { + request_ = value; + } + + rpc_in_value value; + rpc_reader reader; + }; + + struct write_rpc + { + typedef std::shared_ptr ptr; + using rpc_writer = rpc::response_body::writer; + using out_buffer = rpc_writer::out_buffer; + + write_rpc(rpc_out_value&& response) NOEXCEPT + : value{ std::move(response) }, writer{ value } + { + } + + rpc_out_value value; + rpc_writer writer; + }; + // stop // ------------------------------------------------------------------------ void do_stop() NOEXCEPT; void do_async_stop() NOEXCEPT; - asio::socket& get_transport() NOEXCEPT; // wait // ------------------------------------------------------------------------ @@ -186,11 +221,9 @@ class BCT_API socket const count_handler& handler) NOEXCEPT; // tcp (rpc) - void do_rpc_read( - std::reference_wrapper out, + void do_rpc_read(boost_code ec, size_t total, const read_rpc::ptr& in, const count_handler& handler) NOEXCEPT; - void do_rpc_write( - const std::reference_wrapper& in, + void do_rpc_write(boost_code ec, size_t total, const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT; // http (generic) @@ -228,8 +261,10 @@ class BCT_API socket const count_handler& handler) NOEXCEPT; // tcp (rpc) - void handle_rpc_tcp(const boost_code& ec, size_t size, - const count_handler& handler) NOEXCEPT; + 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; // http (generic) void handle_http_read(const boost_code& ec, size_t size, diff --git a/include/bitcoin/network/protocols/protocol_rpc.hpp b/include/bitcoin/network/protocols/protocol_rpc.hpp new file mode 100644 index 000000000..2a87776bc --- /dev/null +++ b/include/bitcoin/network/protocols/protocol_rpc.hpp @@ -0,0 +1,50 @@ +/** + * Copyright (c) 2011-2025 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 . + */ +#ifndef LIBBITCOIN_NETWORK_PROTOCOL_RPC_HPP +#define LIBBITCOIN_NETWORK_PROTOCOL_RPC_HPP + +#include +#include +#include +#include +#include + +namespace libbitcoin { +namespace network { + +class BCT_API protocol_rpc + : public protocol +{ +public: + typedef std::shared_ptr ptr; + using channel_t = channel_rpc; + using options_t = channel_t::options_t; + +protected: + inline protocol_rpc(const session::ptr& session, + const channel::ptr& channel, const options_t&) NOEXCEPT + : protocol(session, channel) + { + } +}; + +} // namespace network +} // namespace libbitcoin + +#endif diff --git a/include/bitcoin/network/protocols/protocols.hpp b/include/bitcoin/network/protocols/protocols.hpp index 4673b45e8..8eec0255e 100644 --- a/include/bitcoin/network/protocols/protocols.hpp +++ b/include/bitcoin/network/protocols/protocols.hpp @@ -20,10 +20,11 @@ #define LIBBITCOIN_NETWORK_PROTOCOLS_HPP #include + +// peer #include #include #include -#include #include #include #include @@ -33,6 +34,10 @@ #include #include #include + +// server +#include +#include #include #endif diff --git a/src/channels/channel_http.cpp b/src/channels/channel_http.cpp index f309446d2..e9572ce3d 100644 --- a/src/channels/channel_http.cpp +++ b/src/channels/channel_http.cpp @@ -152,7 +152,7 @@ void channel_http::send(response&& response, result_handler&& handler) NOEXCEPT { BC_ASSERT(stranded()); - size_json_buffer(response); + assign_json_buffer(response); const auto ptr = system::move_shared(std::move(response)); count_handler complete = std::bind(&channel_http::handle_send, shared_from_base(), _1, _2, ptr, std::move(handler)); @@ -176,7 +176,7 @@ void channel_http::handle_send(const code& ec, size_t, response_ptr&, } // private -void channel_http::size_json_buffer(response& response) NOEXCEPT +void channel_http::assign_json_buffer(response& response) NOEXCEPT { if (const auto& body = response.body(); body.contains()) diff --git a/src/channels/channel_rpc.cpp b/src/channels/channel_rpc.cpp new file mode 100644 index 000000000..fe47d4c90 --- /dev/null +++ b/src/channels/channel_rpc.cpp @@ -0,0 +1,27 @@ +/** + * Copyright (c) 2011-2025 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 + +namespace libbitcoin { +namespace network { + +} // namespace network +} // namespace libbitcoin diff --git a/src/error.cpp b/src/error.cpp index 431df0ea3..01d670a23 100644 --- a/src/error.cpp +++ b/src/error.cpp @@ -316,7 +316,7 @@ code asio_to_error_code(const boost_code& ec) NOEXCEPT return error::unknown; } -code beast_to_error_code(const boost_code& ec) NOEXCEPT +code http_to_error_code(const boost_code& ec) NOEXCEPT { namespace beast = boost::beast::http; diff --git a/src/net/proxy.cpp b/src/net/proxy.cpp index 91117e5c2..e1010c97e 100644 --- a/src/net/proxy.cpp +++ b/src/net/proxy.cpp @@ -194,18 +194,18 @@ void proxy::read(const asio::mutable_buffer& buffer, socket_->read(buffer, std::move(handler)); } -void proxy::write(const asio::const_buffer& payload, +void proxy::write(const asio::const_buffer& buffer, count_handler&& handler) NOEXCEPT { boost::asio::dispatch(strand(), std::bind(&proxy::do_write, - shared_from_this(), payload, std::move(handler))); + shared_from_this(), buffer, std::move(handler))); } // TCP-RPC. // ---------------------------------------------------------------------------- -void proxy::read(rpc::response_t& out, count_handler&& handler) NOEXCEPT +void proxy::read(rpc_in_value& out, count_handler&& handler) NOEXCEPT { boost::asio::dispatch(strand(), std::bind(&proxy::waiting, shared_from_this())); @@ -213,9 +213,9 @@ void proxy::read(rpc::response_t& out, count_handler&& handler) NOEXCEPT socket_->rpc_read(out, std::move(handler)); } -void proxy::write(const rpc::response_t& in, count_handler&& handler) NOEXCEPT +void proxy::write(rpc_out_value&& in, count_handler&& handler) NOEXCEPT { - socket_->rpc_write(in, std::move(handler)); + socket_->rpc_write(std::move(in), std::move(handler)); // TODO: compose? ////boost::asio::dispatch(strand(), diff --git a/src/net/socket.cpp b/src/net/socket.cpp index b532ea2f5..98432921d 100644 --- a/src/net/socket.cpp +++ b/src/net/socket.cpp @@ -29,8 +29,9 @@ namespace libbitcoin { namespace network { - + using namespace system; +using namespace network::rpc; using namespace std::placeholders; namespace beast = boost::beast; @@ -140,13 +141,6 @@ void socket::do_async_stop() NOEXCEPT std::bind(&socket::do_stop, shared_from_this())); } -asio::socket& socket::get_transport() NOEXCEPT -{ - BC_ASSERT(stranded()); - - return websocket() ? beast::get_lowest_layer(*websocket_) : socket_; -} - // Wait. // ---------------------------------------------------------------------------- @@ -182,6 +176,7 @@ void socket::handle_wait(const boost_code& ec, if (ec) { + logx("wait", ec); handler(error::asio_to_error_code(ec)); return; } @@ -285,19 +280,28 @@ void socket::write(const asio::const_buffer& in, /// TCP-RPC. // ---------------------------------------------------------------------------- -void socket::rpc_read(rpc::response_t& out, count_handler&& handler) NOEXCEPT +void socket::rpc_read(rpc_in_value& request, count_handler&& handler) NOEXCEPT { + boost_code ec{}; + const auto in = emplace_shared(request); + in->reader.init({}, ec); + boost::asio::dispatch(strand_, std::bind(&socket::do_rpc_read, - shared_from_this(), std::ref(out), std::move(handler))); + shared_from_this(), ec, zero, in, std::move(handler))); } -void socket::rpc_write(const rpc::response_t& in, +void socket::rpc_write(rpc_out_value&& response, count_handler&& handler) NOEXCEPT { + boost_code ec{}; + const auto out = emplace_shared(std::move(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(), std::cref(in), std::move(handler))); + shared_from_this(), ec, zero, out, std::move(handler))); } // HTTP. @@ -406,23 +410,52 @@ void socket::do_write(const asio::const_buffer& in, // tcp (rpc). // ---------------------------------------------------------------------------- -void socket::do_rpc_read(std::reference_wrapper /* out */, +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 = write_rpc::rpc_writer::default_buffer; - // TODO: implement and land on handle_rpc_tcp. - handle_rpc_tcp(boost_code{}, size_t{}, handler); + if (ec) + { + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("rpc-read", ec); + handler(code, total); + return; + } + + if (is_null(in->value.buffer)) + in->value.buffer = to_shared(); + + get_transport().async_receive(in->value.buffer->prepare(size), + std::bind(&socket::handle_rpc_read, + shared_from_this(), _1, _2, total, in, handler)); } -void socket::do_rpc_write( - const std::reference_wrapper& /* in */, - const count_handler& handler) NOEXCEPT +void socket::do_rpc_write(boost_code ec, size_t total, + const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT { BC_ASSERT(stranded()); - // TODO: implement and land on handle_rpc_tcp. - handle_rpc_tcp(boost_code{}, size_t{}, handler); + const auto buffer = ec ? write_rpc::out_buffer{} : out->writer.get(ec); + if (ec) + { + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("rpc-write", ec); + handler(code, total); + return; + } + + // Finished. + if (!buffer->second) + { + handler(error::success, total); + return; + } + + get_transport().async_send(buffer->first, + std::bind(&socket::handle_rpc_write, + shared_from_this(), _1, _2, total, out, handler)); } // http (generic). @@ -565,12 +598,7 @@ void socket::handle_accept(const boost_code& ec, } const auto code = error::asio_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw accept code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + if (code == error::unknown) logx("accept", ec); handler(code); } @@ -592,12 +620,7 @@ void socket::handle_connect(const boost_code& ec, } const auto code = error::asio_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw connect code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + if (code == error::unknown) logx("connect", ec); handler(code); } @@ -616,37 +639,69 @@ void socket::handle_tcp(const boost_code& ec, size_t size, } const auto code = error::asio_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw tcp code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + if (code == error::unknown) logx("tcp", ec); handler(code, size); } // tcp (rpc) // ---------------------------------------------------------------------------- -void socket::handle_rpc_tcp(const boost_code& ec, size_t size, - const count_handler& handler) NOEXCEPT +void socket::handle_rpc_read(boost_code ec, size_t size, size_t total, + const read_rpc::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, size); + handler(error::channel_stopped, total); return; } - const auto code = error::asio_to_error_code(ec); - if (code == error::unknown) + if (!ec) { - LOGX("Raw tcp code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); + in->value.buffer->commit(size); + const auto data = in->value.buffer->data(); + const auto parsed = in->reader.put(data, ec); + if (!ec) + { + if (parsed < data.size()) + { + handler(error::unexpected_body, total); + return; + } + + in->value.buffer->consume(parsed); + if (in->reader.done()) + { + in->reader.finish(ec); + + // Finished. + if (!ec) + { + handler(error::success, total); + return; + } + } + } } - handler(code, size); + do_rpc_read(ec, total, in, handler); +} + +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; + } + + do_rpc_write(ec, total, out, handler); } // http (generic) @@ -670,13 +725,8 @@ void socket::handle_http_read(const boost_code& ec, size_t size, return; } - const auto code = error::beast_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw beast code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("http", ec); handler(code, size); } @@ -691,13 +741,8 @@ void socket::handle_http_write(const boost_code& ec, size_t size, return; } - const auto code = error::beast_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw beast code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("http", ec); handler(code, size); } @@ -716,13 +761,8 @@ void socket::handle_ws_read(const boost_code& ec, size_t size, return; } - const auto code = error::beast_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw beast code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("ws-read", ec); handler(code, size /*, websocket_->got_binary() */); } @@ -738,13 +778,8 @@ void socket::handle_ws_write(const boost_code& ec, size_t size, return; } - const auto code = error::beast_to_error_code(ec); - if (code == error::unknown) - { - LOGX("Raw beast code (" << ec.value() << ") " << ec.category().name() - << ":" << ec.message()); - } - + const auto code = error::http_to_error_code(ec); + if (code == error::unknown) logx("ws-write", ec); handler(code, size /*, websocket_->got_binary() */); } @@ -853,6 +888,23 @@ code socket::set_websocket(const http::request& request) NOEXCEPT } } +// utility +// ---------------------------------------------------------------------------- + +asio::socket& socket::get_transport() NOEXCEPT +{ + BC_ASSERT(stranded()); + + return websocket() ? beast::get_lowest_layer(*websocket_) : socket_; +} + +void socket::logx(const std::string& context, + const boost_code& ec) const NOEXCEPT +{ + LOGX("Socket " << context << " error (" << ec.value() << ") " + << ec.category().name() << ":" << ec.message()); +} + BC_POP_WARNING() } // namespace network diff --git a/test/channels/channel_rpc.cpp b/test/channels/channel_rpc.cpp new file mode 100644 index 000000000..7158347bb --- /dev/null +++ b/test/channels/channel_rpc.cpp @@ -0,0 +1,28 @@ +/** + * Copyright (c) 2011-2025 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 "../test.hpp" + +BOOST_AUTO_TEST_SUITE(channel_rpc_tests) + +BOOST_AUTO_TEST_CASE(channel_rpc_test) +{ + BOOST_REQUIRE(true); +} + +BOOST_AUTO_TEST_SUITE_END()