Line data Source code
1 : #include "response_sink.hpp"
2 : #include "request.hpp"
3 : #include "../common/http_data.hpp"
4 : #include "../data/out_chunk.hpp"
5 : #include "../data/out_string.hpp"
6 : #include "../../asio/async_event.hpp"
7 : #include "../../util/logger.hpp"
8 : #include <boost/asio/dispatch.hpp>
9 : #include <boost/asio/post.hpp>
10 :
11 : namespace thinger::http {
12 :
13 23512 : connection_sink::connection_sink(const request& req)
14 23512 : : connection_(req.source_ ? req.source_->connection() : std::weak_ptr<server_connection>{}),
15 47024 : stream_(req.http_stream_) {}
16 :
17 34383 : bool connection_sink::write(std::shared_ptr<http_frame> frame) {
18 34383 : auto connection = connection_.lock();
19 34383 : auto stream = stream_.lock();
20 34383 : if (!connection || !stream) {
21 7 : LOG_DEBUG("Connection lost while writing a response");
22 7 : return false;
23 : }
24 34376 : connection->handle_stream(stream, std::move(frame));
25 34376 : return true;
26 34383 : }
27 :
28 17799 : void connection_sink::send(std::shared_ptr<http_response> response) {
29 17799 : write(std::move(response));
30 17799 : }
31 :
32 163 : bool connection_sink::begin(std::shared_ptr<http_response> headers) {
33 652 : headers->add_header("Transfer-Encoding", "chunked");
34 652 : headers->set_header("X-Content-Type-Options", "nosniff");
35 163 : headers->set_last_frame(false); // the stream stays open for the chunks
36 163 : return write(std::move(headers));
37 : }
38 :
39 266 : bool connection_sink::append(const std::string& data) {
40 266 : auto chunk = std::make_shared<http_data>(std::make_shared<data::out_chunk>(data));
41 266 : chunk->set_last_frame(false);
42 532 : return write(std::move(chunk));
43 266 : }
44 :
45 160 : bool connection_sink::finish() {
46 : // Final zero-length chunk
47 160 : auto chunk = std::make_shared<http_data>(std::make_shared<data::out_chunk>());
48 160 : chunk->set_last_frame(true);
49 320 : return write(std::move(chunk));
50 160 : }
51 :
52 : // Streamed response: each frame is awaited until the connection writes it, so the body
53 : // goes at the pace of the client
54 713 : bool connection_sink::begin_stream(std::shared_ptr<http_response> head, body_framing framing) {
55 713 : framing_ = framing;
56 713 : if (framing == body_framing::chunked) {
57 612 : head->set_header(header::transfer_encoding, "chunked");
58 509 : } else if (framing == body_framing::close) {
59 36 : must_close(detail::close_reason::framing_by_close);
60 : }
61 713 : head->set_last_frame(false); // the stream stays open for the body and its end
62 713 : return write(std::move(head));
63 : }
64 :
65 14570 : awaitable<bool> connection_sink::write_stream(std::string data) {
66 : std::shared_ptr<data::out_data> body;
67 : if (framing_ == body_framing::chunked) {
68 : body = std::make_shared<data::out_chunk>(std::move(data));
69 : } else {
70 : auto raw = std::make_shared<data::out_string>();
71 : raw->get_string() = std::move(data);
72 : body = std::move(raw);
73 : }
74 : auto frame = std::make_shared<http_data>(std::move(body));
75 : frame->set_last_frame(false);
76 : co_return co_await write_and_wait(std::move(frame));
77 29140 : }
78 :
79 669 : awaitable<bool> connection_sink::end_stream() {
80 : // The last chunk of a chunked body; nothing for the others, whose end is known
81 : auto frame = framing_ == body_framing::chunked ? std::make_shared<http_data>(std::make_shared<data::out_chunk>())
82 : : std::make_shared<http_data>();
83 : frame->set_last_frame(true);
84 : co_return co_await write_and_wait(std::move(frame));
85 1338 : }
86 :
87 43 : void connection_sink::abort_stream() {
88 : // End the stream without ending the body (no last chunk, or fewer bytes than its
89 : // Content-Length) and close the connection: the client sees the body incomplete
90 43 : auto connection = connection_.lock();
91 43 : if (auto stream = stream_.lock(); connection && stream && stream->observer()) stream->observer()->abort();
92 43 : must_close(detail::close_reason::aborted);
93 43 : auto frame = std::make_shared<http_data>();
94 43 : frame->set_last_frame(true);
95 43 : write(std::move(frame));
96 43 : }
97 :
98 715 : void connection_sink::must_close(detail::close_reason reason) {
99 715 : auto connection = connection_.lock();
100 715 : auto stream = stream_.lock();
101 715 : if (!connection || !stream) return;
102 : // On the connection thread, which owns the stream, and before any frame queued after this
103 : // call (frames from other threads are posted too)
104 715 : boost::asio::dispatch(connection->get_socket()->get_io_context(), [stream, reason] {
105 715 : stream->must_close(reason);
106 715 : });
107 715 : }
108 :
109 15239 : awaitable<bool> connection_sink::write_and_wait(std::shared_ptr<http_frame> frame) {
110 : // The writer waits on an event that the connection sets once it wrote (or dropped) the
111 : // frame: the wait belongs to the io_context like any other operation, so it can be
112 : // cancelled (e.g. the frame waits behind another response that never ends) and is
113 : // released with the io_context
114 : struct waiter {
115 15239 : explicit waiter(const boost::asio::any_io_executor& executor) : executor(executor), done(executor) {}
116 : boost::asio::any_io_executor executor;
117 : asio::async_event done;
118 : bool written = false;
119 : };
120 : auto state = std::make_shared<waiter>(co_await boost::asio::this_coro::executor);
121 15239 : frame->on_written([weak = std::weak_ptr<waiter>(state)](bool written) {
122 15239 : auto waiting = weak.lock();
123 15239 : if (!waiting) return; // the writer is gone
124 : // on the executor of the writer, which owns the event (never inline within the
125 : // connection write loop)
126 15239 : boost::asio::post(waiting->executor, [weak, written] {
127 15218 : auto state = weak.lock();
128 15218 : if (!state) return;
129 15218 : state->written = written;
130 15218 : state->done.set();
131 15218 : });
132 15239 : });
133 : // A frame that cannot be queued is dropped: it reports false
134 : write(std::move(frame));
135 : if (!co_await state->done.wait()) co_return false; // the wait was cancelled
136 : co_return state->written;
137 30478 : }
138 :
139 411 : void connection_sink::observe(std::shared_ptr<response_observer> observer) {
140 : // The connection tells it, as it writes the frames of the stream (on the connection
141 : // thread, as this call). Not once the connection is gone: it no longer detaches the
142 : // stream, which the request may keep (see http_stream::detach())
143 411 : auto connection = connection_.lock();
144 411 : if (auto stream = stream_.lock(); connection && stream) stream->set_observer(std::move(observer));
145 411 : }
146 :
147 4317 : void connection_sink::take_over(std::shared_ptr<http_response> response, takeover_handler handler,
148 : body_framing framing) {
149 4317 : auto connection = connection_.lock();
150 4317 : auto stream = stream_.lock();
151 4317 : if (!connection || !stream) {
152 2 : LOG_WARNING("connection closed before it could be taken over: the takeover handler is not called");
153 2 : return;
154 : }
155 :
156 : // Its head tells the client that the body ends when the connection closes; the
157 : // connection is still handed over
158 4315 : if (framing == body_framing::close) must_close(detail::close_reason::framing_by_close);
159 :
160 : // The connection stops reading requests after this one, and hands the socket over
161 : // once this response has been written
162 4315 : connection->take_over(stream, std::move(response), std::move(handler));
163 4319 : }
164 :
165 : } // namespace thinger::http
|
