42421988994

Committed 07 Nov 2026 10:44PM UTC coverage: 93.582% (-0.3%) from 93.863%

Build # 42421988994

Build Type

Pull #1011

github

Committed by

web-flow

Commit Message

Merge 62b70bb20 into ddc7eabb3

Pull Request Pull Request #1011: Read rpc channel post bodies as json-rpc regardless of content type.

Coverage Stats

70 of 70 new or added lines in 5 files covered. (100.0%)

58 existing lines in 7 files now uncovered.

14567 of 15566 relevant lines covered (93.58%)

2260.09 hits per line

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

83.33

/src/net/socket.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/async/async.hpp>
#include <bitcoin/network/config/config.hpp>
#include <bitcoin/network/define.hpp>
#include <bitcoin/network/log/log.hpp>
#include <bitcoin/network/net/deadline.hpp>

// Async operations are not thread safe for the same socket object.
// Calls are stranded to protect the socket member.

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)

// Bounds query overrun following caller drop, not response latency.
constexpr auto watch_interval = milliseconds(10);

// Construct.
// ----------------------------------------------------------------------------

socket::socket(const logger& log, io::context& service,
    const parameters& parameters, bool proxied) NOEXCEPT
  : socket(log, service, parameters, {}, {}, proxied, true)
{
}

socket::socket(const logger& log, io::context& service,
    const parameters& params, const config::address& address,
    const config::endpoint& endpoint, bool proxied) NOEXCEPT
  : socket(log, service, params, address, endpoint, proxied, false)
{
}

// protected
socket::socket(const logger& log, io::context& service,
    const parameters& params, const config::address& address,
    const config::endpoint& endpoint, bool proxied, bool inbound) NOEXCEPT
  : reporter(log),
    tracker<socket>(log),
    inbound_(inbound),
    proxied_(proxied),
    maximum_(params.maximum_request),
    header_(std::min(params.maximum_request, maximum_http_header)),
    minimum_buffer_(params.minimum_buffer),
    bind_(params.bind),
    slot_(params.slot),
    strand_(service.get_executor()),
    service_(service),
    context_(params.context),
    role_(params.role),
    address_(address),
    endpoint_(endpoint),
    timer_(emplace_shared<deadline>(log, strand_, params.connect_timeout)),
    watch_(emplace_shared<deadline>(log, strand_, watch_interval)),
    socket_(std::in_place_type<io::socket>, strand_),
    response_(params.maximum_buffer)
{
}

socket::~socket() NOEXCEPT
{
    BC_ASSERT_MSG(stopped(), "socket is not stopped");
    if (!stopped_.load()) { LOGF("~socket is not stopped."); }
}

// Wait.
// ----------------------------------------------------------------------------

void socket::watch(result_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_watch,
            shared_from_this(), std::move(handler)));
}

void socket::poll(result_handler&& handler) NOEXCEPT
{
    io::dispatch(strand_,
        std::bind(&socket::do_poll,
            shared_from_this(), std::move(handler)));
}

// private
void socket::do_watch(const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (stopped_.load())
    {
        handler(error::success);
        return;
    }

    watch_->start(std::bind(&socket::handle_watch,
        shared_from_this(), _1, handler));
}

// private
void socket::do_poll(const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (stopped_.load())
    {
        handler(error::success);
        return;
    }

    watch_->start(std::bind(&socket::handle_poll,
        shared_from_this(), _1, handler));
}

// private
void socket::handle_watch(const code& ec,
    const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Only timer stop results in caller not stopping the channel.
    if (ec)
    {
        handler(error::success);
        return;
    }

    if (get_base().half_closed())
    {
        handler(error::peer_disconnect);
        return;
    }

    do_watch(handler);
}

// private
void socket::handle_poll(const code& ec,
    const result_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    // Timer stop (socket stop) results in caller not stopping the channel.
    if (!ec && get_base().half_closed())
    {
        handler(error::peer_disconnect);
        return;
    }

    handler(error::success);
}

// Properties.
// ----------------------------------------------------------------------------

io::context& socket::service() const NOEXCEPT
{
    return service_;
}

io::strand& socket::strand() NOEXCEPT
{
    return strand_;
}

