Line data Source code
1 : #ifndef SSE_CONNECTION_HPP
2 : #define SSE_CONNECTION_HPP
3 :
4 : #include <chrono>
5 : #include <cstdint>
6 : #include <memory>
7 : #include <optional>
8 : #include <string>
9 : #include <string_view>
10 : #include <boost/asio/any_io_executor.hpp>
11 : #include "../common/http_request.hpp"
12 : #include "../common/http_response.hpp"
13 : #include "../../asio/sockets/socket.hpp"
14 : #include "../../asio/watchdog.hpp"
15 : #include "../data/out_string.hpp"
16 : #include "../../util/types.hpp"
17 : #include "upgrade.hpp"
18 : #include "flow_control.hpp"
19 : #include "close_notifier.hpp"
20 :
21 : namespace thinger::http{
22 :
23 : // The queue of the connection (detail/push_queue.hpp), only declared here: not public
24 : namespace detail {
25 : template<typename Item>
26 : class push_queue;
27 : enum class push_failure;
28 : }
29 :
30 : /**
31 : * An event sent whole (see sse_connection::send()): its fields present, in this order, then
32 : * its data, each line a data field of its own, and the empty line that dispatches it. The same
33 : * on the wire as send_event(), send_id(), send_retry() and send_data(), and refused as they
34 : * are: an event type or ID with a CR or LF, an ID with NUL, or a negative retry
35 : */
36 : struct sse_event {
37 : std::optional<std::string> event;
38 : std::optional<std::string> id;
39 : std::optional<std::chrono::milliseconds> retry;
40 : std::string data;
41 : };
42 :
43 : /**
44 : * A Server-Sent Events stream. What is sent is queued, with the flow control of
45 : * websocket_connection (see write_limits), counted in what was sent: each field (send_event(),
46 : * send_id(), send_retry(), send_data()), each sse_event and each comment counts one.
47 : * drop_oldest drops whole events, never one being written nor one sent with send_async();
48 : * close closes, and EventSource reconnects. dropped_messages() counts what the policy drops.
49 : *
50 : * The connection reads what the client sends, only to learn when it leaves (it sends nothing
51 : * after its request; anything it sends is discarded): on_close() tells it.
52 : *
53 : * Threads. It runs on its executor (see get_executor()), which may be a strand. From any
54 : * thread: start(), send*(), stop(), execute(), set_keepalive(), wait_writable(),
55 : * set_write_limits(), on_high_watermark(), on_low_watermark(), queued_bytes(), queued_messages()
56 : * and dropped_messages(). On its executor only (e.g. from execute()): get_write_limits().
57 : * on_shutdown() and on_close() are set in the handler that accepts it.
58 : */
59 : class sse_connection : public std::enable_shared_from_this<sse_connection>, public boost::noncopyable{
60 :
61 : public:
62 :
63 : /// Default limit of the queue, in what is sent (max_messages is one more)
64 : static const int MAX_OUTPUT_MESSAGES = 100;
65 :
66 : /// Default interval of the keep-alive comments (see set_keepalive())
67 : static constexpr std::chrono::milliseconds DEFAULT_KEEPALIVE_INTERVAL{15000};
68 :
69 : /// On `executor` (e.g. a strand), or by default on the io_context of the socket; taken over
70 : /// from a server through `scope`, if any: told when the server shuts down (see
71 : /// on_shutdown()), counted as an event stream until closed
72 : sse_connection(std::shared_ptr<asio::socket> socket, boost::asio::any_io_executor executor = {},
73 : server_scope scope = {});
74 :
75 : virtual ~sse_connection();
76 :
77 : /// The executor of the connection: where it runs, and its callbacks
78 216 : const boost::asio::any_io_executor& get_executor() const { return executor_; }
79 :
80 : /// Run `callback` on the executor of the connection: at once if already there
81 : void execute(sync_callback<void()> callback);
82 :
83 : /// Start reading, to learn when the client leaves, and the keep-alive
84 : void start();
85 :
86 : void stop();
87 :
88 : /// What to do when the server shuts down gracefully (see server_base::shutdown()),
89 : /// e.g. send a last event and stop(). By default it closes (clients reconnect); whatever
90 : /// is open at the end of the grace period is closed anyway
91 27 : void on_shutdown(sync_callback<void()> callback){
92 27 : on_shutdown_callback_ = std::move(callback);
93 27 : }
94 :
95 : /// Once the connection closed, whatever closed it: the client left (or the connection
96 : /// failed), stop(), the server shutdown, the overflow policy or a write timeout (a quiet
97 : /// stream is not closed, see set_keepalive()). Called once; nothing can be sent anymore.
98 : /// Always posted to the executor, never called from within the call closing it (e.g. a
99 : /// send from a loop over the connections, which may then remove this one); what it
100 : /// throws is logged. Set once the connection already closed (e.g. by a coroutine handler
101 : /// that ran late), it is posted at once. Not called if the io_context is stopped and
102 : /// destroyed with it open
103 156 : void on_close(sync_callback<void()> callback){
104 156 : on_close_.set(std::move(callback));
105 156 : }
106 :
107 : /// Send an event whole, without waiting: queued, or handled by the overflow policy.
108 : /// Refused (false, an error logged) as its fields are (see sse_event)
109 : bool send(const sse_event& event);
110 :
111 : /// Send an event and wait until written: true, or false if refused, or if the connection
112 : /// closed or failed first. Never dropped by the overflow policy; a cancelled wait leaves
113 : /// it queued
114 : awaitable<bool> send_async(sse_event event);
115 :
116 : /// The reconnection time of the client, in milliseconds
117 : void send_retry(unsigned long millis);
118 :
119 : /// The type of the event whose data follows. Refused (false, an error logged) if it holds
120 : /// a CR or LF, which would end the field and inject others
121 : bool send_event(const std::string& event_name);
122 :
123 : /// The last event ID of the event whose data follows. Refused (false, an error logged) if
124 : /// it holds a CR, LF or NUL (a browser ignores an ID with NUL)
125 : bool send_id(const std::string& id);
126 :
127 : /// The data of an event, which ends it: each line (split on CRLF, CR or LF) is sent as a
128 : /// data field of its own, so the client gets the value as it was sent, line breaks as LF
129 : /// (WHATWG HTML, Server-sent events)
130 : void send_data(const std::string& data);
131 :
132 : /// A comment (": comment", a line the client ignores), sent with the event that follows:
133 : /// drop_oldest drops them together. Refused (false, an error logged) if it holds a CR or LF
134 : bool send_comment(std::string_view comment);
135 :
136 : /// Send a comment (": keep-alive", a line the client ignores) once `interval` passed
137 : /// without anything else written, so proxies and clients do not take a quiet stream as
138 : /// dead (DEFAULT_KEEPALIVE_INTERVAL, 15 seconds, by default); 0 disables it. Never queued
139 : /// nor counted. A client gone is noticed by reading the connection, whatever this is set to
140 : void set_keepalive(std::chrono::milliseconds interval);
141 :
142 : void set_write_limits(const write_limits& limits);
143 : write_limits get_write_limits() const;
144 :
145 : /// Bytes, and what was sent (see the class), not written yet (the one being written included)
146 : size_t queued_bytes() const;
147 : size_t queued_messages() const;
148 :
149 : /// What was sent without waiting (see the class: an event sent field by field counts each
150 : /// field) that the overflow policy dropped since the connection was created. Not what is
151 : /// awaited, sent once closed, or queued when it closed, nor an overflow with
152 : /// overflow_policy::close, nor the keep-alive comments, never queued. It never decreases, and
153 : /// a send from another thread counts once the executor took it: once all were taken (e.g.
154 : /// after execute()), written + queued + dropped is what was sent while open
155 : uint64_t dropped_messages() const;
156 :
157 : /// The watermarks crossed (see write_limits); not told once the connection is closed
158 : void on_high_watermark(sync_callback<void()> callback);
159 : void on_low_watermark(sync_callback<void()> callback);
160 :
161 : /// Wait while above the high watermark, until drained to the low one: true, or false if
162 : /// the connection closed
163 : awaitable<bool> wait_writable();
164 :
165 : private:
166 :
167 : /// What the queue writes: one field, an event, or a comment, as sent
168 : struct out_field {
169 : std::string data;
170 : /// An event, or its data field: it ends the event (drop_oldest drops whole events)
171 : bool ends_event = false;
172 : /// A keep-alive comment: written when nothing is queued, neither counted nor dropped
173 : bool keepalive = false;
174 : };
175 :
176 : /// Send a field: `data` as send_data(), any other type with a value without CR or LF.
177 : /// False if refused (a value that would break the framing), an error logged
178 : bool handle_write(std::string_view type, std::string_view value);
179 :
180 : /// Queue what was sent, on the executor, or apply the overflow policy
181 : void push(out_field field);
182 :
183 : /// Write a field, moving the keep-alive later (see push_queue)
184 : awaitable<boost::system::error_code> write_field(out_field& field);
185 :
186 : /// The queue failed (overflow, write timeout or error): closed
187 : void queue_failed(detail::push_failure failure, boost::system::error_code ec);
188 :
189 : /// The keep-alive expires `keepalive_` after the last write (nothing if 0)
190 : void arm_keepalive();
191 : /// It expired: a keep-alive comment if nothing is being written nor queued
192 : void keepalive_expired();
193 :
194 : /// Close the connection (on its executor), once
195 : void close();
196 :
197 : /// Read until the client leaves (`self` keeps the connection alive)
198 : awaitable<void> read_loop(std::shared_ptr<sse_connection> self);
199 :
200 : /// Taken over from a server: told when it shuts down
201 : void track_shutdown();
202 :
203 : std::shared_ptr<asio::socket> socket_;
204 :
205 : boost::asio::any_io_executor executor_;
206 :
207 : /// What was sent and is not written yet, against the write limits, and its writer
208 : std::unique_ptr<detail::push_queue<out_field>> queue_;
209 :
210 : /// Keep-alive comments: once keepalive_ passed without anything else written
211 : asio::watchdog keepalive_timer_;
212 : std::chrono::milliseconds keepalive_ = DEFAULT_KEEPALIVE_INTERVAL;
213 :
214 : sync_callback<void()> on_shutdown_callback_;
215 : close_notifier<> on_close_;
216 :
217 : bool closed_ = false;
218 :
219 : /// The server this connection was taken over from, if any
220 : server_scope scope_;
221 : };
222 :
223 : }
224 :
225 : #endif
|
