Line data Source code
1 : #include "server_connection.hpp"
2 : #include "request.hpp"
3 : #include "../../asio/sockets/socket.hpp"
4 : #include <boost/asio/co_spawn.hpp>
5 : #include <boost/asio/dispatch.hpp>
6 : #include <boost/asio/use_awaitable.hpp>
7 :
8 : namespace thinger::http {
9 :
10 : namespace detail {
11 :
12 : // Source of a request read from a server connection: what its client sends after the
13 : // request head, through the connection while it lives (a request may outlive it). Owned
14 : // by the request, as is the stream it refers to
15 : class connection_source final : public request_source {
16 : public:
17 23512 : connection_source(std::weak_ptr<server_connection> connection, http_stream& stream)
18 23512 : : connection_(std::move(connection)), stream_(stream) {}
19 :
20 43615 : awaitable<io_result> read_body_some(uint8_t* buffer, size_t size) override {
21 : auto connection = connection_.lock();
22 : if (!connection) co_return io_result{boost::asio::error::eof, 0};
23 : // On the thread of the connection, whose socket and deadline are not thread-safe: a
24 : // handler that moved to another executor reads there, and resumes on its own
25 : auto& io_context = connection->socket_->get_io_context();
26 : if (!io_context.get_executor().running_in_this_thread()) [[unlikely]] {
27 : co_return co_await boost::asio::co_spawn(io_context, read_body_some(buffer, size),
28 : boost::asio::use_awaitable);
29 : }
30 : // Nothing more is read once the connection closes gracefully: it drops the rest
31 : // itself (see server_connection::linger())
32 : if (connection->lingering()) co_return io_result{boost::asio::error::operation_aborted, 0};
33 : // The body is needed now: ask the client for it (Expect: 100-continue)
34 : if (stream_.get_continue() == http_stream::continue_state::pending) [[unlikely]] {
35 : connection->send_continue(stream_);
36 : }
37 : io_result result{boost::system::errc::make_error_code(read_deadline_expired), 0};
38 : if (!connection->read_deadline_passed()) [[likely]] {
39 : auto caller = (co_await boost::asio::this_coro::cancellation_state).slot();
40 : connection->body_read_started(caller);
41 : result = co_await connection->socket_->read_some_cancellable(buffer, size, connection->read_cancel_.slot());
42 : connection->body_read_ended(caller, result);
43 : }
44 : auto& [ec, bytes] = result;
45 : // A body that keeps arriving keeps the connection alive (e.g. a long upload)
46 : if (bytes > 0) connection->extend_timeout();
47 : // Read only while the body is not complete: its end, or an error, before it is the
48 : // client gone. Not a read cancelled (by the handler, or as the connection closes
49 : // gracefully), nor one of a connection closed here. Past its read deadline the
50 : // client is still there, but the body stays unread: the connection closes after
51 : // the response, which tells the client
52 : if (ec) [[unlikely]] {
53 : if (ec == read_deadline_expired) {
54 : stream_.must_close(close_reason::body_unread);
55 : } else if (ec != boost::asio::error::operation_aborted && !connection->lingering() &&
56 : connection->socket_->is_open()) {
57 : stream_.cancel(cancel_reason::client_gone);
58 : }
59 : }
60 : co_return result;
61 87230 : }
62 :
63 1386 : connection_info info() const override {
64 1386 : auto connection = connection_.lock();
65 1386 : if (!connection) return {};
66 1368 : const auto& socket = *connection->socket_;
67 6840 : return {socket.get_remote_ip(), socket.get_remote_port(), socket.get_local_port(), socket.is_secure()};
68 1386 : }
69 :
70 45 : std::optional<asio::tls_session> tls() const override {
71 45 : auto connection = connection_.lock();
72 90 : return connection ? connection->socket_->tls() : std::nullopt;
73 45 : }
74 :
75 3845 : bool misdirected(std::string_view host) const override {
76 3845 : auto connection = connection_.lock();
77 7690 : return connection && connection->socket_->misdirected(host);
78 3845 : }
79 :
80 225 : void hold_until(std::chrono::steady_clock::time_point time) override {
81 225 : if (auto connection = connection_.lock()) connection->hold_until(time);
82 225 : }
83 :
84 27 : void set_request_deadline(std::chrono::milliseconds deadline) override {
85 27 : if (auto connection = connection_.lock()) connection->set_request_deadline(stream_, deadline);
86 27 : }
87 :
88 515 : void set_read_deadline(std::chrono::steady_clock::time_point deadline) override {
89 515 : auto connection = connection_.lock();
90 515 : if (!connection) return;
91 : // On the thread of the connection, right here if called there. The stream is only
92 : // compared there, as it may be gone by then
93 515 : boost::asio::dispatch(connection->socket_->get_io_context(),
94 1030 : [connection, stream = &stream_, deadline] {
95 515 : connection->set_read_deadline(stream, deadline);
96 515 : });
97 515 : }
98 :
99 23512 : std::weak_ptr<server_connection> connection() const override { return connection_; }
100 :
101 1320 : boost::asio::io_context* io_context() const override {
102 1320 : auto connection = connection_.lock();
103 2640 : return connection ? &connection->socket_->get_io_context() : nullptr;
104 1320 : }
105 :
106 : private:
107 : std::weak_ptr<server_connection> connection_;
108 : http_stream& stream_;
109 : };
110 :
111 : }
112 :
113 23512 : std::unique_ptr<detail::request_source> server_connection::make_source(http_stream& stream) {
114 23512 : return std::make_unique<detail::connection_source>(weak_from_this(), stream);
115 : }
116 :
117 : }
|