bool socket::stranded() const NOEXCEPT
{
    return strand_.running_in_this_thread();
}

bool socket::stopped() const NOEXCEPT
{
    return stopped_.load();
}

bool socket::inbound() const NOEXCEPT
{
    return inbound_;
}

bool socket::websocket() const NOEXCEPT
{
    return websocket_.load();
}

bool socket::downgraded() const NOEXCEPT
{
    return downgraded_.load();
}

bool socket::detected() const NOEXCEPT
{
    return detected_.load();
}

const config::address& socket::address() const NOEXCEPT
{
    return address_;
}

const config::endpoint& socket::endpoint() const NOEXCEPT
{
    return endpoint_;
}

const config::endpoint& socket::binding() const NOEXCEPT
{
    return binding_;
}

size_t socket::slot() const NOEXCEPT
{
    return slot_;
}

void socket::set_address(const config::address& address) NOEXCEPT
{
    BC_ASSERT(stranded());
    address_ = address;
    endpoint_ = address;
}

// Variant state (protected by strand).
// ----------------------------------------------------------------------------

bool socket::secure() const NOEXCEPT
{
    BC_ASSERT(stranded());
    return std::holds_alternative<tls::stream>(socket_);
}

bool socket::encrypted() const NOEXCEPT
{
    BC_ASSERT(stranded());
    return std::holds_alternative<p2ps::stream>(socket_);
}

bool socket::zeromq() const NOEXCEPT
{
    BC_ASSERT(stranded());
    return std::holds_alternative<zmtp::stream>(socket_);
}

// protected
bool socket::is_base() const NOEXCEPT
{
    BC_ASSERT(stranded());
    return std::holds_alternative<io::socket>(socket_);
}

// Variant accessors.
// ----------------------------------------------------------------------------
// protected

socket::tcp_t socket::get_tcp() NOEXCEPT
{
    BC_ASSERT(stranded());

    return std::visit(overload
    {
        [](io::socket& value) NOEXCEPT -> socket::tcp_t
        {
            return std::ref(value);
        },
        [](tls::stream& value) NOEXCEPT -> socket::tcp_t
        {
            return std::ref(value);
        },
        [](p2ps::stream&) NOEXCEPT -> socket::tcp_t
        {
            std::terminate();
        },
        [](zmtp::stream&) NOEXCEPT -> socket::tcp_t
        {
            std::terminate();
        }
    }, socket_);
}

io::socket& socket::get_base() NOEXCEPT
{
    return std::visit(overload
    {
        [](io::socket& value) NOEXCEPT -> io::socket&
        {
            return value;
        },
        [](tls::stream& value) NOEXCEPT -> io::socket&
        {
            return value.next_layer();
        },
        [](p2ps::stream& value) NOEXCEPT -> io::socket&
        {
            return value.next_layer();
        },
        [](zmtp::stream& value) NOEXCEPT -> io::socket&
        {
            return value.next_layer();
        }
    }, socket_);
}

tls::stream& socket::get_tls() NOEXCEPT
{
    BC_ASSERT(stranded());
    BC_ASSERT(secure());

    return std::visit(overload
    {
        [](io::socket&) NOEXCEPT -> tls::stream&
        {
            std::terminate();
        },
        [](tls::stream& value) NOEXCEPT -> tls::stream&
        {
            return value;
        },
        [](p2ps::stream&) NOEXCEPT -> tls::stream&
        {
            std::terminate();
        },
        [](zmtp::stream&) NOEXCEPT -> tls::stream&
        {
            std::terminate();
        }
    }, socket_);
}

p2ps::stream& socket::get_p2ps() NOEXCEPT
{
    BC_ASSERT(stranded());
    BC_ASSERT(encrypted());

    return std::visit(overload
    {
        [](io::socket&) NOEXCEPT -> p2ps::stream&
        {
            std::terminate();
        },
        [](tls::stream&) NOEXCEPT -> p2ps::stream&
        {
            std::terminate();
        },
        [](p2ps::stream& value) NOEXCEPT -> p2ps::stream&
        {
            return value;
        },
        [](zmtp::stream&) NOEXCEPT -> p2ps::stream&
        {
            std::terminate();
        }
    }, socket_);
}

