Line data Source code
1 : #include "sse_connection.hpp"
2 : #include "connection_tracker.hpp"
3 : #include <cassert>
4 : #include <string_view>
5 : #include <vector>
6 : #include "../../asio/io_context_of.hpp"
7 : #include "../../util/logger.hpp"
8 : #include "detail/push_queue.hpp"
9 :
10 : namespace thinger::http {
11 :
12 : namespace {
13 : constexpr std::string_view keepalive_comment = ": keep-alive\n";
14 :
15 : /// Whether `value` holds a character that would end a field
16 12642 : bool breaks_field(std::string_view value, bool nul = false){
17 12642 : return value.find_first_of(nul ? std::string_view("\r\n\0", 3) : std::string_view("\r\n")) != std::string_view::npos;
18 : }
19 :
20 : /// Append the field `name: value`: false, nothing appended, if a line break in the value
21 : /// (or a NUL in an id) would end it and inject fields or events
22 6222 : bool append_field(std::string& out, std::string_view name, std::string_view value){
23 : // the names are the fields of the protocol: no line breaks, nor the colon of the value
24 6222 : assert(!name.empty() && !breaks_field(name) && name.find(':') == std::string_view::npos);
25 6222 : bool id = name == "id";
26 6222 : if(breaks_field(value, id)){
27 108 : LOG_ERROR("sse field refused: the value of '{}' holds a line break{}", name, id ? " or NUL" : "");
28 108 : return false;
29 : }
30 6114 : out.append(name).append(misc_strings::name_value_separator).append(value).append(misc_strings::lf);
31 6114 : return true;
32 : }
33 :
34 : /// Append the data of an event, which ends it: one data field per line (CRLF, CR or LF
35 : /// end a line, WHATWG HTML 9.2.6), then the empty line dispatching it
36 59777 : void append_data(std::string& out, std::string_view value){
37 59777 : out.reserve(out.size() + value.size() + 8);
38 : for(;;){
39 63260 : auto end = value.find_first_of("\r\n");
40 63260 : out.append("data").append(misc_strings::name_value_separator)
41 63260 : .append(value.substr(0, end)).append(misc_strings::lf);
42 63260 : if(end == std::string_view::npos) break;
43 3483 : auto next = end + 1;
44 3483 : if(value[end] == '\r' && next < value.size() && value[next] == '\n') ++next;
45 3483 : value.remove_prefix(next);
46 3483 : }
47 59777 : out.append(misc_strings::lf);
48 59777 : }
49 :
50 : /// An event as written, or nothing if one of its fields is refused
51 24336 : std::optional<std::string> format_event(const sse_event& event){
52 24336 : std::string out;
53 24336 : if(event.event && !append_field(out, "event", *event.event)) return std::nullopt;
54 24318 : if(event.id && !append_field(out, "id", *event.id)) return std::nullopt;
55 24291 : if(event.retry){
56 18 : if(event.retry->count() < 0){
57 9 : LOG_ERROR("sse field refused: a negative retry");
58 9 : return std::nullopt;
59 : }
60 9 : append_field(out, "retry", std::to_string(event.retry->count()));
61 : }
62 24282 : append_data(out, event.data);
63 24282 : return out;
64 24336 : }
65 :
66 18 : awaitable<bool> refused(){
67 : co_return false;
68 36 : }
69 : }
70 :
71 : // (sse_options() and not sse_options{}: the structure budget reads braces at namespace
72 : // scope as a body)
73 : static_assert(sse_options().keepalive == sse_connection::DEFAULT_KEEPALIVE_INTERVAL);
74 :
75 528 : sse_connection::sse_connection(std::shared_ptr<asio::socket> socket, boost::asio::any_io_executor executor,
76 528 : server_scope scope) :
77 528 : socket_(std::move(socket)),
78 528 : executor_(executor ? std::move(executor) : boost::asio::any_io_executor(socket_->get_io_context().get_executor())),
79 528 : queue_(std::make_unique<detail::push_queue<out_field>>(executor_,
80 0 : write_limits{.max_messages = MAX_OUTPUT_MESSAGES + 1},
81 1056 : detail::push_queue<out_field>::hooks{
82 28886 : .write_item = [this](out_field& field){ return write_field(field); },
83 140323 : .item_size = [](const out_field& field){ return field.data.size(); },
84 : // an event, or its data field: drop_oldest drops whole events
85 41747 : .ends_unit = [](const out_field& field){ return field.ends_event; },
86 598 : .on_failure = [this](detail::push_failure failure, boost::system::error_code ec){ queue_failed(failure, ec); },
87 19851 : .owner = [this]{ return std::shared_ptr<void>(shared_from_this()); },
88 900 : .send_backlog = [this]{ return socket_->send_backlog(); }})),
89 : // a member, called only while this lives: not kept alive while waiting
90 528 : keepalive_timer_(executor_, [this]{
91 458 : if(auto self = weak_from_this().lock()) self->keepalive_expired();
92 458 : }),
93 528 : on_close_(executor_),
94 1584 : scope_(std::move(scope))
95 : {
96 528 : LOG_DEBUG("created sse connection");
97 528 : }
98 :
99 528 : sse_connection::~sse_connection()
100 : {
101 528 : LOG_DEBUG("releasing sse connection");
102 528 : }
103 :
104 57374 : void sse_connection::execute(sync_callback<void()> callback){
105 57374 : if(asio::running_in_this_thread(executor_)){
106 40274 : callback();
107 40274 : return;
108 : }
109 17100 : boost::asio::dispatch(executor_, [self = shared_from_this(), callback = std::move(callback)](){
110 17100 : callback();
111 17100 : });
112 : }
113 :
114 528 : void sse_connection::start(){
115 528 : execute([this, self = shared_from_this()]{
116 528 : track_shutdown();
117 528 : arm_keepalive();
118 528 : co_spawn(executor_, read_loop(shared_from_this()), detached);
119 528 : });
120 528 : }
121 :
122 303 : void sse_connection::stop(){
123 303 : execute([this, self = shared_from_this()]{
124 303 : close();
125 303 : });
126 303 : }
127 :
128 546 : void sse_connection::close(){
129 546 : if(closed_) return;
130 528 : closed_ = true;
131 528 : keepalive_timer_.cancel();
132 528 : socket_->close();
133 528 : on_shutdown_callback_ = nullptr;
134 : // nothing else is written: producers waiting are told it closed, nobody that it drained
135 528 : queue_->close();
136 : // no longer counts for the server
137 528 : scope_.closed();
138 : // told last, once, posted (the callback is released here: it may own this connection)
139 528 : on_close_.notify();
140 : }
141 :
142 528 : awaitable<void> sse_connection::read_loop([[maybe_unused]] std::shared_ptr<sse_connection> self){
143 : // However the loop ends (also with the io_context destroyed), the queue closes (its
144 : // watermark callbacks released, producers waiting told) and the callbacks are released:
145 : // they may hold this connection
146 528 : auto guard = queue_->guard_cycles([this]{
147 528 : on_shutdown_callback_ = nullptr;
148 528 : on_close_.release();
149 528 : });
150 :
151 : uint8_t buffer[512];
152 : while(!closed_){
153 : auto [ec, bytes] = co_await socket_->read_some(buffer, sizeof(buffer));
154 : if(ec) break;
155 : }
156 : if(!closed_){
157 : LOG_DEBUG("sse client left");
158 : close();
159 : }
160 1056 : }
161 :
162 528 : void sse_connection::track_shutdown(){
163 528 : scope_.set_type(connection_type::sse);
164 : // called on the io_context of the socket: the callback and the close run on the executor
165 528 : scope_.on_shutdown([weak = weak_from_this()]{
166 36 : auto self = weak.lock();
167 36 : if(!self) return;
168 36 : self->execute([self]{
169 36 : if(self->on_shutdown_callback_) self->on_shutdown_callback_();
170 18 : else self->close();
171 36 : });
172 36 : });
173 528 : }
174 :
175 19506 : bool sse_connection::send(const sse_event& event){
176 19506 : auto data = format_event(event);
177 19506 : if(!data) return false;
178 19470 : push(out_field{.data = std::move(*data), .ends_event = true});
179 19470 : return true;
180 19506 : }
181 :
182 4830 : awaitable<bool> sse_connection::send_async(sse_event event){
183 4830 : auto data = format_event(event);
184 4830 : if(!data) return refused();
185 9624 : return queue_->push_and_wait(out_field{.data = std::move(*data), .ends_event = true});
186 4830 : }
187 :
188 27 : void sse_connection::send_retry(unsigned long millis){
189 27 : handle_write("retry", std::to_string(millis));
190 27 : }
191 :
192 510 : bool sse_connection::send_event(const std::string& event_name){
193 510 : return handle_write("event", event_name);
194 : }
195 :
196 249 : bool sse_connection::send_id(const std::string& id){
197 249 : return handle_write("id", id);
198 : }
199 :
200 35495 : void sse_connection::send_data(const std::string& data){
201 35495 : handle_write("data", data);
202 35495 : }
203 :
204 198 : bool sse_connection::send_comment(std::string_view comment){
205 198 : if(breaks_field(comment)){
206 18 : LOG_ERROR("sse comment refused: it holds a line break");
207 18 : return false;
208 : }
209 180 : out_field field;
210 180 : field.data.reserve(comment.size() + 3);
211 180 : field.data.append(": ").append(comment).append(misc_strings::lf);
212 180 : push(std::move(field));
213 180 : return true;
214 180 : }
215 :
216 36281 : bool sse_connection::handle_write(std::string_view type, std::string_view value){
217 36281 : out_field field;
218 : // a data field ends its event
219 36281 : field.ends_event = type == "data";
220 36281 : if(field.ends_event) append_data(field.data, value);
221 786 : else if(!append_field(field.data, type, value)) return false;
222 36218 : push(std::move(field));
223 36218 : return true;
224 36281 : }
225 :
226 55868 : void sse_connection::push(out_field field){
227 55868 : execute([this, field = std::move(field)]() mutable {
228 : // (once closed, nothing is sent: not a congestion)
229 55868 : if(queue_->push(std::move(field)) == detail::push_queue<out_field>::admission::drop && !queue_->closed()){
230 7850 : LOG_DEBUG("sse connection is congested. discarding what was sent");
231 : }
232 55868 : });
233 55868 : }
234 :
235 28358 : awaitable<boost::system::error_code> sse_connection::write_field(out_field& field){
236 : // past a slice, with a write deadline, in slices: each one is progress, so a slow client
237 : // reading a large event is told from one that stopped reading
238 : boost::system::error_code ec;
239 : if(field.data.size() > asio::socket::WRITE_SLICE_SIZE && queue_->get_write_limits().write_timeout.count() > 0){
240 : std::vector<boost::asio::const_buffer> buffers{boost::asio::buffer(field.data)};
241 14310 : ec = std::get<0>(co_await socket_->write_slices(buffers, [this]{ queue_->write_progressed(); }));
242 : }else{
243 : ec = std::get<0>(co_await socket_->write(std::string_view(field.data)));
244 : }
245 : // anything else written puts the keep-alive off
246 : if(!ec && !field.keepalive) arm_keepalive();
247 : co_return ec;
248 56716 : }
249 :
250 70 : void sse_connection::queue_failed(detail::push_failure failure, boost::system::error_code ec){
251 70 : switch(failure){
252 24 : case detail::push_failure::overflow:
253 24 : LOG_WARNING("sse connection is congested. closing it");
254 24 : break;
255 33 : case detail::push_failure::write_timeout:
256 33 : LOG_WARNING("closing sse connection: its write timed out");
257 33 : break;
258 13 : case detail::push_failure::write_error:
259 : // nothing else can be sent
260 13 : LOG_DEBUG("sse write failed: {}", ec.message());
261 13 : break;
262 : }
263 70 : close();
264 70 : }
265 :
266 615 : void sse_connection::set_keepalive(std::chrono::milliseconds interval){
267 615 : execute([this, interval]{
268 615 : keepalive_ = interval;
269 : // the next one counts from now
270 615 : keepalive_timer_.cancel();
271 615 : arm_keepalive();
272 615 : });
273 615 : }
274 :
275 29563 : void sse_connection::arm_keepalive(){
276 29563 : if(closed_ || keepalive_.count() <= 0) return;
277 : // a later deadline does not touch the timer (see watchdog)
278 25621 : keepalive_timer_.expires_at(std::chrono::steady_clock::now() + keepalive_);
279 : }
280 :
281 458 : void sse_connection::keepalive_expired(){
282 458 : if(closed_) return;
283 : // only when quiet: otherwise what is written moves it later
284 916 : queue_->write_idle(out_field{.data = std::string(keepalive_comment), .keepalive = true});
285 458 : arm_keepalive();
286 : }
287 :
288 8418 : awaitable<bool> sse_connection::wait_writable(){
289 8418 : return queue_->wait_writable();
290 : }
291 :
292 180 : void sse_connection::set_write_limits(const write_limits& limits){
293 180 : queue_->set_write_limits(limits);
294 180 : }
295 :
296 9 : write_limits sse_connection::get_write_limits() const{
297 9 : return queue_->get_write_limits();
298 : }
299 :
300 2619 : size_t sse_connection::queued_bytes() const{
301 2619 : return queue_->queued_bytes();
302 : }
303 :
304 11839 : size_t sse_connection::queued_messages() const{
305 11839 : return queue_->queued_messages();
306 : }
307 :
308 17905256 : uint64_t sse_connection::dropped_messages() const{
309 17905256 : return queue_->dropped_messages();
310 : }
311 :
312 12 : void sse_connection::on_high_watermark(sync_callback<void()> callback){
313 12 : queue_->on_high_watermark(std::move(callback));
314 12 : }
315 :
316 18 : void sse_connection::on_low_watermark(sync_callback<void()> callback){
317 18 : queue_->on_low_watermark(std::move(callback));
318 18 : }
319 :
320 : }
|
