41739478557

Committed 03 Nov 2026 11:28PM UTC coverage: 93.912% (-0.1%) from 94.038%

Build # 41739478557

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1004 from evoskuil/master

Define rpc body members w/o specialize, dispatch without cross-thread wakes.

Coverage Stats

209 of 219 new or added lines in 4 files covered. (95.43%)

134 existing lines in 39 files now uncovered.

14531 of 15473 relevant lines covered (93.91%)

2488.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.48

/src/net/socket_http.cpp

/**
 * Copyright (c) 2011-2026 libbitcoin developers
 *
 * 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 <http://www.gnu.org/licenses/>.
 */
#include <bitcoin/network/net/socket.hpp>

#include <bitcoin/network/config/config.hpp>
#include <bitcoin/network/define.hpp>
#include <bitcoin/network/log/log.hpp>
#include <bitcoin/network/messages/messages.hpp>
#include <bitcoin/network/settings.hpp>

namespace libbitcoin {
namespace network {
    
using namespace system;
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)

// DETECT (first read).
// ----------------------------------------------------------------------------

// local
constexpr bool is_json_open(char byte) NOEXCEPT
{
    // A json value opens with an object or array, and neither is a valid
    // http method token character (rfc9110 tchar).
    return byte == '{' || byte == '[';
}

// local
constexpr bool is_line_break(char byte) NOEXCEPT
{
    // A server tolerates empty lines preceding a request line (rfc9112).
    return byte == '\r' || byte == '\n';
}

// local
// False implies inconclusive (the buffer holds only empty lines).
inline bool detect_json(bool& json, const http::flat_buffer& buffer) NOEXCEPT
{
    const auto data = buffer.data();
    const std::string_view bytes{ static_cast<const char*>(data.data()),
        data.size() };

    for (const auto byte: bytes)
    {
        if (is_line_break(byte))
            continue;

        json = is_json_open(byte);
        return true;
    }

    return false;
}

void socket::detect(http::flat_buffer& buffer,
    count_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_detect, shared_from_this(),
            std::ref(buffer), std::move(handler)));
}

// private
void socket::do_detect(const ref<http::flat_buffer>& buffer,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    bool json{};
    if (detect_json(json, buffer.get()))
    {
        detected_.store(true);
        downgraded_.store(json);
        handler(error::success, zero);
        return;
    }

    // Tolerated empty lines only, so read more (bounded by the buffer).
    async_read(buffer.get(),
        std::bind(&socket::handle_detect,
            shared_from_this(), _1, _2, buffer, handler));
}

// private
void socket::handle_detect(const code& ec, size_t size,
    const ref<http::flat_buffer>& buffer,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (ec)
    {
        handler(ec, size);
        return;
    }

    // async_read prepares, the caller commits (as does handle_body_read).
    buffer.get().commit(size);
    do_detect(buffer, handler);
}

// HTTP/WS (read).
// ----------------------------------------------------------------------------

void socket::http_read(http::flat_buffer& buffer,
    http::request& request, count_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_http_read, shared_from_this(),
            std::ref(buffer), std::ref(request), std::move(handler)));
}

// private
void socket::do_http_read(ref<http::flat_buffer> buffer,
    const ref<http::request>& request,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    async_read_http(buffer.get(), request.get(), handler);
}

// private
void socket::handle_http_read(const code& ec, size_t size,
    const ref<http::request>& request, const http_parser_ptr& parser,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (!ec && http::is_upgrade(parser->get()))
    {
        request.get() = parser->release();
        handler(error::upgrade, size);
        return;
    }

    if (!ec)
    {
        request.get() = parser->release();
        if (request.get().method() == http::verb::unknown)
        {
            handler(error::bad_method, size);
            return;
        }
    }

    handler(ec, size);
}

// HTTP/WS (write).
// ----------------------------------------------------------------------------

void socket::http_write(http::response&& response,
    count_handler&& handler) NOEXCEPT
{
    const auto out = move_shared(std::move(response));
    io::dispatch(strand_,
        std::bind(&socket::do_http_write, shared_from_this(),
            out, std::move(handler)));
}

// private
void socket::do_http_write(const http::response_ptr& response,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    async_write_http(std::move(*response), handler);
}

// private
void socket::handle_http_write(const code& ec, size_t size,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    reclaim_buffer();
    handler(ec, size);
}

// Codec (parse).
// ----------------------------------------------------------------------------
// The buffered bytes are parsed before reading more. The header parse returns
// once the header is complete, the some parse once it progresses, and the
// full parse once the message is complete. A body reader may pause any parse
// (need_buffer), which is returned to the caller. A websocket reads each
// message whole, and other transports read the bytes available.