zmtp::stream& socket::get_zmtp() NOEXCEPT
{
    BC_ASSERT(stranded());
    BC_ASSERT(zeromq());

    return std::visit(overload
    {
        [](io::socket&) NOEXCEPT -> zmtp::stream&
        {
            std::terminate();
        },
        [](tls::stream&) NOEXCEPT -> zmtp::stream&
        {
            std::terminate();
        },
        [](p2ps::stream&) NOEXCEPT -> zmtp::stream&
        {
            std::terminate();
        },
        [](zmtp::stream& value) NOEXCEPT -> zmtp::stream&
        {
            return value;
        }
    }, socket_);
}

// Variant (ws vs. tcp)  helpers.
// ----------------------------------------------------------------------------
// protected

// write message frame (ws, finish closes the message) or unframed fixed
// size bytes (tcp). The binary/text setting applies only at the start of a
// message, so it is harmless to pass it on every frame of the message.
void socket::async_write(const io::const_buffer& buffer, bool binary,
    bool finish, const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (websocket())
    {
        ws_send_data(buffer, binary, finish, handler);
    }
    else
    {
        VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),
            buffer, std::bind(&socket::handle_async,
                shared_from_this(), _1, _2, handler, "async_write"));
    }
}

// read arbitrary tcp bytes, iterate (up to buffer capacity).
void socket::async_read_some(const io::mutable_buffer& buffer,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (websocket())
    {
        // Websockets read messages, use async_read(flat_buffer).
        handler(error::operation_failed, {});
    }
    else
    {
        VARIANT_DISPATCH_METHOD(get_tcp(),
            async_read_some(buffer, std::bind(&socket::handle_async,
                shared_from_this(), _1, _2, handler, "async_read_some")));
    }
}

// read arbitrary tcp bytes, iterate (up to buffer capacity).
// read whole ws message in any number of frames (up to buffer capacity).
void socket::async_read(http::flat_buffer& buffer,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    try
    {
        if (websocket())
        {
            buffer.consume(buffer.size());
            do_ws_read_message(std::ref(buffer), handler);
        }
        else
        {
            auto remain = floored_subtract(buffer.max_size(), buffer.size());
            if (is_zero(remain))
            {
                handler(error::buffer_overflow, {});
                return;
            }

            VARIANT_DISPATCH_METHOD(get_tcp(),
                async_read_some(buffer.prepare(remain),
                    std::bind(&socket::handle_async, shared_from_this(),
                    _1, _2, handler, "async_read_some")));
        }
    }
    catch (const std::exception& e)
    {
        LOGF("Exception @ async_read: " << e.what());
        handler(error::operation_failed, {});
    }
}

// read tcp fixed count of bytes (waits until mutable buffer is filled).
void socket::async_read(const io::mutable_buffer& buffer,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (websocket())
    {
        // Websockets read frame not size, use async_read(flat_buffer).
        handler(error::operation_failed, {});
    }
    else
    {
        // Fixed-size semantics.
        VARIANT_DISPATCH_FUNCTION(io::async_read, get_tcp(),
            buffer, std::bind(&socket::handle_async,
                shared_from_this(), _1, _2, handler, "async_read"));
    }
}

void socket::async_read_http(http::flat_buffer& buffer, http::request& request,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());

    if (websocket())
    {
        // The body reader processes a websocket read just like an http
        // read but without the headers. The expected body type (parser)
        // must be preselected on the request object.
        body_read(buffer, request, move_copy(handler));
    }
    else
    {
        const auto parser = to_shared<http_parser>();
        parser->body_limit(maximum_);
        parser->header_limit(header_);
        do_parse(std::ref(buffer), parser, parse_mode::all, zero,
            std::bind(&socket::handle_http_read,
                shared_from_this(), _1, _2, std::ref(request), parser,
                handler));
    }
}

void socket::async_write_http(http::response&& response,
    const count_handler& handler) NOEXCEPT
{
    BC_ASSERT(stranded());
    assign_buffer(response.body());

    if (websocket())
    {
        // The body writer processes a websocket write just like an http
        // write but without the headers. The expected body type
        // (serializer) is determined by caller-assigned body.
        body_write(std::move(response), move_copy(handler));
    }
    else
    {
        const auto out = emplace_shared<serialize_state<http_serializer>>(
            std::move(response));

        do_serialize(out, zero,
            std::bind(&socket::handle_http_write,
                shared_from_this(), _1, _2, handler));
    }
}

// Handlers.
// ----------------------------------------------------------------------------

// private
void socket::handle_async(const code& ec, size_t size,
    const count_handler& handler, const std::string& operation) NOEXCEPT
{
    BC_ASSERT(stranded());

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


    if (ec == error::unknown)
        logx(operation, ec);

    handler(ec, size);
}

// Logging.
// ----------------------------------------------------------------------------
// private

void socket::logx(const std::string& context,
    const code& ec) const NOEXCEPT
{
    LOGX("Socket " << context << " error (" << ec.value() << ") "
        << ec.category().name() << " : " << ec.message());
}

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/async/async.hpp>
22	#include <bitcoin/network/config/config.hpp>
23	#include <bitcoin/network/define.hpp>
24	#include <bitcoin/network/log/log.hpp>
25	#include <bitcoin/network/net/deadline.hpp>
26
27	// Async operations are not thread safe for the same socket object.
28	// Calls are stranded to protect the socket member.
29
30	namespace libbitcoin {
31	namespace network {
32
33	using namespace system;
34	using namespace std::placeholders;
35
36	// Shared pointers required in handler parameters so closures control lifetime.
37	BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR)
38	BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED)
39	BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
40
41	// Bounds query overrun following caller drop, not response latency.
42	constexpr auto watch_interval = milliseconds(10);
43
44	// Construct.
45	// ----------------------------------------------------------------------------
46
47	socket::socket(const logger& log, io::context& service,	367 ✔
48	const parameters& parameters, bool proxied) NOEXCEPT	367 ✔
49	: socket(log, service, parameters, {}, {}, proxied, true)	367 ✔
50	{
51	}	367 ✔
52
53	socket::socket(const logger& log, io::context& service,	135 ✔
54	const parameters& params, const config::address& address,
55	const config::endpoint& endpoint, bool proxied) NOEXCEPT	135 ✔
56	: socket(log, service, params, address, endpoint, proxied, false)	135 ✔
57	{
58	}	135 ✔
59
60	// protected
61	socket::socket(const logger& log, io::context& service,	502 ✔
62	const parameters& params, const config::address& address,
63	const config::endpoint& endpoint, bool proxied, bool inbound) NOEXCEPT	502 ✔
64	: reporter(log),
65	tracker<socket>(log),
66	inbound_(inbound),	502 ✔
67	proxied_(proxied),	502 ✔
68	maximum_(params.maximum_request),	502 ✔
69	header_(std::min(params.maximum_request, maximum_http_header)),	502 ✔
70	minimum_buffer_(params.minimum_buffer),	502 ✔
71	bind_(params.bind),	502 ✔
72	slot_(params.slot),	502 ✔
73	strand_(service.get_executor()),	502 ✔
74	service_(service),	502 ✔
75	context_(params.context),	502 ✔
76	role_(params.role),	502 ✔
77	address_(address),	502 ✔
78	endpoint_(endpoint),	502 ✔
79	timer_(emplace_shared<deadline>(log, strand_, params.connect_timeout)),	502 ✔
80	watch_(emplace_shared<deadline>(log, strand_, watch_interval)),	502 ✔
81	socket_(std::in_place_type<io::socket>, strand_),	502 ✔
82	response_(params.maximum_buffer)	2,008 ✔
83	{
84	}	502 ✔
85
86	socket::~socket() NOEXCEPT	502 ✔
87	{
88	BC_ASSERT_MSG(stopped(), "socket is not stopped");	502 ✔
89	if (!stopped_.load()) { LOGF("~socket is not stopped."); }	502 ✔
90	}	502 ✔
91
92	// Wait.
93	// ----------------------------------------------------------------------------
94
95	void socket::watch(result_handler&& handler) NOEXCEPT	18 ✔
96	{
97	io::dispatch(strand_,	18 ✔
98	std::bind(&socket::do_watch,	36 ✔
99	shared_from_this(), std::move(handler)));	36 ✔
100	}	18 ✔
101
102	void socket::poll(result_handler&& handler) NOEXCEPT	3 ✔
103	{
104	io::dispatch(strand_,	3 ✔
105	std::bind(&socket::do_poll,	6 ✔
106	shared_from_this(), std::move(handler)));	6 ✔
107	}	3 ✔
108
109	// private
110	void socket::do_watch(const result_handler& handler) NOEXCEPT	21 ✔
111	{
112	BC_ASSERT(stranded());	21 ✔
113
114	if (stopped_.load())	21 ✔
115	{
116	handler(error::success);	×
117	return;	×
118	}
119
120	watch_->start(std::bind(&socket::handle_watch,	21 ✔
121	shared_from_this(), _1, handler));	42 ✔
122	}
123
124	// private
125	void socket::do_poll(const result_handler& handler) NOEXCEPT	3 ✔
126	{
127	BC_ASSERT(stranded());	3 ✔
128
129	if (stopped_.load())	3 ✔
130	{
131	handler(error::success);	1 ✔
132	return;	1 ✔
133	}
134
135	watch_->start(std::bind(&socket::handle_poll,	2 ✔
136	shared_from_this(), _1, handler));	4 ✔
137	}
138
139	// private
140	void socket::handle_watch(const code& ec,	21 ✔
141	const result_handler& handler) NOEXCEPT
142	{
143	BC_ASSERT(stranded());	21 ✔
144
145	// Only timer stop results in caller not stopping the channel.
146	if (ec)	21 ✔
147	{
148	handler(error::success);	17 ✔
149	return;	17 ✔
150	}
151
152	if (get_base().half_closed())	4 ✔
153	{
154	handler(error::peer_disconnect);	1 ✔
155	return;	1 ✔
156	}
157
158	do_watch(handler);	3 ✔
159	}
160
161	// private
162	void socket::handle_poll(const code& ec,	2 ✔
163	const result_handler& handler) NOEXCEPT
164	{
165	BC_ASSERT(stranded());	2 ✔
166
167	// Timer stop (socket stop) results in caller not stopping the channel.
168	if (!ec && get_base().half_closed())	2 ✔
169	{
170	handler(error::peer_disconnect);	1 ✔
171	return;	1 ✔
172	}
173
174	handler(error::success);	1 ✔
175	}
176
177	// Properties.
178	// ----------------------------------------------------------------------------
179
180	io::context& socket::service() const NOEXCEPT	2 ✔
181	{
182	return service_;	2 ✔
183	}
184
185	io::strand& socket::strand() NOEXCEPT	2,835 ✔
186	{
187	return strand_;	2,835 ✔
188	}
189
190	bool socket::stranded() const NOEXCEPT	404,667 ✔
191	{
192	return strand_.running_in_this_thread();	404,667 ✔
193	}
194
195	bool socket::stopped() const NOEXCEPT	82,349 ✔
196	{
197	return stopped_.load();	82,349 ✔
198	}
199
200	bool socket::inbound() const NOEXCEPT	793 ✔
201	{
202	return inbound_;	793 ✔
203	}
204
205	bool socket::websocket() const NOEXCEPT	77,371 ✔
206	{
207	return websocket_.load();	77,371 ✔
208	}
209
210	bool socket::downgraded() const NOEXCEPT	95 ✔
211	{
212	return downgraded_.load();	95 ✔
213	}
214
215	bool socket::detected() const NOEXCEPT	43 ✔
216	{
217	return detected_.load();	43 ✔
218	}
219
220	const config::address& socket::address() const NOEXCEPT	314 ✔
221	{
222	return address_;	314 ✔
223	}
224
225	const config::endpoint& socket::endpoint() const NOEXCEPT	307 ✔
226	{
227	return endpoint_;	307 ✔
228	}
229
230	const config::endpoint& socket::binding() const NOEXCEPT	9 ✔
231	{
232	return binding_;	9 ✔
233	}
234
235	size_t socket::slot() const NOEXCEPT	4 ✔
236	{
237	return slot_;	4 ✔
238	}
239
240	void socket::set_address(const config::address& address) NOEXCEPT	2 ✔
241	{
242	BC_ASSERT(stranded());	2 ✔
243	address_ = address;	2 ✔
244	endpoint_ = address;	2 ✔
245	}	2 ✔
246
247	// Variant state (protected by strand).
248	// ----------------------------------------------------------------------------
249
250	bool socket::secure() const NOEXCEPT	16 ✔
251	{
252	BC_ASSERT(stranded());	16 ✔
253	return std::holds_alternative<tls::stream>(socket_);	16 ✔
254	}
255
256	bool socket::encrypted() const NOEXCEPT	1,031 ✔
257	{
258	BC_ASSERT(stranded());	1,031 ✔
259	return std::holds_alternative<p2ps::stream>(socket_);	1,031 ✔
260	}
261
262	bool socket::zeromq() const NOEXCEPT	219 ✔
263	{
264	BC_ASSERT(stranded());	219 ✔
265	return std::holds_alternative<zmtp::stream>(socket_);	219 ✔
266	}
267
268	// protected
269	bool socket::is_base() const NOEXCEPT	1 ✔
270	{
271	BC_ASSERT(stranded());	1 ✔
272	return std::holds_alternative<io::socket>(socket_);	1 ✔
273	}
274
275	// Variant accessors.
276	// ----------------------------------------------------------------------------
277	// protected
278
279	socket::tcp_t socket::get_tcp() NOEXCEPT	76,812 ✔
280	{
281	BC_ASSERT(stranded());	76,812 ✔
282
283	return std::visit(overload	153,624 ✔
284	{
285	[](io::socket& value) NOEXCEPT -> socket::tcp_t	76,803 ✔
286	{
287	return std::ref(value);	76,803 ✔
288	},
289	[](tls::stream& value) NOEXCEPT -> socket::tcp_t	9 ✔
290	{
291	return std::ref(value);	9 ✔
292	},
293	[](p2ps::stream&) NOEXCEPT -> socket::tcp_t	×
294	{
295	std::terminate();	×
296	},
297	[](zmtp::stream&) NOEXCEPT -> socket::tcp_t	×
298	{
299	std::terminate();	×
300	}
301	}, socket_);	76,812 ✔
302	}
303
304	io::socket& socket::get_base() NOEXCEPT	2,386 ✔
305	{
306	return std::visit(overload	4,772 ✔
307	{
308	[](io::socket& value) NOEXCEPT -> io::socket&	2,327 ✔
309	{
310	return value;	2,327 ✔
311	},
312	[](tls::stream& value) NOEXCEPT -> io::socket&	12 ✔
313	{
314	return value.next_layer();	12 ✔
315	},
316	[](p2ps::stream& value) NOEXCEPT -> io::socket&	4 ✔
317	{
318	return value.next_layer();	4 ✔
319	},
320	[](zmtp::stream& value) NOEXCEPT -> io::socket&	43 ✔
321	{
322	return value.next_layer();	43 ✔
323	}
324	}, socket_);	2,386 ✔
325	}
326
UNCOV 327	tls::stream& socket::get_tls() NOEXCEPT	×
328	{
UNCOV 329	BC_ASSERT(stranded());	×
UNCOV 330	BC_ASSERT(secure());	×
331
UNCOV 332	return std::visit(overload	×
333	{
334	[](io::socket&) NOEXCEPT -> tls::stream&	×
335	{
336	std::terminate();	×
337	},
UNCOV 338	[](tls::stream& value) NOEXCEPT -> tls::stream&	×
339	{
UNCOV 340	return value;	×
341	},
342	[](p2ps::stream&) NOEXCEPT -> tls::stream&	×
343	{
344	std::terminate();	×
345	},
346	[](zmtp::stream&) NOEXCEPT -> tls::stream&	×
347	{
348	std::terminate();	×
349	}
UNCOV 350	}, socket_);	×
351	}
352
353	p2ps::stream& socket::get_p2ps() NOEXCEPT	8 ✔
354	{
355	BC_ASSERT(stranded());	8 ✔
356	BC_ASSERT(encrypted());	8 ✔
357
358	return std::visit(overload	16 ✔
359	{
360	[](io::socket&) NOEXCEPT -> p2ps::stream&	×
361	{
362	std::terminate();	×
363	},
364	[](tls::stream&) NOEXCEPT -> p2ps::stream&	×
365	{
366	std::terminate();	×
367	},
368	[](p2ps::stream& value) NOEXCEPT -> p2ps::stream&	8 ✔
369	{
370	return value;	8 ✔
371	},
372	[](zmtp::stream&) NOEXCEPT -> p2ps::stream&	×
373	{
374	std::terminate();	×
375	}
376	}, socket_);	8 ✔
377	}
378
379	zmtp::stream& socket::get_zmtp() NOEXCEPT	94 ✔
380	{
381	BC_ASSERT(stranded());	94 ✔
382	BC_ASSERT(zeromq());	94 ✔
383
384	return std::visit(overload	188 ✔
385	{
386	[](io::socket&) NOEXCEPT -> zmtp::stream&	×
387	{
388	std::terminate();	×
389	},
390	[](tls::stream&) NOEXCEPT -> zmtp::stream&	×
391	{
392	std::terminate();	×
393	},
394	[](p2ps::stream&) NOEXCEPT -> zmtp::stream&	×
395	{
396	std::terminate();	×
397	},
398	[](zmtp::stream& value) NOEXCEPT -> zmtp::stream&	94 ✔
399	{
400	return value;	94 ✔
401	}
402	}, socket_);	94 ✔
403	}
404
405	// Variant (ws vs. tcp) helpers.
406	// ----------------------------------------------------------------------------
407	// protected
408
409	// write message frame (ws, finish closes the message) or unframed fixed
410	// size bytes (tcp). The binary/text setting applies only at the start of a
411	// message, so it is harmless to pass it on every frame of the message.
412	void socket::async_write(const io::const_buffer& buffer, bool binary,	118 ✔
413	bool finish, const count_handler& handler) NOEXCEPT
414	{
415	BC_ASSERT(stranded());	118 ✔
416
417	if (websocket())	118 ✔
418	{
419	ws_send_data(buffer, binary, finish, handler);	1 ✔
420	}
421	else
422	{
423	VARIANT_DISPATCH_FUNCTION(io::async_write, get_tcp(),	234 ✔
424	buffer, std::bind(&socket::handle_async,
425	shared_from_this(), _1, _2, handler, "async_write"));
426	}
427	}	118 ✔
428
429	// read arbitrary tcp bytes, iterate (up to buffer capacity).
430	void socket::async_read_some(const io::mutable_buffer& buffer,	81 ✔
431	const count_handler& handler) NOEXCEPT
432	{
433	BC_ASSERT(stranded());	81 ✔
434
435	if (websocket())	81 ✔
436	{
437	// Websockets read messages, use async_read(flat_buffer).
438	handler(error::operation_failed, {});	×
439	}
440	else
441	{
442	VARIANT_DISPATCH_METHOD(get_tcp(),	162 ✔
443	async_read_some(buffer, std::bind(&socket::handle_async,
444	shared_from_this(), _1, _2, handler, "async_read_some")));
445	}
446	}	81 ✔
447
448	// read arbitrary tcp bytes, iterate (up to buffer capacity).
449	// read whole ws message in any number of frames (up to buffer capacity).
450	void socket::async_read(http::flat_buffer& buffer,	25 ✔
451	const count_handler& handler) NOEXCEPT
452	{
453	BC_ASSERT(stranded());	25 ✔
454
455	try
456	{
457	if (websocket())	25 ✔
458	{
459	buffer.consume(buffer.size());	9 ✔
460	do_ws_read_message(std::ref(buffer), handler);	9 ✔
461	}
462	else
463	{
464	auto remain = floored_subtract(buffer.max_size(), buffer.size());	16 ✔
465	if (is_zero(remain))	16 ✔
466	{
467	handler(error::buffer_overflow, {});	1 ✔
468	return;	1 ✔
469	}
470
471	VARIANT_DISPATCH_METHOD(get_tcp(),	30 ✔
472	async_read_some(buffer.prepare(remain),
473	std::bind(&socket::handle_async, shared_from_this(),
474	_1, _2, handler, "async_read_some")));
475	}
476	}
477	catch (const std::exception& e)	×
478	{
479	LOGF("Exception @ async_read: " << e.what());	×
480	handler(error::operation_failed, {});	×
481	}	×
482	}
483
484	// read tcp fixed count of bytes (waits until mutable buffer is filled).
485	void socket::async_read(const io::mutable_buffer& buffer,	76,001 ✔
486	const count_handler& handler) NOEXCEPT
487	{
488	BC_ASSERT(stranded());	76,001 ✔
489
490	if (websocket())	76,001 ✔
491	{
492	// Websockets read frame not size, use async_read(flat_buffer).
493	handler(error::operation_failed, {});	1 ✔
494	}
495	else
496	{
497	// Fixed-size semantics.
498	VARIANT_DISPATCH_FUNCTION(io::async_read, get_tcp(),	152,000 ✔
499	buffer, std::bind(&socket::handle_async,
500	shared_from_this(), _1, _2, handler, "async_read"));
501	}
502	}	76,001 ✔
503
504	void socket::async_read_http(http::flat_buffer& buffer, http::request& request,	22 ✔
505	const count_handler& handler) NOEXCEPT
506	{
507	BC_ASSERT(stranded());	22 ✔
508
509	if (websocket())	22 ✔
510	{
511	// The body reader processes a websocket read just like an http
512	// read but without the headers. The expected body type (parser)
513	// must be preselected on the request object.
514	body_read(buffer, request, move_copy(handler));	4 ✔
515	}
516	else
517	{
518	const auto parser = to_shared<http_parser>();	18 ✔
519	parser->body_limit(maximum_);	18 ✔
520	parser->header_limit(header_);	18 ✔
521	do_parse(std::ref(buffer), parser, parse_mode::all, zero,	18 ✔
522	std::bind(&socket::handle_http_read,	36 ✔
523	shared_from_this(), _1, _2, std::ref(request), parser,	36 ✔
524	handler));
525	}	18 ✔
526	}	22 ✔
527
528	void socket::async_write_http(http::response&& response,	22 ✔
529	const count_handler& handler) NOEXCEPT
530	{
531	BC_ASSERT(stranded());	22 ✔
532	assign_buffer(response.body());	22 ✔
533
534	if (websocket())	22 ✔
535	{
536	// The body writer processes a websocket write just like an http
537	// write but without the headers. The expected body type
538	// (serializer) is determined by caller-assigned body.
539	body_write(std::move(response), move_copy(handler));	3 ✔
540	}
541	else
542	{
543	const auto out = emplace_shared<serialize_state<http_serializer>>(
544	std::move(response));	19 ✔
545
546	do_serialize(out, zero,	19 ✔
547	std::bind(&socket::handle_http_write,	38 ✔
548	shared_from_this(), _1, _2, handler));	38 ✔
549	}	19 ✔
550	}	22 ✔
551
552	// Handlers.
553	// ----------------------------------------------------------------------------
554
555	// private
556	void socket::handle_async(const code& ec, size_t size,	76,225 ✔
557	const count_handler& handler, const std::string& operation) NOEXCEPT
558	{
559	BC_ASSERT(stranded());	76,225 ✔
560
561	if (ec == error::operation_canceled)	76,225 ✔
562	{
563	handler(error::channel_stopped, size);	22 ✔
564	return;	22 ✔
565	}
566
567
568	if (ec == error::unknown)	76,203 ✔
569	logx(operation, ec);	×
570
571	handler(ec, size);	76,203 ✔
572	}
573
574	// Logging.
575	// ----------------------------------------------------------------------------
576	// private
577
578	void socket::logx(const std::string& context,	×
579	const code& ec) const NOEXCEPT
580	{
581	LOGX("Socket " << context << " error (" << ec.value() << ") "	×
582	<< ec.category().name() << " : " << ec.message());
583	}	×
584
585	BC_POP_WARNING()
586	BC_POP_WARNING()
587	BC_POP_WARNING()
588
589	} // namespace network
590	} // namespace libbitcoin

libbitcoin / libbitcoin-network / 42421988994

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