Line data Source code
1 : #ifndef THINGER_HTTP_WEBSOCKET_CLIENT_HPP
2 : #define THINGER_HTTP_WEBSOCKET_CLIENT_HPP
3 :
4 : #include <memory>
5 : #include <string>
6 : #include <string_view>
7 : #include <optional>
8 : #include <functional>
9 : #include <atomic>
10 : #include <chrono>
11 :
12 : #include "../../asio/sockets/websocket.hpp"
13 : #include "../../util/types.hpp"
14 :
15 : namespace thinger::http {
16 :
17 : /**
18 : * Represents an established WebSocket connection from the client side.
19 : *
20 : * Obtained via client.websocket(), async_client.websocket() or executor_client.websocket().
21 : * It has a synchronous API, for http::client and threads that do not run the io_context of
22 : * the socket (see Synchronous API), a coroutine API, and a message loop with callbacks
23 : * (run()), for the callbacks and coroutines of async_client and executor_client.
24 : *
25 : * Usage:
26 : * http::client client;
27 : * if (auto ws = client.websocket("ws://server.com/path")) {
28 : * ws->send_text("Hello!");
29 : * auto [msg, binary] = ws->receive();
30 : * ws->close();
31 : * }
32 : *
33 : * async_client.websocket("ws://server.com/path", [](auto ws) -> awaitable<void> {
34 : * if (ws) {
35 : * co_await ws->send_text_async("Hello!");
36 : * auto [msg, binary] = co_await ws->receive_async();
37 : * co_await ws->close_async();
38 : * }
39 : * });
40 : *
41 : * Keepalive: on by default, as on the server (asio::websocket::start_timeout()): it pings the
42 : * peer once nothing was received from it for 60 s, and closes the connection once nothing was
43 : * received for 120 s, so a read or a write never waits for ever on a dead peer (lost behind a
44 : * NAT, a black hole). set_keepalive() changes both, (0, 0) turns it off. The idle time counts
45 : * time without receiving any byte, pongs included: a peer sending a byte within each idle
46 : * timeout is alive, even if no message completes. A connection closed for being idle (its
47 : * close frame has code 1001, going away) ends what is pending and fails what comes after:
48 : * receive() returns an empty message, send_text() and send_binary() return false, is_open() is
49 : * false and get_close_code() is 1001 (CLOSE_GOING_AWAY).
50 : *
51 : * The keepalive runs on the io_context of the socket: where something runs it (the workers of
52 : * an async_client, the io_context of an executor_client), it acts on time. With the
53 : * synchronous API of http::client nothing runs between two calls: the keepalive acts at the
54 : * next one, which ends as above if the connection was found idle (it never hangs). Bytes that arrived meanwhile (e.g. the pongs of a
55 : * client that only sends) count as received; and if the idle timeout passed with no ping
56 : * sent, it pings then instead of closing, and closes only if that ping is not answered within
57 : * the idle timeout minus the ping interval. A client that never reads is closed at the idle
58 : * timeout once its socket buffer is full of the frames it does not read (weeks of pongs at
59 : * the defaults, sooner if the peer sends it data): run() or receive() read them.
60 : */
61 : class websocket_client {
62 : public:
63 : // Largest message received (as the server side, websocket_connection)
64 : static constexpr size_t MAX_MESSAGE_SIZE = 16 * 1024 * 1024;
65 :
66 : using message_callback = sync_callback<void(const std::string&, bool)>;
67 : using close_callback = sync_callback<void()>;
68 : using error_callback = sync_callback<void(const std::string&)>;
69 :
70 : /**
71 : * Construct from an established WebSocket.
72 : * Not intended for direct use - use client.websocket() instead.
73 : */
74 : explicit websocket_client(std::shared_ptr<asio::websocket> ws);
75 :
76 : ~websocket_client();
77 :
78 : // Non-copyable
79 : websocket_client(const websocket_client&) = delete;
80 : websocket_client& operator=(const websocket_client&) = delete;
81 :
82 : // Movable
83 : websocket_client(websocket_client&& other) noexcept;
84 : websocket_client& operator=(websocket_client&& other) noexcept;
85 :
86 : // ============================================
87 : // Synchronous API
88 : //
89 : // For http::client, and for threads that do not run the io_context of the socket. On a
90 : // WebSocket of http::client, whose io_context only these calls run, a call waits running
91 : // it until its own operation completes (not until it runs out of work: the keepalive keeps
92 : // it busy); one thread at a time, as for http::client. On a thread running the io_context
93 : // (an async_client worker, so its callbacks; a coroutine of executor_client; a strand of
94 : // either) a call throws std::logic_error before starting anything: there it could only
95 : // wait for a handler it holds up, or run the handlers of other connections within its
96 : // caller. On any other thread it waits for the operation without running the io_context,
97 : // so something else must run it (the workers, the threads of the executor_client). In
98 : // callbacks and coroutines use the coroutine API below, run() with on_message(), or
99 : // stop() to close without waiting
100 : // ============================================
101 :
102 : /**
103 : * Send a text message.
104 : * @return true if sent successfully
105 : */
106 : bool send_text(std::string_view message);
107 :
108 : /**
109 : * Send a binary message.
110 : * @return true if sent successfully
111 : */
112 : bool send_binary(const void* data, size_t size);
113 : bool send_binary(std::string_view data);
114 :
115 : /**
116 : * Receive the next message.
117 : * Blocks until a message is received or connection closes.
118 : * @return {message, is_binary} pair, empty message on error/close (an empty message
119 : * received is valid too: is_open() tells them apart)
120 : */
121 : std::pair<std::string, bool> receive();
122 :
123 : /**
124 : * Close the connection gracefully.
125 : */
126 : void close();
127 :
128 : /**
129 : * Check if connection is open.
130 : */
131 : bool is_open() const;
132 :
133 : /**
134 : * Check if connection is valid (for use with optional).
135 : */
136 : explicit operator bool() const { return is_open(); }
137 :
138 : /**
139 : * Subprotocol accepted by the server (Sec-WebSocket-Protocol), empty if none.
140 : */
141 9 : std::string get_protocol() const { return websocket_ ? websocket_->get_protocol() : std::string{}; }
142 :
143 : /**
144 : * Status code of the closing handshake (asio::websocket::get_close_code()): the one of the
145 : * peer's close frame, else the one this end closed with (1001 for the keepalive); 1006 if
146 : * none was exchanged or once released.
147 : */
148 27 : uint16_t get_close_code() const {
149 27 : return websocket_ ? websocket_->get_close_code() : asio::websocket::CLOSE_ABNORMAL;
150 : }
151 :
152 : /**
153 : * Keepalive (see above): ping the peer once nothing was received from it for
154 : * `ping_interval`, close the connection once nothing was received for `idle_timeout`; 0
155 : * disables each one, (0, 0) the keepalive. Applies from now on, the idle time counted from
156 : * now. On the executor of the WebSocket it applies at once; anywhere else it is a
157 : * synchronous call (see Synchronous API above)
158 : */
159 : void set_keepalive(std::chrono::milliseconds ping_interval, std::chrono::milliseconds idle_timeout);
160 : // The keepalive in force: 60 s and 120 s unless changed; 0 once released
161 : std::chrono::milliseconds get_ping_interval() const;
162 : std::chrono::milliseconds get_idle_timeout() const;
163 :
164 : /**
165 : * Release ownership of the underlying WebSocket.
166 : * It is a byte stream socket (asio::socket), usable by any protocol or with a
167 : * socket_pipe; it writes text messages unless set_binary(true) is called.
168 : * After calling this method, the websocket_client becomes invalid.
169 : * Use this to take full control of the raw WebSocket for advanced use cases.
170 : * The keepalive in force (the default or the one set) stays on the WebSocket released:
171 : * its new owner changes it with asio::websocket::set_keepalive(), or turns it off with
172 : * set_keepalive(0, 0).
173 : * @return The underlying WebSocket, or nullptr if already released
174 : */
175 : std::shared_ptr<asio::websocket> release_socket();
176 :
177 : // ============================================
178 : // Async/Callback API (for async_client usage)
179 : // ============================================
180 :
181 : /**
182 : * Set callback for incoming messages.
183 : * Used with run() for event-driven operation.
184 : */
185 3 : void on_message(message_callback callback) { on_message_ = std::move(callback); }
186 :
187 : /**
188 : * Set callback for connection close.
189 : */
190 12 : void on_close(close_callback callback) { on_close_ = std::move(callback); }
191 :
192 : /**
193 : * Set callback for errors.
194 : */
195 : void on_error(error_callback callback) { on_error_ = std::move(callback); }
196 :
197 : /**
198 : * Start the message receive loop (non-blocking).
199 : * Messages are delivered via on_message callback, on the executor of the WebSocket, while
200 : * this websocket_client lives (the destructor stops it).
201 : */
202 : void run();
203 :
204 : /**
205 : * Stop the message loop and close connection.
206 : */
207 : void stop();
208 :
209 : // ============================================
210 : // Coroutine API (for advanced use)
211 : // ============================================
212 :
213 : /**
214 : * Send text message asynchronously.
215 : */
216 : awaitable<bool> send_text_async(std::string message);
217 :
218 : /**
219 : * Send binary message asynchronously.
220 : */
221 : awaitable<bool> send_binary_async(std::vector<uint8_t> data);
222 :
223 : /**
224 : * Receive message asynchronously: as receive().
225 : */
226 : awaitable<std::pair<std::string, bool>> receive_async();
227 :
228 : /**
229 : * Close connection asynchronously.
230 : */
231 : awaitable<void> close_async();
232 :
233 : private:
234 : friend class http_client_base;
235 :
236 : // As above, on an io_context owned by the client that opened it (see owned_io_context_)
237 : websocket_client(std::shared_ptr<asio::websocket> ws, bool owned_io_context);
238 :
239 : std::shared_ptr<asio::websocket> websocket_;
240 : std::atomic<bool> running_{false};
241 : // Whether the io_context of the socket belongs to the client that opened it, which runs
242 : // it only within its synchronous calls (http::client): the synchronous calls wait running
243 : // it. Otherwise something else runs it (the workers of an async_client...), and they wait
244 : // without running it
245 : bool owned_io_context_{false};
246 :
247 : // Callbacks
248 : message_callback on_message_;
249 : close_callback on_close_;
250 : error_callback on_error_;
251 :
252 : // Internal message loop coroutine
253 : awaitable<void> message_loop();
254 :
255 : // The next whole message (it may be empty) and whether it is binary, or nothing once
256 : // reading fails (closed or lost)
257 : awaitable<std::optional<std::pair<std::string, bool>>> read_message();
258 : };
259 :
260 : } // namespace thinger::http
261 :
262 : #endif // THINGER_HTTP_WEBSOCKET_CLIENT_HPP
|
