Line data Source code
1 : #ifndef THINGER_HTTP_SERVER_MEMORY_RESPONSE_HPP
2 : #define THINGER_HTTP_SERVER_MEMORY_RESPONSE_HPP
3 :
4 : #include "../common/http_response.hpp"
5 : #include "response_sink.hpp"
6 : #include "../../util/types.hpp"
7 : #include <boost/asio/experimental/concurrent_channel.hpp>
8 : #include <boost/asio/steady_timer.hpp>
9 : #include <algorithm>
10 : #include <chrono>
11 : #include <cstring>
12 : #include <memory>
13 : #include <mutex>
14 : #include <string>
15 :
16 : namespace thinger::http {
17 :
18 : // Collects the response of a request dispatched in memory (see http_application::dispatch).
19 : // It is shared by every copy of the response object, so a handler may keep a copy and
20 : // answer later, from any thread.
21 : class memory_response : public response_sink, public std::enable_shared_from_this<memory_response> {
22 : public:
23 4077 : explicit memory_response(const boost::asio::any_io_executor& executor) : done_(executor, 1) {}
24 :
25 : // Whole response
26 3945 : void send(std::shared_ptr<http_response> response) override {
27 : size_t bytes;
28 : {
29 3945 : std::lock_guard lock(mutex_);
30 3945 : if (complete_) return;
31 3032 : response_ = std::move(response);
32 3032 : if (observer_) observer_->head(response_->get_status(), false);
33 3032 : bytes = response_->get_content().size();
34 3032 : finish_locked();
35 3945 : }
36 3032 : observed_end(bytes);
37 : }
38 :
39 : // Chunked response, collected whole: headers first, then the data, then the end
40 15 : bool begin(std::shared_ptr<http_response> headers) override {
41 15 : std::lock_guard lock(mutex_);
42 15 : if (complete_ || response_) return false;
43 15 : response_ = std::move(headers);
44 15 : if (observer_) observer_->head(response_->get_status(), true);
45 15 : return true;
46 15 : }
47 :
48 24 : bool append(const std::string& data) override {
49 24 : std::lock_guard lock(mutex_);
50 24 : if (complete_ || !response_) return false;
51 18 : response_->get_content().append(data);
52 18 : return true;
53 24 : }
54 :
55 12 : bool finish() override {
56 : size_t bytes;
57 : {
58 12 : std::lock_guard lock(mutex_);
59 12 : if (complete_ || !response_) return false;
60 6 : bytes = response_->get_content().size();
61 6 : response_->set_content_length(bytes);
62 6 : response_->set_last_frame(true);
63 6 : finish_locked();
64 12 : }
65 6 : observed_end(bytes);
66 6 : return true;
67 : }
68 :
69 : // A streamed response that could not be completed: its body is incomplete, so the
70 : // result is a 502 instead
71 3 : void abort_stream() override {
72 3 : if (observer_) observer_->abort();
73 3 : send(http_response::stock_http_reply(http_response::status::bad_gateway));
74 3 : }
75 :
76 759 : void observe(std::shared_ptr<response_observer> observer) override {
77 : // Weak: the observer is released with this object, and may report from its destructor
78 759 : observer->on_reported([self = weak_from_this()] {
79 759 : if (auto response = self.lock()) response->settle_report();
80 759 : });
81 759 : std::lock_guard lock(mutex_);
82 759 : observer_ = std::move(observer);
83 759 : }
84 :
85 4206 : bool is_complete() const {
86 4206 : std::lock_guard lock(mutex_);
87 4206 : return complete_;
88 4206 : }
89 :
90 : // The application is gone before the response completed (a response in parts included):
91 : // there will be none. Wakes wait(), and whatever is sent later is dropped. The observer is
92 : // not told: it reports the response as aborted when released, once the request ends too
93 1752 : void abandon() {
94 1752 : std::lock_guard lock(mutex_);
95 : // Nothing waits for the response hook any longer (see wait_reported())
96 1752 : report_settled_ = true;
97 1752 : if (complete_) {
98 734 : done_.try_send(boost::system::error_code{});
99 734 : return;
100 : }
101 1018 : response_.reset();
102 1018 : finish_locked();
103 1752 : }
104 :
105 1159 : bool abandoned() const {
106 1159 : std::lock_guard lock(mutex_);
107 2318 : return complete_ && !response_;
108 1159 : }
109 :
110 : // Wait until the response is complete; null if it does not complete within `timeout`, or it
111 : // was abandoned. The timer wakes the same wait: the caller may run on an executor of several
112 : // threads, where the two branches of an `||` finishing at once would race on the cancellation
113 : // signal of each other
114 4206 : awaitable<std::shared_ptr<http_response>> wait(std::chrono::milliseconds timeout) {
115 : if (!is_complete()) {
116 : boost::asio::steady_timer timer(co_await boost::asio::this_coro::executor, timeout);
117 1820 : timer.async_wait([self = shared_from_this()](const boost::system::error_code& ec) {
118 1820 : if (!ec) self->done_.try_send(boost::system::error_code{});
119 1820 : });
120 : co_await done_.async_receive(use_nothrow_awaitable);
121 : }
122 : std::lock_guard lock(mutex_);
123 : co_return complete_ ? response_ : nullptr;
124 8412 : }
125 :
126 : // With a response hook, once the response is complete: wait until the hook has run, for at
127 : // most `timeout`, or the application is gone (see abandon()). False if the wait was cancelled
128 473 : awaitable<bool> wait_reported(std::chrono::milliseconds timeout) {
129 : if (report_settled()) co_return true;
130 : boost::asio::steady_timer timer(co_await boost::asio::this_coro::executor, timeout);
131 458 : timer.async_wait([self = shared_from_this()](const boost::system::error_code& ec) {
132 458 : if (!ec) self->done_.try_send(boost::system::error_code{});
133 458 : });
134 : auto cancellation = co_await boost::asio::this_coro::cancellation_state;
135 : // A wake-up may be left from the response: each one is checked again
136 : while (!report_settled() && timer.expiry() > boost::asio::steady_timer::clock_type::now()) {
137 : if (cancellation.cancelled() != boost::asio::cancellation_type::none) co_return false;
138 : auto [ec] = co_await done_.async_receive(use_nothrow_awaitable);
139 : if (ec) co_return false;
140 : }
141 : co_return true;
142 946 : }
143 :
144 : private:
145 473 : void settle_report() {
146 473 : std::lock_guard lock(mutex_);
147 473 : report_settled_ = true;
148 473 : done_.try_send(boost::system::error_code{});
149 473 : }
150 :
151 1603 : bool report_settled() const {
152 1603 : std::lock_guard lock(mutex_);
153 1603 : return report_settled_;
154 1603 : }
155 :
156 4056 : void finish_locked() {
157 4056 : complete_ = true;
158 4056 : done_.try_send(boost::system::error_code{});
159 4056 : }
160 :
161 : // Tell the observer the response is complete, with the size of its body (once, by the call
162 : // completing it), without the lock: it may call the response hook. Not reading the response,
163 : // which the dispatch may have handed to its caller by then
164 3038 : void observed_end(size_t bytes) {
165 3038 : if (!observer_) return;
166 473 : observer_->add_bytes(bytes);
167 473 : observer_->ended(response_outcome::completed);
168 : }
169 :
170 : mutable std::mutex mutex_;
171 : std::shared_ptr<http_response> response_;
172 : bool complete_ = false;
173 : // Told about the response, while a response hook is set (see http_application::on_response)
174 : std::shared_ptr<response_observer> observer_;
175 : // The response hook has run, or the application is gone and it is not waited for
176 : bool report_settled_ = false;
177 : boost::asio::experimental::concurrent_channel<void(boost::system::error_code)> done_;
178 : };
179 :
180 : namespace detail {
181 :
182 : // Source of a request dispatched in memory (see http_application::dispatch): its body, as if
183 : // just received from a client, then the end. Its peer is the address the dispatch gives, without
184 : // TLS, and no socket is ever read
185 : class memory_source : public request_source {
186 : public:
187 4107 : memory_source(std::string body, std::string peer_ip)
188 4107 : : body_(std::move(body)), peer_ip_(std::move(peer_ip)) {}
189 :
190 447 : awaitable<io_result> read_body_some(uint8_t* buffer, size_t size) override {
191 : size_t bytes = std::min(size, body_.size() - read_);
192 : if (bytes == 0) co_return io_result{boost::asio::error::eof, 0};
193 : std::memcpy(buffer, body_.data() + read_, bytes);
194 : read_ += bytes;
195 : co_return io_result{boost::system::error_code{}, bytes};
196 894 : }
197 :
198 1089 : connection_info info() const override { return {.peer_ip = peer_ip_}; }
199 :
200 : private:
201 : std::string body_;
202 : size_t read_ = 0;
203 : std::string peer_ip_;
204 : };
205 :
206 : } // namespace detail
207 :
208 : } // namespace thinger::http
209 :
210 : #endif // THINGER_HTTP_SERVER_MEMORY_RESPONSE_HPP
|