// private
template <class Parser>
void socket::do_parse(const ref<http::flat_buffer>& buffer,
    const std::shared_ptr<Parser>& parser, parse_mode mode, size_t total,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    auto& in = buffer.get();

    while (true)
    {
        code ec{};
        const auto used = parser->put(in.data(), ec);
        in.consume(used);
        total = ceilinged_add(total, used);

        if (ec == error::http_error_t::need_more)
            break;

        if (ec)
        {
            const auto code = error::rpc_to_error_code(ec);
            if (code == error::unknown) logx("parse", ec);
            handler(code, total);
            return;
        }

        if (parser->is_done() || (mode == parse_mode::header &&
            parser->is_header_done()) || (mode == parse_mode::some &&
            !is_zero(used)))
        {
            handler(error::success, total);
            return;
        }

        if (is_zero(used))
            break;
    }

    if (websocket())
    {
        async_read(in,
            std::bind(&socket::handle_parse<Parser>,
                shared_from_this(), _1, _2, buffer, parser, mode, total,
                handler));
        return;
    }

    // Read into available capacity, growing by up to the header limit.
    const auto limit = floored_subtract(in.max_size(), in.size());
    const auto size = std::min(limit, std::max(in.capacity() - in.size(),
        header_));

    if (is_zero(size))
    {
        handler(error::buffer_overflow, total);
        return;
    }

    async_read_some(in.prepare(size),
        std::bind(&socket::handle_parse<Parser>,
            shared_from_this(), _1, _2, buffer, parser, mode, total,
            handler));
}

// private
template <class Parser>
void socket::handle_parse(const code& ec, size_t size,
    const ref<http::flat_buffer>& buffer,
    const std::shared_ptr<Parser>& parser, parse_mode mode, size_t total,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // The end of the stream completes a body so delimited, or is premature.
    if (ec == error::peer_disconnect)
    {
        code eof{};
        parser->put_eof(eof);
        if (eof)
            handler(error::http_to_error_code(eof), total);
        else
            handler(parser->is_done() ? error::success : ec, total);

        return;
    }

    if (ec)
    {
        handler(ec, total);
        return;
    }

    if (!websocket())
        buffer.get().commit(size);

    do_parse(buffer, parser, mode, total, handler);
}

// Codec (serialize).
// ----------------------------------------------------------------------------
// Each set of buffers is written in turn. A websocket writes each buffer as a
// frame of one message, which the last buffer closes.

// private
template <class Serializer>
void socket::do_serialize(
    const std::shared_ptr<serialize_state<Serializer>>& out, size_t total,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    while (true)
    {
        code ec{};
        out->buffers = out->serializer.next(ec);
        if (ec)
        {
            const auto code = error::http_to_error_code(ec);
            if (code == error::unknown) logx("serialize", ec);
            handler(code, total);
            return;
        }

        if (!out->buffers.empty())
            break;

        if (out->serializer.is_done())
        {
            handler(error::success, total);
            return;
        }
    }

    if (websocket())
    {
        BC_ASSERT(is_one(out->buffers.size()));
        ws_send_data(out->buffers.front(), out->serializer.binary(),
            out->serializer.is_done(),
            std::bind(&socket::handle_serialize<Serializer>,
                shared_from_this(), _1, _2, out, total, handler));
        return;
    }

    VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),
        out->buffers, std::bind(&socket::handle_serialize<Serializer>,
            shared_from_this(), _1, _2, out, total, handler));
}

// private
template <class Serializer>
void socket::handle_serialize(const code& ec, size_t size,
    const std::shared_ptr<serialize_state<Serializer>>& out, size_t total,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    total = ceilinged_add(total, size);
    if (ec == error::operation_canceled)
    {
        handler(error::channel_stopped, total);
        return;
    }

    if (ec)
    {
        if (ec == error::unknown) logx("write", ec);
        handler(ec, total);
        return;
    }

    do_serialize(out, total, handler);
}

template void socket::do_parse<socket::http_parser>(
    const ref<http::flat_buffer>&, const http_parser_ptr&, parse_mode, size_t,
    const count_handler&) NOEXCEPT;
template void socket::do_parse<socket::body_parser>(
    const ref<http::flat_buffer>&, const body_parser_ptr&, parse_mode, size_t,
    const count_handler&) NOEXCEPT;
template void socket::do_serialize<socket::http_serializer>(
    const std::shared_ptr<serialize_state<http_serializer>>&, size_t,
    const count_handler&) NOEXCEPT;
template void socket::do_serialize<socket::body_serializer>(
    const std::shared_ptr<serialize_state<body_serializer>>&, size_t,
    const count_handler&) NOEXCEPT;
template void socket::do_serialize<socket::notify_serializer>(
    const std::shared_ptr<serialize_state<notify_serializer>>&, size_t,
    const count_handler&) NOEXCEPT;

// HTTP (progressive read).
// ----------------------------------------------------------------------------
// The caller owns the parser, which owns http framing (content length and
// chunk decoding) across the message. A batched json-rpc body delivers each
// message from the paused parse (need_buffer), mapped to success here.

void socket::http_read_header(http::flat_buffer& buffer,
    const http_parser_ptr& parser, count_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_http_read_header, shared_from_this(),
            std::ref(buffer), parser, std::move(handler)));
}

