• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

realm / realm-core / 1757

13 Oct 2023 04:18PM UTC coverage: 91.612% (+0.02%) from 91.59%
1757

push

Evergreen

realm-ci
Updated release notes

94276 of 173488 branches covered (0.0%)

230559 of 251668 relevant lines covered (91.61%)

6451035.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.24
/src/realm/sync/network/default_socket.cpp
1
#include <realm/sync/network/default_socket.hpp>
2

3
#include <realm/sync/binding_callback_thread_observer.hpp>
4
#include <realm/sync/network/network.hpp>
5
#include <realm/sync/network/network_ssl.hpp>
6
#include <realm/sync/network/websocket.hpp>
7
#include <realm/util/random.hpp>
8
#include <realm/util/scope_exit.hpp>
9

10
namespace realm::sync::websocket {
11

12
namespace {
13

14
///
15
/// DefaultWebSocketImpl - websocket implementation for the default socket provider
16
///
17
class DefaultWebSocketImpl final : public DefaultWebSocket, public Config {
18
public:
19
    DefaultWebSocketImpl(const std::shared_ptr<util::Logger>& logger_ptr, network::Service& service,
20
                         std::mt19937_64& random, const std::string user_agent,
21
                         std::unique_ptr<WebSocketObserver> observer, WebSocketEndpoint&& endpoint)
22
        : m_logger_ptr{logger_ptr}
23
        , m_logger{*m_logger_ptr}
24
        , m_random{random}
25
        , m_service{service}
26
        , m_user_agent{user_agent}
27
        , m_observer{std::move(observer)}
28
        , m_endpoint{std::move(endpoint)}
29
        , m_websocket(*this)
30
    {
3,236✔
31
        initiate_resolve();
3,236✔
32
    }
3,236✔
33

34
    void async_write_binary(util::Span<const char> data, SyncSocketProvider::FunctionHandler&& handler) override
35
    {
99,560✔
36
        m_websocket.async_write_binary(data.data(), data.size(), [handler = std::move(handler)]() {
98,976✔
37
            handler(Status::OK());
98,092✔
38
        });
98,092✔
39
    }
99,560✔
40

41
    std::string_view get_appservices_request_id() const noexcept override
42
    {
3,114✔
43
        return m_app_services_coid;
3,114✔
44
    }
3,114✔
45

46
    void force_handshake_response_for_testing(int status_code, std::string body = "") override
47
    {
12✔
48
        m_websocket.force_handshake_response_for_testing(status_code, body);
12✔
49
    }
12✔
50

51
    // public for HTTPClient CRTP, but not on the EZSocket interface, so de-facto private
52
    void async_read(char*, std::size_t, ReadCompletionHandler) override;
53
    void async_read_until(char*, std::size_t, char, ReadCompletionHandler) override;
54
    void async_write(const char*, std::size_t, WriteCompletionHandler) override;
55

56
private:
57
    using milliseconds_type = std::int_fast64_t;
58

59
    const std::shared_ptr<util::Logger>& websocket_get_logger() noexcept override
60
    {
3,236✔
61
        return m_logger_ptr;
3,236✔
62
    }
3,236✔
63
    std::mt19937_64& websocket_get_random() noexcept override
64
    {
102,770✔
65
        return m_random;
102,770✔
66
    }
102,770✔
67

68
    void websocket_handshake_completion_handler(const HTTPHeaders& headers) override
69
    {
3,118✔
70
        const std::string empty;
3,118✔
71
        if (auto it = headers.find("X-Appservices-Request-Id"); it != headers.end()) {
3,118✔
72
            m_app_services_coid = it->second;
1,270✔
73
        }
1,270✔
74
        auto it = headers.find("Sec-WebSocket-Protocol");
3,118✔
75
        m_observer->websocket_connected_handler(it == headers.end() ? empty : it->second);
3,118✔
76
    }
3,118✔
77
    void websocket_read_error_handler(std::error_code ec) override
78
    {
476✔
79
        m_logger.error("Reading failed: %1", ec.message()); // Throws
476✔
80
        constexpr bool was_clean = false;
476✔
81
        websocket_error_and_close_handler(was_clean, WebSocketError::websocket_read_error, ec.message());
476✔
82
    }
476✔
83
    void websocket_write_error_handler(std::error_code ec) override
84
    {
×
85
        m_logger.error("Writing failed: %1", ec.message()); // Throws
×
86
        constexpr bool was_clean = false;
×
87
        websocket_error_and_close_handler(was_clean, WebSocketError::websocket_write_error, ec.message());
×
88
    }
×
89
    void websocket_handshake_error_handler(std::error_code ec, const HTTPHeaders*,
90
                                           const std::string_view* body) override
91
    {
48✔
92
        WebSocketError error = WebSocketError::websocket_ok;
48✔
93
        bool was_clean = true;
48✔
94

24✔
95
        if (ec == websocket::HttpError::bad_response_301_moved_permanently ||
48✔
96
            ec == websocket::HttpError::bad_response_308_permanent_redirect) {
44✔
97
            error = WebSocketError::websocket_moved_permanently;
12✔
98
        }
12✔
99
        else if (ec == websocket::HttpError::bad_response_3xx_redirection) {
36✔
100
            error = WebSocketError::websocket_retry_error;
×
101
            was_clean = false;
×
102
        }
×
103
        else if (ec == websocket::HttpError::bad_response_401_unauthorized) {
36✔
104
            error = WebSocketError::websocket_unauthorized;
36✔
105
        }
36✔
106
        else if (ec == websocket::HttpError::bad_response_403_forbidden) {
×
107
            error = WebSocketError::websocket_forbidden;
×
108
        }
×
109
        else if (ec == websocket::HttpError::bad_response_5xx_server_error ||
×
110
                 ec == websocket::HttpError::bad_response_500_internal_server_error ||
×
111
                 ec == websocket::HttpError::bad_response_502_bad_gateway ||
×
112
                 ec == websocket::HttpError::bad_response_503_service_unavailable ||
×
113
                 ec == websocket::HttpError::bad_response_504_gateway_timeout) {
×
114
            error = WebSocketError::websocket_internal_server_error;
×
115
            was_clean = false;
×
116
        }
×
117
        else {
×
118
            error = WebSocketError::websocket_fatal_error;
×
119
            was_clean = false;
×
120
            if (body) {
×
121
                std::string_view identifier = "REALM_SYNC_PROTOCOL_MISMATCH";
×
122
                auto i = body->find(identifier);
×
123
                if (i != std::string_view::npos) {
×
124
                    std::string_view rest = body->substr(i + identifier.size());
×
125
                    // FIXME: Use std::string_view::begins_with() in C++20.
126
                    auto begins_with = [](std::string_view string, std::string_view prefix) {
×
127
                        return (string.size() >= prefix.size() &&
×
128
                                std::equal(string.data(), string.data() + prefix.size(), prefix.data()));
×
129
                    };
×
130
                    if (begins_with(rest, ":CLIENT_TOO_OLD")) {
×
131
                        error = WebSocketError::websocket_client_too_old;
×
132
                    }
×
133
                    else if (begins_with(rest, ":CLIENT_TOO_NEW")) {
×
134
                        error = WebSocketError::websocket_client_too_new;
×
135
                    }
×
136
                    else {
×
137
                        // Other more complicated forms of mismatch
138
                        error = WebSocketError::websocket_protocol_mismatch;
×
139
                    }
×
140
                    was_clean = true;
×
141
                }
×
142
            }
×
143
        }
×
144

24✔
145
        websocket_error_and_close_handler(was_clean, error, ec.message());
48✔
146
    }
48✔
147
    void websocket_protocol_error_handler(std::error_code ec) override
148
    {
×
149
        constexpr bool was_clean = false;
×
150
        websocket_error_and_close_handler(was_clean, WebSocketError::websocket_protocol_error, ec.message());
×
151
    }
×
152
    bool websocket_close_message_received(WebSocketError code, std::string_view message) override
153
    {
80✔
154
        constexpr bool was_clean = true;
80✔
155

40✔
156
        return websocket_error_and_close_handler(was_clean, code, message);
80✔
157
    }
80✔
158
    bool websocket_error_and_close_handler(bool was_clean, WebSocketError code, std::string_view reason)
159
    {
618✔
160
        if (!was_clean) {
618✔
161
            m_observer->websocket_error_handler();
490✔
162
        }
490✔
163
        return m_observer->websocket_closed_handler(was_clean, code, reason);
618✔
164
    }
618✔
165
    bool websocket_binary_message_received(const char* ptr, std::size_t size) override
166
    {
80,994✔
167
        return m_observer->websocket_binary_message_received(util::Span<const char>(ptr, size));
80,994✔
168
    }
80,994✔
169

170
    void initiate_resolve();
171
    void handle_resolve(std::error_code, network::Endpoint::List);
172
    void initiate_tcp_connect(network::Endpoint::List, std::size_t);
173
    void handle_tcp_connect(std::error_code, network::Endpoint::List, std::size_t);
174
    void initiate_http_tunnel();
175
    void initiate_websocket_or_ssl_handshake();
176
    void initiate_ssl_handshake();
177
    void handle_ssl_handshake(std::error_code);
178
    void initiate_websocket_handshake();
179
    void handle_connection_established();
180

181
    void schedule_urgent_ping();
182
    void initiate_ping_delay(milliseconds_type now);
183
    void handle_ping_delay();
184
    void initiate_pong_timeout();
185
    void handle_pong_timeout();
186

187
    const std::shared_ptr<util::Logger> m_logger_ptr;
188
    util::Logger& m_logger;
189
    std::mt19937_64& m_random;
190
    network::Service& m_service;
191
    const std::string m_user_agent;
192
    std::string m_app_services_coid;
193

194
    std::unique_ptr<WebSocketObserver> m_observer;
195

196
    const WebSocketEndpoint m_endpoint;
197
    util::Optional<network::Resolver> m_resolver;
198
    util::Optional<network::Socket> m_socket;
199
    util::Optional<network::ssl::Context> m_ssl_context;
200
    util::Optional<network::ssl::Stream> m_ssl_stream;
201
    network::ReadAheadBuffer m_read_ahead_buffer;
202
    websocket::Socket m_websocket;
203
    util::Optional<HTTPClient<DefaultWebSocketImpl>> m_proxy_client;
204
};
205

206

207
void DefaultWebSocketImpl::async_read(char* buffer, std::size_t size, ReadCompletionHandler handler)
208
{
180,992✔
209
    REALM_ASSERT(m_socket);
180,992✔
210
    if (m_ssl_stream) {
180,992✔
211
        m_ssl_stream->async_read(buffer, size, m_read_ahead_buffer, std::move(handler)); // Throws
154✔
212
    }
154✔
213
    else {
180,838✔
214
        m_socket->async_read(buffer, size, m_read_ahead_buffer, std::move(handler)); // Throws
180,838✔
215
    }
180,838✔
216
}
180,992✔
217

218

219
void DefaultWebSocketImpl::async_read_until(char* buffer, std::size_t size, char delim, ReadCompletionHandler handler)
220
{
27,478✔
221
    REALM_ASSERT(m_socket);
27,478✔
222
    if (m_ssl_stream) {
27,478✔
223
        m_ssl_stream->async_read_until(buffer, size, delim, m_read_ahead_buffer, std::move(handler)); // Throws
98✔
224
    }
98✔
225
    else {
27,380✔
226
        m_socket->async_read_until(buffer, size, delim, m_read_ahead_buffer, std::move(handler)); // Throws
27,380✔
227
    }
27,380✔
228
}
27,478✔
229

230

231
void DefaultWebSocketImpl::async_write(const char* data, std::size_t size, WriteCompletionHandler handler)
232
{
102,780✔
233
    REALM_ASSERT(m_socket);
102,780✔
234
    if (m_ssl_stream) {
102,780✔
235
        m_ssl_stream->async_write(data, size, std::move(handler)); // Throws
84✔
236
    }
84✔
237
    else {
102,696✔
238
        m_socket->async_write(data, size, std::move(handler)); // Throws
102,696✔
239
    }
102,696✔
240
}
102,780✔
241

242

243
void DefaultWebSocketImpl::initiate_resolve()
244
{
3,238✔
245
    const std::string& address = m_endpoint.proxy ? m_endpoint.proxy->address : m_endpoint.address;
3,238✔
246
    const port_type& port = m_endpoint.proxy ? m_endpoint.proxy->port : m_endpoint.port;
3,238✔
247

1,608✔
248
    if (m_endpoint.proxy) {
3,238✔
249
        // logger.detail("Using %1 proxy", proxy->type); // Throws
250
    }
×
251

1,608✔
252
    m_logger.detail("Resolving '%1:%2'", address, port); // Throws
3,238✔
253

1,608✔
254
    network::Resolver::Query query(address, util::to_string(port)); // Throws
3,238✔
255
    auto handler = [this](std::error_code ec, network::Endpoint::List endpoints) {
3,234✔
256
        // If the operation is aborted, the connection object may have been
1,606✔
257
        // destroyed.
1,606✔
258
        if (ec != util::error::operation_aborted)
3,232✔
259
            handle_resolve(ec, std::move(endpoints)); // Throws
3,230✔
260
    };
3,232✔
261
    m_resolver.emplace(m_service);                                   // Throws
3,238✔
262
    m_resolver->async_resolve(std::move(query), std::move(handler)); // Throws
3,238✔
263
}
3,238✔
264

265

266
void DefaultWebSocketImpl::handle_resolve(std::error_code ec, network::Endpoint::List endpoints)
267
{
3,230✔
268
    if (ec) {
3,230✔
269
        m_logger.error("Failed to resolve '%1:%2': %3", m_endpoint.address, m_endpoint.port, ec.message()); // Throws
4✔
270
        constexpr bool was_clean = false;
4✔
271
        websocket_error_and_close_handler(was_clean, WebSocketError::websocket_resolve_failed,
4✔
272
                                          ec.message()); // Throws
4✔
273
        return;
4✔
274
    }
4✔
275

1,602✔
276
    initiate_tcp_connect(std::move(endpoints), 0); // Throws
3,226✔
277
}
3,226✔
278

279

280
void DefaultWebSocketImpl::initiate_tcp_connect(network::Endpoint::List endpoints, std::size_t i)
281
{
3,226✔
282
    REALM_ASSERT(i < endpoints.size());
3,226✔
283

1,602✔
284
    network::Endpoint ep = *(endpoints.begin() + i);
3,226✔
285
    std::size_t n = endpoints.size();
3,226✔
286
    m_socket.emplace(m_service); // Throws
3,226✔
287
    m_socket->async_connect(ep, [this, endpoints = std::move(endpoints), i](std::error_code ec) mutable {
3,226✔
288
        // If the operation is aborted, the connection object may have been
1,602✔
289
        // destroyed.
1,602✔
290
        if (ec != util::error::operation_aborted)
3,226✔
291
            handle_tcp_connect(ec, std::move(endpoints), i); // Throws
3,222✔
292
    });
3,226✔
293
    m_logger.detail("Connecting to endpoint '%1:%2' (%3/%4)", ep.address(), ep.port(), (i + 1), n); // Throws
3,226✔
294
}
3,226✔
295

296

297
void DefaultWebSocketImpl::handle_tcp_connect(std::error_code ec, network::Endpoint::List endpoints, std::size_t i)
298
{
3,222✔
299
    REALM_ASSERT(i < endpoints.size());
3,222✔
300
    const network::Endpoint& ep = *(endpoints.begin() + i);
3,222✔
301
    if (ec) {
3,222✔
302
        m_logger.error("Failed to connect to endpoint '%1:%2': %3", ep.address(), ep.port(),
×
303
                       ec.message()); // Throws
×
304
        std::size_t i_2 = i + 1;
×
305
        if (i_2 < endpoints.size()) {
×
306
            initiate_tcp_connect(std::move(endpoints), i_2); // Throws
×
307
            return;
×
308
        }
×
309
        // All endpoints failed
310
        m_logger.error("Failed to connect to '%1:%2': All endpoints failed", m_endpoint.address, m_endpoint.port);
×
311
        constexpr bool was_clean = false;
×
312
        websocket_error_and_close_handler(was_clean, WebSocketError::websocket_connection_failed,
×
313
                                          ec.message()); // Throws
×
314
        return;
×
315
    }
×
316

1,598✔
317
    REALM_ASSERT(m_socket);
3,222✔
318
    network::Endpoint ep_2 = m_socket->local_endpoint();
3,222✔
319
    m_logger.info("Connected to endpoint '%1:%2' (from '%3:%4')", ep.address(), ep.port(), ep_2.address(),
3,222✔
320
                  ep_2.port()); // Throws
3,222✔
321

1,598✔
322
    // TODO: Handle HTTPS proxies
1,598✔
323
    if (m_endpoint.proxy) {
3,222✔
324
        initiate_http_tunnel(); // Throws
×
325
        return;
×
326
    }
×
327

1,598✔
328
    initiate_websocket_or_ssl_handshake(); // Throws
3,222✔
329
}
3,222✔
330

331
void DefaultWebSocketImpl::initiate_websocket_or_ssl_handshake()
332
{
3,222✔
333
    if (m_endpoint.is_ssl) {
3,222✔
334
        initiate_ssl_handshake(); // Throws
24✔
335
    }
24✔
336
    else {
3,198✔
337
        initiate_websocket_handshake(); // Throws
3,198✔
338
    }
3,198✔
339
}
3,222✔
340

341
void DefaultWebSocketImpl::initiate_http_tunnel()
342
{
×
343
    HTTPRequest req;
×
344
    req.method = HTTPMethod::Connect;
×
345
    req.headers.emplace("Host", util::format("%1:%2", m_endpoint.address, m_endpoint.port));
×
346
    // TODO handle proxy authorization
347

348
    m_proxy_client.emplace(*this, m_logger_ptr);
×
349
    auto handler = [this](HTTPResponse response, std::error_code ec) {
×
350
        if (ec && ec != util::error::operation_aborted) {
×
351
            m_logger.error("Failed to establish HTTP tunnel: %1", ec.message());
×
352
            constexpr bool was_clean = false;
×
353
            websocket_error_and_close_handler(was_clean, WebSocketError::websocket_connection_failed,
×
354
                                              ec.message()); // Throws
×
355
            return;
×
356
        }
×
357

358
        if (response.status != HTTPStatus::Ok) {
×
359
            m_logger.error("Proxy server returned response '%1 %2'", response.status, response.reason); // Throws
×
360
            constexpr bool was_clean = false;
×
361
            websocket_error_and_close_handler(was_clean, WebSocketError::websocket_connection_failed,
×
362
                                              response.reason); // Throws
×
363
            return;
×
364
        }
×
365

366
        initiate_websocket_or_ssl_handshake(); // Throws
×
367
    };
×
368

369
    m_proxy_client->async_request(req, std::move(handler)); // Throws
×
370
}
×
371

372
void DefaultWebSocketImpl::initiate_ssl_handshake()
373
{
24✔
374
    using namespace network::ssl;
24✔
375

14✔
376
    if (!m_ssl_context) {
24✔
377
        m_ssl_context.emplace(); // Throws
24✔
378
        if (m_endpoint.verify_servers_ssl_certificate) {
24✔
379
            if (m_endpoint.ssl_trust_certificate_path) {
20✔
380
                m_ssl_context->use_verify_file(*m_endpoint.ssl_trust_certificate_path); // Throws
10✔
381
            }
10✔
382
            else if (!m_endpoint.ssl_verify_callback) {
10✔
383
                m_ssl_context->use_default_verify(); // Throws
4✔
384
#if REALM_INCLUDE_CERTS
2✔
385
                // On platforms like Windows or Android where OpenSSL is not normally found
2✔
386
                // `use_default_verify()` won't actually be able to load any default certificates.
2✔
387
                // That's why we bundle a set of trusted certificates ourselves.
2✔
388
                m_ssl_context->use_included_certificate_roots(); // Throws
2✔
389
#endif
2✔
390
            }
4✔
391
        }
20✔
392
    }
24✔
393

14✔
394
    m_ssl_stream.emplace(*m_socket, *m_ssl_context, Stream::client); // Throws
24✔
395
    m_ssl_stream->set_logger(m_logger_ptr.get());
24✔
396
    m_ssl_stream->set_host_name(m_endpoint.address); // Throws
24✔
397
    if (m_endpoint.verify_servers_ssl_certificate) {
24✔
398
        m_ssl_stream->set_verify_mode(VerifyMode::peer); // Throws
20✔
399
        m_ssl_stream->set_server_port(m_endpoint.port);
20✔
400
        if (!m_endpoint.ssl_trust_certificate_path) {
20✔
401
            if (m_endpoint.ssl_verify_callback) {
10✔
402
                m_ssl_stream->use_verify_callback(m_endpoint.ssl_verify_callback);
6✔
403
            }
6✔
404
        }
10✔
405
    }
20✔
406

14✔
407
    auto handler = [this](std::error_code ec) {
24✔
408
        // If the operation is aborted, the connection object may have been
14✔
409
        // destroyed.
14✔
410
        if (ec != util::error::operation_aborted)
24✔
411
            handle_ssl_handshake(ec); // Throws
24✔
412
    };
24✔
413
    m_ssl_stream->async_handshake(std::move(handler)); // Throws
24✔
414

14✔
415
    // FIXME: We also need to perform the SSL shutdown operation somewhere
14✔
416
}
24✔
417

418

419
void DefaultWebSocketImpl::handle_ssl_handshake(std::error_code ec)
420
{
24✔
421
    if (ec) {
24✔
422
        REALM_ASSERT(ec != util::error::operation_aborted);
10✔
423
        constexpr bool was_clean = false;
10✔
424
        WebSocketError parsed_error_code;
10✔
425
        if (ec == network::ssl::Errors::tls_handshake_failed) {
10✔
426
            parsed_error_code = WebSocketError::websocket_tls_handshake_failed;
10✔
427
        }
10✔
428
        else {
×
429
            parsed_error_code = WebSocketError::websocket_connection_failed;
×
430
        }
×
431

6✔
432
        websocket_error_and_close_handler(was_clean, parsed_error_code, ec.message()); // Throws
10✔
433
        return;
10✔
434
    }
10✔
435

8✔
436
    initiate_websocket_handshake(); // Throws
14✔
437
}
14✔
438

439

440
void DefaultWebSocketImpl::initiate_websocket_handshake()
441
{
3,212✔
442
    auto headers = HTTPHeaders(m_endpoint.headers.begin(), m_endpoint.headers.end());
3,212✔
443
    headers["User-Agent"] = m_user_agent;
3,212✔
444

1,592✔
445
    // Compute the value of the "Host" header.
1,592✔
446
    const std::uint_fast16_t default_port = (m_endpoint.is_ssl ? 443 : 80);
3,206✔
447
    auto host = m_endpoint.port == default_port ? m_endpoint.address
1,592✔
448
                                                : util::format("%1:%2", m_endpoint.address, m_endpoint.port);
3,212✔
449

1,592✔
450
    // Convert the list of protocols to a string
1,592✔
451
    std::ostringstream protocol_list;
3,212✔
452
    protocol_list.exceptions(std::ios_base::failbit | std::ios_base::badbit);
3,212✔
453
    protocol_list.imbue(std::locale::classic());
3,212✔
454
    if (m_endpoint.protocols.size() > 1)
3,212✔
455
        std::copy(m_endpoint.protocols.begin(), m_endpoint.protocols.end() - 1,
3,212✔
456
                  std::ostream_iterator<std::string>(protocol_list, ", "));
3,212✔
457
    protocol_list << m_endpoint.protocols.back();
3,212✔
458

1,592✔
459
    m_websocket.initiate_client_handshake(m_endpoint.path, std::move(host), protocol_list.str(),
3,212✔
460
                                          std::move(headers)); // Throws
3,212✔
461
}
3,212✔
462
} // namespace
463

464
///
465
/// DefaultSocketProvider - default socket provider implementation
466
///
467

468
DefaultSocketProvider::DefaultSocketProvider(const std::shared_ptr<util::Logger>& logger,
469
                                             const std::string user_agent,
470
                                             const std::shared_ptr<BindingCallbackThreadObserver>& observer_ptr,
471
                                             AutoStart auto_start)
472
    : m_logger_ptr{logger}
473
    , m_observer_ptr{observer_ptr}
474
    , m_service{}
475
    , m_random{}
476
    , m_user_agent{user_agent}
477
    , m_mutex{}
478
    , m_state{State::Stopped}
479
    , m_state_cv{}
480
    , m_thread{}
481
{
8,970✔
482
    REALM_ASSERT(m_logger_ptr);                     // Make sure the logger is valid
8,970✔
483
    util::seed_prng_nondeterministically(m_random); // Throws
8,970✔
484
    if (auto_start) {
8,970✔
485
        start();
8,218✔
486
    }
8,218✔
487
}
8,970✔
488

489
DefaultSocketProvider::~DefaultSocketProvider()
490
{
8,970✔
491
    m_logger_ptr->trace("Default event loop teardown");
8,970✔
492
    // Wait for the thread to stop
4,416✔
493
    stop(true);
8,970✔
494
    // Shutting down - no need to lock mutex before check
4,416✔
495
    REALM_ASSERT(m_state == State::Stopped);
8,970✔
496
}
8,970✔
497

498
void DefaultSocketProvider::start()
499
{
8,970✔
500
    std::unique_lock<std::mutex> lock(m_mutex);
8,970✔
501
    // Has the thread already been started or is running
4,416✔
502
    if (m_state == State::Starting || m_state == State::Running)
8,970✔
503
        return; // early return
×
504

4,416✔
505
    // If the thread has been previously run, make sure it has been joined first
4,416✔
506
    if (m_state == State::Stopping) {
8,970✔
507
        state_wait_for(lock, State::Stopped);
×
508
    }
×
509

4,416✔
510
    m_logger_ptr->trace("Default event loop: start()");
8,970✔
511
    REALM_ASSERT(m_state == State::Stopped);
8,970✔
512

4,416✔
513
    do_state_update(lock, State::Starting);
8,970✔
514
    m_thread = std::thread{&DefaultSocketProvider::event_loop, this};
8,970✔
515
    // Wait for the thread to start before continuing
4,416✔
516
    state_wait_for(lock, State::Running);
8,970✔
517
}
8,970✔
518

519
void DefaultSocketProvider::OnlyForTesting::run_event_loop_on_current_thread(DefaultSocketProvider* provider)
520
{
4✔
521
    {
4✔
522
        std::unique_lock<std::mutex> lk(provider->m_mutex);
4✔
523
        REALM_ASSERT(provider->m_state == State::Stopped);
4✔
524
        provider->do_state_update(lk, State::Starting);
4✔
525
    }
4✔
526

2✔
527
    provider->event_loop();
4✔
528
}
4✔
529

530
void DefaultSocketProvider::OnlyForTesting::prep_event_loop_for_restart(DefaultSocketProvider* provider)
531
{
4✔
532
    std::unique_lock<std::mutex> lk(provider->m_mutex);
4✔
533
    REALM_ASSERT(provider->m_state == State::Stopped);
4✔
534
    provider->m_service.reset();
4✔
535
}
4✔
536

537
void DefaultSocketProvider::event_loop()
538
{
8,974✔
539
    m_logger_ptr->trace("Default event loop: thread running");
8,974✔
540
    // Calls will_destroy_thread() when destroyed
4,418✔
541
    auto will_destroy_thread = util::make_scope_exit([&]() noexcept {
8,974✔
542
        m_logger_ptr->trace("Default event loop: thread exiting");
8,974✔
543
        if (m_observer_ptr)
8,974✔
544
            m_observer_ptr->will_destroy_thread();
×
545

4,418✔
546
        std::unique_lock<std::mutex> lock(m_mutex);
8,974✔
547
        // Did we get here due to an unhandled exception?
4,418✔
548
        if (m_state != State::Stopping) {
8,974✔
549
            m_logger_ptr->error("Default event loop: thread exited unexpectedly");
4✔
550
        }
4✔
551
        m_state = State::Stopped;
8,974✔
552
        lock.unlock();
8,974✔
553
        m_state_cv.notify_all();
8,974✔
554
    });
8,974✔
555

4,418✔
556
    if (m_observer_ptr)
8,974✔
557
        m_observer_ptr->did_create_thread();
×
558

4,418✔
559
    {
8,974✔
560
        std::lock_guard<std::mutex> lock(m_mutex);
8,974✔
561
        REALM_ASSERT(m_state == State::Starting);
8,974✔
562
    }
8,974✔
563

4,418✔
564
    // We update the state to Running from inside the event loop so that start() is blocked until
4,418✔
565
    // the event loop is actually ready to receive work.
4,418✔
566
    m_service.post([this, my_generation = ++m_event_loop_generation](Status status) {
8,974✔
567
        if (status == ErrorCodes::OperationAborted) {
8,974✔
568
            return;
×
569
        }
×
570

4,418✔
571
        REALM_ASSERT(status.is_ok());
8,974✔
572

4,418✔
573
        std::unique_lock<std::mutex> lock(m_mutex);
8,974✔
574
        // This is a callback from a previous generation
4,418✔
575
        if (m_event_loop_generation != my_generation) {
8,974✔
576
            return;
4✔
577
        }
4✔
578
        if (m_state == State::Stopping) {
8,970✔
579
            return;
×
580
        }
×
581
        m_logger_ptr->trace("Default event loop: service run");
8,970✔
582
        REALM_ASSERT(m_state == State::Starting);
8,970✔
583
        do_state_update(lock, State::Running);
8,970✔
584
    });
8,970✔
585

4,418✔
586
    // If there is no event loop observer or handle_error function registered, then just
4,418✔
587
    // allow the exception to bubble to the top so we can get a true stack trace
4,418✔
588
    if (!m_observer_ptr || !m_observer_ptr->has_handle_error()) {
8,974!
589
        m_service.run_until_stopped(); // Throws
8,974✔
590
    }
8,974✔
591
    else {
×
592
        try {
×
593
            m_service.run_until_stopped(); // Throws
×
594
        }
×
595
        catch (const std::exception& e) {
×
596
            REALM_ASSERT(m_observer_ptr); // should not change while event loop is running
×
597
            std::unique_lock<std::mutex> lock(m_mutex);
×
598
            // Service is no longer running, event loop thread is stopping
599
            do_state_update(lock, State::Stopping);
×
600
            lock.unlock();
×
601
            m_logger_ptr->error("Default event loop exception: ", e.what());
×
602
            // If the error was not handled by the thread loop observer, then rethrow
603
            if (!m_observer_ptr->handle_error(e))
×
604
                throw;
×
605
        }
×
606
    }
×
607
}
8,974✔
608

609
void DefaultSocketProvider::stop(bool wait_for_stop)
610
{
8,974✔
611
    std::unique_lock<std::mutex> lock(m_mutex);
8,974✔
612

4,418✔
613
    // Do nothing if the thread is not started or running or stop has already been called
4,418✔
614
    if (m_state == State::Starting || m_state == State::Running) {
8,974✔
615
        m_logger_ptr->trace("Default event loop: stop()");
8,970✔
616
        do_state_update(lock, State::Stopping);
8,970✔
617
        // Updating state to Stopping will free a start() if it is waiting for the thread to
4,416✔
618
        // start and may cause the thread to exit early before calling service.run()
4,416✔
619
        m_service.stop(); // Unblocks m_service.run()
8,970✔
620
    }
8,970✔
621

4,418✔
622
    // Wait until the thread is stopped (exited) if requested
4,418✔
623
    if (wait_for_stop) {
8,974✔
624
        m_logger_ptr->trace("Default event loop: wait for stop");
8,974✔
625
        state_wait_for(lock, State::Stopped);
8,974✔
626
        if (m_thread.joinable()) {
8,974✔
627
            m_thread.join();
8,970✔
628
        }
8,970✔
629
    }
8,974✔
630
}
8,974✔
631

632
//                    +---------------------------------------+
633
//                   \/                                       |
634
// State Machine: Stopped -> Starting -> Running -> Stopping -+
635
//                              |           |          ^
636
//                              +----------------------+
637

638
void DefaultSocketProvider::do_state_update(std::unique_lock<std::mutex>&, State new_state)
639
{
26,914✔
640
    // m_state_mutex should already be locked...
13,250✔
641
    m_state = new_state;
26,914✔
642
    m_state_cv.notify_all(); // Let any waiters check the state
26,914✔
643
}
26,914✔
644

645
void DefaultSocketProvider::state_wait_for(std::unique_lock<std::mutex>& lock, State expected_state)
646
{
17,944✔
647
    // Check for condition already met or superseded
8,834✔
648
    if (m_state >= expected_state)
17,944✔
649
        return;
4✔
650

8,832✔
651
    m_state_cv.wait(lock, [this, expected_state]() {
35,880✔
652
        // are we there yet?
17,664✔
653
        if (m_state < expected_state)
35,880✔
654
            return false;
17,940✔
655
        return true;
17,940✔
656
    });
17,940✔
657
}
17,940✔
658

659
std::unique_ptr<WebSocketInterface> DefaultSocketProvider::connect(std::unique_ptr<WebSocketObserver> observer,
660
                                                                   WebSocketEndpoint&& endpoint)
661
{
3,238✔
662
    return std::make_unique<DefaultWebSocketImpl>(m_logger_ptr, m_service, m_random, m_user_agent,
3,238✔
663
                                                  std::move(observer), std::move(endpoint));
3,238✔
664
}
3,238✔
665

666
} // namespace realm::sync::websocket
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc