Line data Source code
1 : #ifndef THINGER_WEBSOCKET_CONNECTION_HPP
2 : #define THINGER_WEBSOCKET_CONNECTION_HPP
3 :
4 : #include <chrono>
5 : #include <cstdint>
6 : #include <memory>
7 : #include <string>
8 : #include <boost/asio/streambuf.hpp>
9 : #include "../../asio/async_event.hpp"
10 : #include "../../asio/io_context_of.hpp"
11 : #include "../../asio/sockets/websocket.hpp"
12 : #include "../../asio/sockets/socket.hpp"
13 : #include "../data/out_data.hpp"
14 : #include "../../util/types.hpp"
15 : #include "upgrade.hpp"
16 : #include "flow_control.hpp"
17 : #include "close_notifier.hpp"
18 :
19 : namespace thinger::http{
20 :
21 : // The queue of the connection (detail/push_queue.hpp), only declared here: not public
22 : namespace detail {
23 : template<typename Item>
24 : class push_queue;
25 : enum class push_failure;
26 : }
27 :
28 : /// How a server WebSocket closed (see websocket_connection::on_close() and
29 : /// asio::websocket_close). The status code (RFC 6455, 7.4) is the one of the peer's close frame
30 : /// (1005 if it had none, 1002 if it was not valid); else the one this end closed with, as 1000
31 : /// (stop(), or the code given to stop(code, reason)), 1001 (server shutdown, or idle), 1007 or
32 : /// 1009 (a message not valid, or too large), 1011 (on_message() threw) or 1013 (it did not keep
33 : /// up, see write_limits); 1006 if no close frame was exchanged: the connection was lost, closed
34 : /// by the server at the end of a shutdown's grace period, or the close frame of stop() not
35 : /// written whole by its deadline. When this end closed first and the peer answered while the
36 : /// websocket was still reading (stop(), the shutdown), it is the code of the peer's answer, its
37 : /// echo, as Node's ws reports it; if the peer did not answer, the one sent. The reason is always
38 : /// the one of the peer's close frame, not the one given to stop()
39 : using websocket_close = asio::websocket_close;
40 :
41 : /**
42 : * A server WebSocket exchanging whole messages: received with on_message(), sent with
43 : * send_text()/send_binary() (queued, without waiting) or send_text_async()/send_binary_async()
44 : * (waiting until written, so a producer goes at the pace of the peer).
45 : *
46 : * Flow control: the queue holds 100 messages and drops the newest sent to a full queue, unless
47 : * set_write_limits() says otherwise (see write_limits in flow_control.hpp), and
48 : * dropped_messages() counts what the policy drops. pause_reading() lets TCP backpressure reach
49 : * the peer.
50 : *
51 : * Closing: stop() closes with 1000, stop(code, reason) with a code and a reason of the
52 : * application. Both do the closing handshake and end within asio::websocket::CLOSE_TIMEOUT,
53 : * whatever the peer does, and on_close() tells how it ended.
54 : *
55 : * Threads. Callbacks run on the websocket thread. From any thread: send_*(), wait_writable(),
56 : * set_write_limits(), on_high_watermark(), on_low_watermark(), pause_reading(),
57 : * resume_reading(), stop(), execute(), queued_bytes(), queued_messages() and dropped_messages().
58 : * On the websocket thread only (e.g. from its callbacks, or execute()): get_write_limits(), queue_full(),
59 : * is_reading_paused() and release_socket(). on_message() and on_shutdown() are set before
60 : * start(), in the handler that accepts it; on_close() too, or later on the websocket thread.
61 : */
62 : class websocket_connection : public std::enable_shared_from_this<websocket_connection>{
63 :
64 : public:
65 :
66 : /// Size of a read, and the default largest message received (a larger one closes with
67 : /// 1009, see websocket_accept_options::max_message_size)
68 : static constexpr size_t DEFAULT_BUFFER_SIZE = 1024; // 1KB
69 : static constexpr size_t MAX_BUFFER_SIZE = 16*1024*1024; // 16MB
70 :
71 : /// Default max_messages of the queue
72 : static constexpr int MAX_OUTPUT_MESSAGES = 100;
73 :
74 : /// On `socket`, with the largest message and the keepalive of `options` (its subprotocols
75 : /// and binary are those of an upgrade, not used here), and taken over from a server through
76 : /// `scope`, if any: told when the server shuts down (see on_shutdown()), counted as a
77 : /// websocket until closed
78 : websocket_connection(std::shared_ptr<asio::websocket> socket, const websocket_accept_options& options = {},
79 : server_scope scope = {});
80 :
81 : virtual ~websocket_connection();
82 :
83 : /// Each message received, whole, and whether it is binary
84 : void on_message(sync_callback<void(std::string, bool binary)> callback);
85 :
86 : /// What to do when the server shuts down gracefully (see server_base::shutdown()),
87 : /// e.g. send a last message and stop(). By default it closes with 1001 (Going Away);
88 : /// whatever is open at the end of the grace period is closed anyway
89 : void on_shutdown(sync_callback<void()> callback);
90 :
91 : /// Once the WebSocket closed, whatever closed it (the peer, an error, stop(), the server
92 : /// shutdown, the overflow policy...), and how: called once, after its last on_message();
93 : /// nothing can be sent anymore. Always posted to the websocket thread, never called from
94 : /// within the call closing it (e.g. a send from a loop over the connections, which may then
95 : /// remove this one); what it throws is logged. Set once the WebSocket closed, it is posted
96 : /// at once. Not called once released (release_socket()), nor if the io_context is stopped
97 : /// and destroyed with it open. See websocket_close for the code it gets; after stop(), it is
98 : /// called within asio::websocket::CLOSE_TIMEOUT of the call
99 : void on_close(sync_callback<void(const websocket_close&)> callback);
100 :
101 : /// The executor of the websocket (its thread, or strand): the one of its socket
102 672 : const boost::asio::any_io_executor& get_executor() const { return ws_->get_executor(); }
103 :
104 : /// Run `callback` on the websocket thread
105 2437190 : inline void execute(sync_callback<void()> callback){
106 : // Already there (e.g. from on_message()): at once, as dispatch() does, without its cost
107 : // through a type-erased executor on every message
108 2437190 : if(asio::running_in_this_thread(ws_->get_executor())){
109 36233 : callback();
110 36233 : return;
111 : }
112 2400957 : boost::asio::dispatch(ws_->get_executor(), [self = shared_from_this(), callback = std::move(callback)](){
113 2400957 : callback();
114 2400957 : });
115 : }
116 :
117 : /// Start reading, and the keepalive: the connection lives while it is open
118 : void start();
119 :
120 : /// Close with 1000 (Normal Closure) and no reason, as stop(code, reason). It never throws
121 : void stop();
122 :
123 : /**
124 : * Close with `code` and `reason`, as ws.close(code, reason) in Node: the closing handshake
125 : * (the frame goes after the message being written, ahead of those still queued, which are
126 : * not sent; then the peer answers), the queue closed and the producers waiting told false,
127 : * the keepalive stopped, and on_close() called once.
128 : *
129 : * It ends within asio::websocket::CLOSE_TIMEOUT (5 s) of the call, whatever the peer does
130 : * (it does not answer, it does not read, it reads a few bytes at a time): at the deadline the
131 : * socket is closed without the handshake, even with a message being written, which also
132 : * bounds a slow peer that keeps reading: to get a large message out first, await its send.
133 : * write_limits::write_timeout is not waited for, and does not turn it into a 1013. on_close()
134 : * gets the code of the peer's answer; `code` if the peer did not answer; 1006 if the close
135 : * frame was not written whole.
136 : *
137 : * Throws std::invalid_argument, in the calling thread and without closing, if a close frame
138 : * may not carry them: `code` is not 1000-1014 (but 1004, 1005 and 1006) nor 3000-4999, or
139 : * `reason` is above 123 bytes or not valid UTF-8, whatever the state of the connection. Once
140 : * closing or closed it does nothing else: the first code decides, the close of a server
141 : * shutdown included
142 : */
143 : void stop(uint16_t code, std::string reason = {});
144 :
145 : /// Whether the next message sent without waiting overflows the queue (not the watermarks)
146 : bool queue_full() const;
147 :
148 : /// Send without waiting: queued, or handled by the overflow policy if the queue is full
149 : void send_text(std::string text);
150 : void send_binary(std::string data);
151 :
152 : /// Send and wait until written to the socket: true, or false if the connection closed or
153 : /// failed first. Never dropped by the overflow policy; a cancelled wait leaves it queued
154 : awaitable<bool> send_text_async(std::string text);
155 : awaitable<bool> send_binary_async(std::string data);
156 :
157 : void set_write_limits(const write_limits& limits);
158 : write_limits get_write_limits() const;
159 :
160 : /// Payload bytes, and messages, not written yet (the one being written included)
161 : size_t queued_bytes() const;
162 : size_t queued_messages() const;
163 :
164 : /// Messages sent without waiting that the overflow policy dropped since the connection was
165 : /// created: the new one (drop_newest, or drop_oldest with no room to make), or the older ones
166 : /// (drop_oldest). Not those awaited, those sent once closing or closed, those queued when it
167 : /// closed, nor an overflow with overflow_policy::close. It never decreases, and a send from
168 : /// another thread counts once the websocket thread took it: once all were taken (e.g. after
169 : /// execute()), written + queued + dropped is what was sent while open
170 : uint64_t dropped_messages() const;
171 :
172 : /// The watermarks crossed (see write_limits); not told once the websocket is closing
173 : void on_high_watermark(sync_callback<void()> callback);
174 : void on_low_watermark(sync_callback<void()> callback);
175 :
176 : /// Wait while above the high watermark, until drained to the low one: true, or false if
177 : /// the connection closed
178 : awaitable<bool> wait_writable();
179 :
180 : /**
181 : * Stop reading the socket, so the peer's writes block once the TCP buffers fill; writes go
182 : * on. Messages already read are delivered: no other one if called on the websocket thread
183 : * (e.g. from on_message()), one more may be from another thread. While paused, the pings
184 : * and close frame of the peer are not read and the keepalive is suspended, so a dead peer
185 : * is not noticed: bound the pause, or set a write_timeout for a peer that stops reading
186 : * too. stop() still closes, discarding the messages before the peer's close frame
187 : */
188 : void pause_reading();
189 :
190 : /// Read again, in order and from where reading stopped; the keepalive is restarted. Nothing
191 : /// once closing: the closing handshake reads the socket
192 : void resume_reading();
193 :
194 : bool is_reading_paused() const;
195 :
196 : /// Release the socket for another handler: no other call may be made to this instance, and
197 : /// on_close() is not called. To close with a code of your own, stop(code, reason)
198 : std::shared_ptr<asio::socket> release_socket();
199 :
200 : private:
201 :
202 : void start_read_loop();
203 : awaitable<void> read_loop();
204 :
205 : /// The read loop, closing the websocket if on_message() throws
206 : awaitable<void> read_messages();
207 :
208 : /// How the websocket closed (see on_close())
209 : websocket_close close_status() const;
210 :
211 : /// Deliver the message read whole: false if it is not valid
212 : bool deliver_message();
213 :
214 : /// Wait while reading is paused: false if the websocket closed, or was released
215 : awaitable<bool> wait_resumed();
216 : void wake_reader();
217 :
218 : /// The keepalive is suspended while paused
219 : void suspend_keepalive();
220 : void restore_keepalive();
221 :
222 : /// Start the closing handshake with `code` and `reason`, unless closing already, and end the
223 : /// read loop once closed if paused
224 : void start_close(uint16_t code, std::string reason = {});
225 :
226 : /// A message to write
227 : struct out_message {
228 : std::string data;
229 : bool binary = false;
230 : };
231 :
232 : /// Write a message, closing the connection if the queue fails (see push_queue)
233 : awaitable<boost::system::error_code> write_message(out_message& message);
234 : void queue_failed(detail::push_failure failure, boost::system::error_code ec);
235 :
236 : /// Queue a message sent without waiting, or apply the overflow policy
237 : void push_message(std::string data, bool binary);
238 :
239 : /// Close a websocket that does not keep up: 1013 (Try Again Later)
240 : void close_congested(const char* reason);
241 :
242 : awaitable<bool> send_async(std::string data, bool binary);
243 :
244 : /// The server shuts down (on the websocket thread)
245 : void server_shutdown();
246 :
247 : private:
248 :
249 : std::shared_ptr<asio::websocket> ws_;
250 :
251 : /// Messages not written yet, against the write limits, and their writer
252 : std::unique_ptr<detail::push_queue<out_message>> queue_;
253 :
254 : /// Incoming data: the message being received, grown as its bytes arrive (never to the size
255 : /// a frame header announces), up to max_message_size; given up after a message above 64 KiB
256 : boost::asio::streambuf buffer_{MAX_BUFFER_SIZE};
257 :
258 : sync_callback<void(std::string, bool)> on_frame_callback_;
259 : sync_callback<void()> on_shutdown_callback_;
260 : close_notifier<websocket_close> on_close_;
261 :
262 : /// The server this connection was taken over from, if any
263 : server_scope scope_;
264 :
265 : /// Reading paused (see pause_reading()), and the event the read loop waits on meanwhile
266 : bool paused_ = false;
267 : asio::async_event resumed_;
268 :
269 : /// Started (start()), and released (release_socket())
270 : bool started_ = false;
271 : bool released_ = false;
272 :
273 : /// The connection was lost: a read failed while the websocket was open, with no close frame
274 : /// out
275 : bool lost_ = false;
276 :
277 : /// The closing handshake started (see start_close())
278 : bool closing_ = false;
279 :
280 : /// Keepalive started by start(), and its settings while suspended
281 : std::chrono::milliseconds ping_interval_;
282 : std::chrono::milliseconds idle_timeout_;
283 3102 : std::chrono::milliseconds keepalive_ping_{0};
284 3102 : std::chrono::milliseconds keepalive_idle_{0};
285 : };
286 :
287 : }
288 :
289 : #endif
|