// private
void socket::do_http_read_header(ref<http::flat_buffer> buffer,
    const http_parser_ptr& parser, const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Websocket messages have no http headers.
    if (websocket())
    {
        handler(error::operation_failed, {});
        return;
    }

    parser->body_limit(maximum_);
    parser->header_limit(header_);
    do_parse(buffer, parser, parse_mode::header, zero,
        std::bind(&socket::handle_http_read_header,
            shared_from_this(), _1, _2, parser, handler));
}

// private
void socket::handle_http_read_header(const code& ec, size_t size,
    const http_parser_ptr& parser, const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (!ec && http::is_upgrade(parser->get()))
    {
        handler(error::upgrade, size);
        return;
    }

    if (!ec && parser->get().method() == http::verb::unknown)
    {
        handler(error::bad_method, size);
        return;
    }

    handler(ec, size);
}

void socket::http_read_some(http::flat_buffer& buffer,
    const http_parser_ptr& parser, count_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_http_read_body, shared_from_this(),
            std::ref(buffer), parser, parse_mode::some,
            std::move(handler)));
}

void socket::http_read_body(http::flat_buffer& buffer,
    const http_parser_ptr& parser, count_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_http_read_body, shared_from_this(),
            std::ref(buffer), parser, parse_mode::all, std::move(handler)));
}

// private
void socket::do_http_read_body(ref<http::flat_buffer> buffer,
    const http_parser_ptr& parser, parse_mode mode,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Websocket messages have no http framing.
    if (websocket())
    {
        handler(error::operation_failed, {});
        return;
    }

    do_parse(buffer, parser, mode, zero, handler);
}

// HTTP (progressive write).
// ----------------------------------------------------------------------------
// The header is written once (chunked, no content length), and each response
// part is then written as chunk data, the batch close part terminating the
// chunked body. This mirrors the progressive read (nothing is materialized).

void socket::http_write_header(http::response&& response,
    count_handler&& handler) NOEXCEPT
{
    const auto out = emplace_shared<header_state>(
        http::serialize_header(response.base()));

    io::dispatch(strand_,
        std::bind(&socket::do_http_write_header, shared_from_this(),
            out, std::move(handler)));
}

// private
void socket::do_http_write_header(const header_state::ptr& out,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Websocket messages have no http headers.
    if (websocket())
    {
        handler(error::operation_failed, {});
        return;
    }

    const io::const_buffer data{ out->header.data(),
        out->header.size() };

    VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),
        data, std::bind(&socket::handle_http_write_header,
            shared_from_this(), _1, _2, out, handler));
}

// private
void socket::handle_http_write_header(const code& ec, size_t size,
    const header_state::ptr&, const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (ec == error::operation_canceled)
    {
        handler(error::channel_stopped, size);
        return;
    }

    if (ec == error::unknown) logx("http-write-header", ec);
    handler(ec, size);
}

void socket::rpc_write_chunk(rpc::response&& response,
    count_handler&& handler) NOEXCEPT
{
    http::response wrapper{};
    wrapper.body() = std::move(response);
    assign_buffer(wrapper.body());

    code ec{};
    const auto out = emplace_shared<chunk_state>(std::move(wrapper));
    out->writer.init(ec);

    io::dispatch(strand_,
        std::bind(&socket::do_chunk_write,
            shared_from_this(), ec, zero, out, std::move(handler)));
}

// private
void socket::do_chunk_write(code ec, size_t total,
    const chunk_state::ptr& out, const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Websocket messages have no http framing.
    if (websocket())
    {
        handler(error::operation_failed, total);
        return;
    }

    const auto buffer = ec ? chunk_state::out_buffer{} : out->writer.get(ec);
    if (ec)
    {
        const auto code = error::http_to_error_code(ec);
        if (code == error::unknown) logx("chunk-write", ec);
        handler(code, total);
        return;
    }

    if (!buffer.has_value())
    {
        out->more = false;
        handle_chunk_write({}, zero, total, out, handler);
        return;
    }

    out->more = buffer.value().second;
    const auto& data = buffer.value().first;

    // Frame the part buffer as http chunk data (chunk_state lifetime).
    out->prefix = http::chunk_header(data.size());
    out->buffers =
    {
        io::const_buffer{ out->prefix.data(), out->prefix.size() },
        data,
        io::const_buffer{ http::chunk_end.data(), http::chunk_end.size() }
    };

    VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),
        out->buffers, std::bind(&socket::handle_chunk_write,
            shared_from_this(), _1, _2, total, out, handler));
}

// private
void socket::handle_chunk_write(const code& ec, size_t size,
    size_t total, const chunk_state::ptr& out,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    total = ceilinged_add(total, size);
    if (ec == error::operation_canceled)
    {
        reclaim_buffer();
        handler(error::channel_stopped, total);
        return;
    }

    if (ec)
    {
        reclaim_buffer();
        if (ec == error::unknown) logx("chunk-write", ec);
        handler(ec, total);
        return;
    }

    if (out->more)
    {
        do_chunk_write({}, total, out, handler);
        return;
    }

    reclaim_buffer();

    // Only the batch close part terminates the chunked body.
    if (!out->last())
    {
        handler(error::success, total);
        return;
    }

    const io::const_buffer last{ http::last_chunk.data(),
        http::last_chunk.size() };

    VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),
        last, std::bind(&socket::handle_chunk_last,
            shared_from_this(), _1, _2, total, handler));
}

// private
void socket::handle_chunk_last(const code& ec, size_t size,
    size_t total, const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    total = ceilinged_add(total, size);
    if (ec == error::operation_canceled)
    {
        handler(error::channel_stopped, total);
        return;
    }

    if (ec == error::unknown) logx("chunk-last", ec);
    handler(ec, total);
}

BC_POP_WARNING()
BC_POP_WARNING()
BC_POP_WARNING()

} // namespace network
} // namespace libbitcoin

1	/**
2	* Copyright (c) 2011-2026 libbitcoin developers
3	*
4	* This file is part of libbitcoin.
5	*
6	* This program is free software: you can redistribute it and/or modify
7	* it under the terms of the GNU Affero General Public License as published by
8	* the Free Software Foundation, either version 3 of the License, or
9	* (at your option) any later version.
10	*
11	* This program is distributed in the hope that it will be useful,
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	* GNU Affero General Public License for more details.
15	*
16	* You should have received a copy of the GNU Affero General Public License
17	* along with this program. If not, see <http://www.gnu.org/licenses/>.
18	*/
19	#include <bitcoin/network/net/socket.hpp>
20
21	#include <bitcoin/network/config/config.hpp>
22	#include <bitcoin/network/define.hpp>
23	#include <bitcoin/network/log/log.hpp>
24	#include <bitcoin/network/messages/messages.hpp>
25	#include <bitcoin/network/settings.hpp>
26
27	namespace libbitcoin {
28	namespace network {
29
30	using namespace system;
31	using namespace std::placeholders;
32
33	// Shared pointers required in handler parameters so closures control lifetime.
34	BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR)
35	BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED)
36	BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
37
38	// DETECT (first read).
39	// ----------------------------------------------------------------------------
40
41	// local
42	constexpr bool is_json_open(char byte) NOEXCEPT	12 ✔
43	{
44	// A json value opens with an object or array, and neither is a valid
45	// http method token character (rfc9110 tchar).
46	return byte == '{' \|\| byte == '[';	12 ✔
47	}
48
49	// local
50	constexpr bool is_line_break(char byte) NOEXCEPT	15 ✔
51	{
52	// A server tolerates empty lines preceding a request line (rfc9112).
53	return byte == '\r' \|\| byte == '\n';	15 ✔
54	}
55
56	// local
57	// False implies inconclusive (the buffer holds only empty lines).
58	inline bool detect_json(bool& json, const http::flat_buffer& buffer) NOEXCEPT	26 ✔
59	{
60	const auto data = buffer.data();	26 ✔
61	const std::string_view bytes{ static_cast<const char*>(data.data()),	26 ✔
62	data.size() };	26 ✔
63
64	for (const auto byte: bytes)	29 ✔
65	{
66	if (is_line_break(byte))	15 ✔
67	continue;
68
69	json = is_json_open(byte);	12 ✔
70	return true;	12 ✔
71	}
72
73	return false;
74	}
75
76	void socket::detect(http::flat_buffer& buffer,	14 ✔
77	count_handler&& handler) NOEXCEPT
78	{
79	io::dispatch(strand_,	14 ✔
80	std::bind(&socket::do_detect, shared_from_this(),	28 ✔
81	std::ref(buffer), std::move(handler)));	14 ✔
82	}	14 ✔
83
84	// private
85	void socket::do_detect(const ref<http::flat_buffer>& buffer,	26 ✔
86	const count_handler& handler) NOEXCEPT
87	{
88	BC_ASSERT(stranded());	26 ✔
89
90	bool json{};	26 ✔
91	if (detect_json(json, buffer.get()))	26 ✔
92	{
93	detected_.store(true);	12 ✔
94	downgraded_.store(json);	12 ✔
95	handler(error::success, zero);	12 ✔
96	return;	12 ✔
97	}
98
99	// Tolerated empty lines only, so read more (bounded by the buffer).
100	async_read(buffer.get(),	14 ✔
101	std::bind(&socket::handle_detect,	28 ✔
102	shared_from_this(), _1, _2, buffer, handler));	28 ✔
103	}
104
105	// private
106	void socket::handle_detect(const code& ec, size_t size,	14 ✔
107	const ref<http::flat_buffer>& buffer,
108	const count_handler& handler) NOEXCEPT
109	{
110	BC_ASSERT(stranded());	14 ✔
111
112	if (ec)	14 ✔
113	{
114	handler(ec, size);	2 ✔
115	return;	2 ✔
116	}
117
118	// async_read prepares, the caller commits (as does handle_body_read).
119	buffer.get().commit(size);	12 ✔
120	do_detect(buffer, handler);	12 ✔
121	}
122
123	// HTTP/WS (read).
124	// ----------------------------------------------------------------------------
125
126	void socket::http_read(http::flat_buffer& buffer,	22 ✔
127	http::request& request, count_handler&& handler) NOEXCEPT
128	{
129	io::dispatch(strand_,	22 ✔
130	std::bind(&socket::do_http_read, shared_from_this(),	44 ✔
131	std::ref(buffer), std::ref(request), std::move(handler)));	22 ✔
132	}	22 ✔
133
134	// private
135	void socket::do_http_read(ref<http::flat_buffer> buffer,	22 ✔
136	const ref<http::request>& request,
137	const count_handler& handler) NOEXCEPT
138	{
139	BC_ASSERT(stranded());	22 ✔
140	async_read_http(buffer.get(), request.get(), handler);	22 ✔
141	}	22 ✔
142
143	// private
144	void socket::handle_http_read(const code& ec, size_t size,	18 ✔
145	const ref<http::request>& request, const http_parser_ptr& parser,
146	const count_handler& handler) NOEXCEPT
147	{
148	BC_ASSERT(stranded());	18 ✔
149
150	if (!ec && http::is_upgrade(parser->get()))	18 ✔
151	{
152	request.get() = parser->release();	12 ✔
153	handler(error::upgrade, size);	12 ✔
154	return;	12 ✔
155	}
156
157	if (!ec)	6 ✔
158	{
159	request.get() = parser->release();	3 ✔
160	if (request.get().method() == http::verb::unknown)	3 ✔
161	{
162	handler(error::bad_method, size);	1 ✔
163	return;	1 ✔
164	}
165	}
166
167	handler(ec, size);	5 ✔
168	}
169
170	// HTTP/WS (write).
171	// ----------------------------------------------------------------------------
172
173	void socket::http_write(http::response&& response,	21 ✔
174	count_handler&& handler) NOEXCEPT
175	{
176	const auto out = move_shared(std::move(response));	21 ✔
177	io::dispatch(strand_,	21 ✔
178	std::bind(&socket::do_http_write, shared_from_this(),	42 ✔
179	out, std::move(handler)));	21 ✔
180	}	21 ✔
181
182	// private
183	void socket::do_http_write(const http::response_ptr& response,	21 ✔
184	const count_handler& handler) NOEXCEPT
185	{
186	BC_ASSERT(stranded());	21 ✔
187	async_write_http(std::move(*response), handler);	21 ✔
188	}	21 ✔
189
190	// private
191	void socket::handle_http_write(const code& ec, size_t size,	409 ✔
192	const count_handler& handler) NOEXCEPT
193	{
194	BC_ASSERT(stranded());	409 ✔
195	reclaim_buffer();	409 ✔
196	handler(ec, size);	409 ✔
197	}	409 ✔
198
199	// Codec (parse).
200	// ----------------------------------------------------------------------------
201	// The buffered bytes are parsed before reading more. The header parse returns
202	// once the header is complete, the some parse once it progresses, and the
203	// full parse once the message is complete. A body reader may pause any parse
204	// (need_buffer), which is returned to the caller. A websocket reads each
205	// message whole, and other transports read the bytes available.
206
207	// private
208	template <class Parser>
209	void socket::do_parse(const ref<http::flat_buffer>& buffer,	183 ✔
210	const std::shared_ptr<Parser>& parser, parse_mode mode, size_t total,
211	const count_handler& handler) NOEXCEPT
212	{
213	BC_ASSERT(stranded());	183 ✔
214	auto& in = buffer.get();	183 ✔
215
216	while (true)	×
217	{
218	code ec{};	183 ✔
219	const auto used = parser->put(in.data(), ec);	183 ✔
220	in.consume(used);	183 ✔
221	total = ceilinged_add(total, used);	183 ✔
222
223	if (ec == error::http_error_t::need_more)	183 ✔
224	break;
225
226	if (ec)	100 ✔
227	{
228	const auto code = error::rpc_to_error_code(ec);	23 ✔
229	if (code == error::unknown) logx("parse", ec);	23 ✔
230	handler(code, total);	23 ✔
231	return;
232	}
233
234	if (parser->is_done() \|\| (mode == parse_mode::header &&	10 ✔
235	parser->is_header_done()) \|\| (mode == parse_mode::some &&	87 ✔
UNCOV 236	!is_zero(used)))	×
237	{
238	handler(error::success, total);	77 ✔
239	return;	77 ✔
240	}
241
242	if (is_zero(used))	×
243	break;
244	}
245
246	if (websocket())	83 ✔
247	{
248	async_read(in,	12 ✔
249	std::bind(&socket::handle_parse<Parser>,	12 ✔
250	shared_from_this(), _1, _2, buffer, parser, mode, total,
251	handler));
252	return;	6 ✔
253	}
254
255	// Read into available capacity, growing by up to the header limit.
256	const auto limit = floored_subtract(in.max_size(), in.size());	77 ✔
257	const auto size = std::min(limit, std::max(in.capacity() - in.size(),	77 ✔
258	header_));	77 ✔
259
260	if (is_zero(size))	77 ✔
261	{
262	handler(error::buffer_overflow, total);	×
263	return;	×
264	}
265
266	async_read_some(in.prepare(size),	231 ✔
267	std::bind(&socket::handle_parse<Parser>,	154 ✔
268	shared_from_this(), _1, _2, buffer, parser, mode, total,
269	handler));
270	}
271
272	// private
273	template <class Parser>
274	void socket::handle_parse(const code& ec, size_t size,	83 ✔
275	const ref<http::flat_buffer>& buffer,
276	const std::shared_ptr<Parser>& parser, parse_mode mode, size_t total,
277	const count_handler& handler) NOEXCEPT
278	{
279	BC_ASSERT(stranded());	83 ✔
280
281	// The end of the stream completes a body so delimited, or is premature.
282	if (ec == error::peer_disconnect)	83 ✔
283	{
284	code eof{};	4 ✔
285	parser->put_eof(eof);	4 ✔
286	if (eof)	4 ✔
287	handler(error::http_to_error_code(eof), total);	1 ✔
288	else
289	handler(parser->is_done() ? error::success : ec, total);	3 ✔
290
291	return;
292	}
293
294	if (ec)	79 ✔
295	{
296	handler(ec, total);	8 ✔
297	return;	8 ✔
298	}
299
300	if (!websocket())	71 ✔
301	buffer.get().commit(size);	66 ✔
302
303	do_parse(buffer, parser, mode, total, handler);	71 ✔
304	}
305
306	// Codec (serialize).
307	// ----------------------------------------------------------------------------
308	// Each set of buffers is written in turn. A websocket writes each buffer as a
309	// frame of one message, which the last buffer closes.
310
311	// private
312	template <class Serializer>
313	void socket::do_serialize(	964 ✔
314	const std::shared_ptr<serialize_state<Serializer>>& out, size_t total,
315	const count_handler& handler) NOEXCEPT
316	{
317	BC_ASSERT(stranded());	964 ✔
318
319	while (true)	×
320	{
321	code ec{};	964 ✔
322	out->buffers = out->serializer.next(ec);	964 ✔
323	if (ec)	964 ✔
324	{
325	const auto code = error::http_to_error_code(ec);	×
326	if (code == error::unknown) logx("serialize", ec);	×
327	handler(code, total);	×
328	return;
329	}
330
331	if (!out->buffers.empty())	964 ✔
332	break;
333
334	if (out->serializer.is_done())	404 ✔
335	{
336	handler(error::success, total);	404 ✔
337	return;	404 ✔
338	}
339	}
340
341	if (websocket())	560 ✔
342	{
343	BC_ASSERT(is_one(out->buffers.size()));	129 ✔
344	ws_send_data(out->buffers.front(), out->serializer.binary(),	387 ✔
345	out->serializer.is_done(),	129 ✔
346	std::bind(&socket::handle_serialize<Serializer>,	258 ✔
347	shared_from_this(), _1, _2, out, total, handler));
348	return;	129 ✔
349	}
350
351	VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),	862 ✔
352	out->buffers, std::bind(&socket::handle_serialize<Serializer>,
353	shared_from_this(), _1, _2, out, total, handler));
354	}
355
356	// private
357	template <class Serializer>
358	void socket::handle_serialize(const code& ec, size_t size,	560 ✔
359	const std::shared_ptr<serialize_state<Serializer>>& out, size_t total,
360	const count_handler& handler) NOEXCEPT
361	{
362	BC_ASSERT(stranded());	560 ✔
363
364	total = ceilinged_add(total, size);	560 ✔
365	if (ec == error::operation_canceled)	560 ✔
366	{
367	handler(error::channel_stopped, total);	×
368	return;	×
369	}
370
371	if (ec)	560 ✔
372	{
373	if (ec == error::unknown) logx("write", ec);	5 ✔
374	handler(ec, total);	5 ✔
375	return;	5 ✔
376	}
377
378	do_serialize(out, total, handler);	555 ✔
379	}
380
381	template void socket::do_parse<socket::http_parser>(
382	const ref<http::flat_buffer>&, const http_parser_ptr&, parse_mode, size_t,
383	const count_handler&) NOEXCEPT;
384	template void socket::do_parse<socket::body_parser>(
385	const ref<http::flat_buffer>&, const body_parser_ptr&, parse_mode, size_t,
386	const count_handler&) NOEXCEPT;
387	template void socket::do_serialize<socket::http_serializer>(
388	const std::shared_ptr<serialize_state<http_serializer>>&, size_t,
389	const count_handler&) NOEXCEPT;
390	template void socket::do_serialize<socket::body_serializer>(
391	const std::shared_ptr<serialize_state<body_serializer>>&, size_t,
392	const count_handler&) NOEXCEPT;
393	template void socket::do_serialize<socket::notify_serializer>(
394	const std::shared_ptr<serialize_state<notify_serializer>>&, size_t,
395	const count_handler&) NOEXCEPT;
396
397	// HTTP (progressive read).
398	// ----------------------------------------------------------------------------
399	// The caller owns the parser, which owns http framing (content length and
400	// chunk decoding) across the message. A batched json-rpc body delivers each
401	// message from the paused parse (need_buffer), mapped to success here.
402
403	void socket::http_read_header(http::flat_buffer& buffer,	38 ✔
404	const http_parser_ptr& parser, count_handler&& handler) NOEXCEPT
405	{
406	io::dispatch(strand_,	38 ✔
407	std::bind(&socket::do_http_read_header, shared_from_this(),	76 ✔
408	std::ref(buffer), parser, std::move(handler)));	38 ✔
409	}	38 ✔
410
411	// private
412	void socket::do_http_read_header(ref<http::flat_buffer> buffer,	38 ✔
413	const http_parser_ptr& parser, const count_handler& handler) NOEXCEPT
414	{
415	BC_ASSERT(stranded());	38 ✔
416
417	// Websocket messages have no http headers.
418	if (websocket())	38 ✔
419	{
420	handler(error::operation_failed, {});	1 ✔
421	return;	1 ✔
422	}
423
424	parser->body_limit(maximum_);	37 ✔
425	parser->header_limit(header_);	37 ✔
426	do_parse(buffer, parser, parse_mode::header, zero,	37 ✔
427	std::bind(&socket::handle_http_read_header,	74 ✔
428	shared_from_this(), _1, _2, parser, handler));	74 ✔
429	}
430
431	// private
432	void socket::handle_http_read_header(const code& ec, size_t size,	37 ✔
433	const http_parser_ptr& parser, const count_handler& handler) NOEXCEPT
434	{
435	BC_ASSERT(stranded());	37 ✔
436
437	if (!ec && http::is_upgrade(parser->get()))	37 ✔
438	{
439	handler(error::upgrade, size);	14 ✔
440	return;	14 ✔
441	}
442
443	if (!ec && parser->get().method() == http::verb::unknown)	23 ✔
444	{
445	handler(error::bad_method, size);	2 ✔
446	return;	2 ✔
447	}
448
449	handler(ec, size);	21 ✔
450	}
451
452	void socket::http_read_some(http::flat_buffer& buffer,	6 ✔
453	const http_parser_ptr& parser, count_handler&& handler) NOEXCEPT
454	{
455	io::dispatch(strand_,	6 ✔
456	std::bind(&socket::do_http_read_body, shared_from_this(),	12 ✔
457	std::ref(buffer), parser, parse_mode::some,	12 ✔
458	std::move(handler)));	6 ✔
459	}	6 ✔
460
461	void socket::http_read_body(http::flat_buffer& buffer,	15 ✔
462	const http_parser_ptr& parser, count_handler&& handler) NOEXCEPT
463	{
464	io::dispatch(strand_,	15 ✔
465	std::bind(&socket::do_http_read_body, shared_from_this(),	30 ✔
466	std::ref(buffer), parser, parse_mode::all, std::move(handler)));	15 ✔
467	}	15 ✔
468
469	// private
470	void socket::do_http_read_body(ref<http::flat_buffer> buffer,	21 ✔
471	const http_parser_ptr& parser, parse_mode mode,
472	const count_handler& handler) NOEXCEPT
473	{
474	BC_ASSERT(stranded());	21 ✔
475
476	// Websocket messages have no http framing.
477	if (websocket())	21 ✔
478	{
479	handler(error::operation_failed, {});	2 ✔
480	return;	2 ✔
481	}
482
483	do_parse(buffer, parser, mode, zero, handler);	19 ✔
484	}
485
486	// HTTP (progressive write).
487	// ----------------------------------------------------------------------------
488	// The header is written once (chunked, no content length), and each response
489	// part is then written as chunk data, the batch close part terminating the
490	// chunked body. This mirrors the progressive read (nothing is materialized).
491
492	void socket::http_write_header(http::response&& response,	4 ✔
493	count_handler&& handler) NOEXCEPT
494	{
495	const auto out = emplace_shared<header_state>(
496	http::serialize_header(response.base()));	4 ✔
497
498	io::dispatch(strand_,	4 ✔
499	std::bind(&socket::do_http_write_header, shared_from_this(),	8 ✔
500	out, std::move(handler)));	4 ✔
501	}	4 ✔
502
503	// private
504	void socket::do_http_write_header(const header_state::ptr& out,	4 ✔
505	const count_handler& handler) NOEXCEPT
506	{
507	BC_ASSERT(stranded());	4 ✔
508
509	// Websocket messages have no http headers.
510	if (websocket())	4 ✔
511	{
512	handler(error::operation_failed, {});	1 ✔
513	return;	1 ✔
514	}
515
516	const io::const_buffer data{ out->header.data(),	3 ✔
517	out->header.size() };	3 ✔
518
519	VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),	6 ✔
520	data, std::bind(&socket::handle_http_write_header,
521	shared_from_this(), _1, _2, out, handler));
522	}
523
524	// private
525	void socket::handle_http_write_header(const code& ec, size_t size,	3 ✔
526	const header_state::ptr&, const count_handler& handler) NOEXCEPT
527	{
528	BC_ASSERT(stranded());	3 ✔
529
530	if (ec == error::operation_canceled)	3 ✔
531	{
532	handler(error::channel_stopped, size);	×
533	return;	×
534	}
535
536	if (ec == error::unknown) logx("http-write-header", ec);	3 ✔
537	handler(ec, size);	3 ✔
538	}
539
540	void socket::rpc_write_chunk(rpc::response&& response,	7 ✔
541	count_handler&& handler) NOEXCEPT
542	{
543	http::response wrapper{};	7 ✔
544	wrapper.body() = std::move(response);	7 ✔
545	assign_buffer(wrapper.body());	7 ✔
546
547	code ec{};	7 ✔
548	const auto out = emplace_shared<chunk_state>(std::move(wrapper));	7 ✔
549	out->writer.init(ec);	7 ✔
550
551	io::dispatch(strand_,	7 ✔
552	std::bind(&socket::do_chunk_write,	14 ✔
553	shared_from_this(), ec, zero, out, std::move(handler)));	14 ✔
554	}	7 ✔
555
556	// private
557	void socket::do_chunk_write(code ec, size_t total,	10 ✔
558	const chunk_state::ptr& out, const count_handler& handler) NOEXCEPT
559	{
560	BC_ASSERT(stranded());	10 ✔
561
562	// Websocket messages have no http framing.
563	if (websocket())	10 ✔
564	{
565	handler(error::operation_failed, total);	1 ✔
566	return;	1 ✔
567	}
568
569	const auto buffer = ec ? chunk_state::out_buffer{} : out->writer.get(ec);	9 ✔
570	if (ec)	9 ✔
571	{
572	const auto code = error::http_to_error_code(ec);	×
573	if (code == error::unknown) logx("chunk-write", ec);	×
574	handler(code, total);	×
575	return;
576	}
577
578	if (!buffer.has_value())	9 ✔
579	{
580	out->more = false;	×
581	handle_chunk_write({}, zero, total, out, handler);	×
582	return;	×
583	}
584
585	out->more = buffer.value().second;	9 ✔
586	const auto& data = buffer.value().first;	9 ✔
587
588	// Frame the part buffer as http chunk data (chunk_state lifetime).
589	out->prefix = http::chunk_header(data.size());	9 ✔
590	out->buffers =	9 ✔
591	{
592	io::const_buffer{ out->prefix.data(), out->prefix.size() },	9 ✔
593	data,
594	io::const_buffer{ http::chunk_end.data(), http::chunk_end.size() }
595	};	9 ✔
596
597	VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),	18 ✔
598	out->buffers, std::bind(&socket::handle_chunk_write,
599	shared_from_this(), _1, _2, total, out, handler));
600	}
601
602	// private
603	void socket::handle_chunk_write(const code& ec, size_t size,	9 ✔
604	size_t total, const chunk_state::ptr& out,
605	const count_handler& handler) NOEXCEPT
606	{
607	BC_ASSERT(stranded());	9 ✔
608
609	total = ceilinged_add(total, size);	9 ✔
610	if (ec == error::operation_canceled)	9 ✔
611	{
612	reclaim_buffer();	×
613	handler(error::channel_stopped, total);	×
614	return;	7 ✔
615	}
616
617	if (ec)	9 ✔
618	{
619	reclaim_buffer();	1 ✔
620	if (ec == error::unknown) logx("chunk-write", ec);	1 ✔
621	handler(ec, total);	1 ✔
622	return;	1 ✔
623	}
624
625	if (out->more)	8 ✔
626	{
627	do_chunk_write({}, total, out, handler);	3 ✔
628	return;	3 ✔
629	}
630
631	reclaim_buffer();	5 ✔
632
633	// Only the batch close part terminates the chunked body.
634	if (!out->last())	5 ✔
635	{
636	handler(error::success, total);	3 ✔
637	return;	3 ✔
638	}
639
640	const io::const_buffer last{ http::last_chunk.data(),	2 ✔
641	http::last_chunk.size() };
642
643	VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),	4 ✔
644	last, std::bind(&socket::handle_chunk_last,
645	shared_from_this(), _1, _2, total, handler));
646	}
647
648	// private
649	void socket::handle_chunk_last(const code& ec, size_t size,	2 ✔
650	size_t total, const count_handler& handler) NOEXCEPT
651	{
652	BC_ASSERT(stranded());	2 ✔
653
654	total = ceilinged_add(total, size);	2 ✔
655	if (ec == error::operation_canceled)	2 ✔
656	{
657	handler(error::channel_stopped, total);	×
658	return;	×
659	}
660
661	if (ec == error::unknown) logx("chunk-last", ec);	2 ✔
662	handler(ec, total);	2 ✔
663	}
664
665	BC_POP_WARNING()
666	BC_POP_WARNING()
667	BC_POP_WARNING()
668
669	} // namespace network
670	} // namespace libbitcoin

libbitcoin / libbitcoin-network / 41739478557

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous