Add scheduler abstraction + default impl (#35)

Author: BrennanConroy

include/signalrclient/hub_connection.h

@@ -55,7 +55,7 @@ namespace signalr
         friend class hub_connection_builder;
 
         explicit hub_connection(const std::string& url, trace_level trace_level = trace_level::info,
-            std::shared_ptr<log_writer> log_writer = nullptr, std::shared_ptr<http_client> http_client = nullptr,
+            std::shared_ptr<log_writer> log_writer = nullptr, std::function<std::shared_ptr<http_client>(const signalr_client_config&)> http_client_factory = nullptr,
             std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> websocket_factory = nullptr, bool skip_negotiation = false);
 
         std::shared_ptr<hub_connection_impl> m_pImpl;

include/signalrclient/hub_connection_builder.h

@@ -31,7 +31,7 @@ namespace signalr
 
         SIGNALRCLIENT_API hub_connection_builder& with_websocket_factory(std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> factory);
 
-        SIGNALRCLIENT_API hub_connection_builder& with_http_client(std::shared_ptr<http_client> http_client);
+        SIGNALRCLIENT_API hub_connection_builder& with_http_client_factory(std::function<std::shared_ptr<http_client>(const signalr_client_config&)> http_client_factory);
 
         SIGNALRCLIENT_API hub_connection_builder& skip_negotiation(bool skip = true);
 
@@ -43,7 +43,7 @@ namespace signalr
         std::shared_ptr<log_writer> m_logger;
         trace_level m_log_level;
         std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> m_websocket_factory;
-        std::shared_ptr<http_client> m_http_client;
+        std::function<std::shared_ptr<http_client>(const signalr_client_config&)> m_http_client_factory;
         bool m_skip_negotiation = false;
     };
 }

include/signalrclient/scheduler.h

@@ -0,0 +1,21 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+#pragma once
+
+#include <exception>
+#include <functional>
+#include <chrono>
+
+namespace signalr
+{
+    typedef std::function<void()> signalr_base_cb;
+
+    struct scheduler
+    {
+        virtual void schedule(const signalr_base_cb& cb, std::chrono::milliseconds delay = std::chrono::milliseconds::zero()) = 0;
+
+        virtual ~scheduler() {}
+    };
+}

include/signalrclient/signalr_client_config.h

@@ -15,6 +15,8 @@
 #include "_exports.h"
 #include <map>
 #include <string>
+#include "scheduler.h"
+#include <memory>
 
 namespace signalr
 {
@@ -36,15 +38,20 @@ namespace signalr
         SIGNALRCLIENT_API void __cdecl set_websocket_client_config(const web::websockets::client::websocket_client_config& websocket_client_config);
 #endif
 
+        SIGNALRCLIENT_API __cdecl signalr_client_config();
+
         SIGNALRCLIENT_API const std::map<std::string, std::string>& __cdecl get_http_headers() const noexcept;
         SIGNALRCLIENT_API std::map<std::string, std::string>& __cdecl get_http_headers() noexcept;
         SIGNALRCLIENT_API void __cdecl set_http_headers(const std::map<std::string, std::string>& http_headers);
+        SIGNALRCLIENT_API void __cdecl set_scheduler(std::shared_ptr<scheduler> scheduler);
+        SIGNALRCLIENT_API const std::shared_ptr<scheduler>& __cdecl get_scheduler() const noexcept;
 
     private:
 #ifdef USE_CPPRESTSDK
         web::http::client::http_client_config m_http_client_config;
         web::websockets::client::websocket_client_config m_websocket_client_config;
 #endif
         std::map<std::string, std::string> m_http_headers;
+        std::shared_ptr<scheduler> m_scheduler;
     };
 }

run-tests.sh

@@ -11,7 +11,7 @@ eval $* &
 PID=$!
 
 # Timeout duration
-TIME=120
+TIME=800
 
 while (($TIME > 0)); do
   # polling interval
@@ -34,4 +34,4 @@ done
 kill -9 $PID
 
 echo "Process took too long"
-exit $EXITCODE
+exit 1

src/signalrclient/CMakeLists.txt

@@ -20,6 +20,7 @@ set (SOURCES
   transport_factory.cpp
   url_builder.cpp
   websocket_transport.cpp
+  signalr_default_scheduler.cpp
   ../../third_party_code/cpprestsdk/uri.cpp
   ../../third_party_code/cpprestsdk/uri_builder.cpp
 )

src/signalrclient/cancellation_token.h

@@ -27,17 +27,37 @@ namespace signalr
         cancellation_token(const cancellation_token&) = delete;
         cancellation_token& operator=(const cancellation_token&) = delete;
 
+        ~cancellation_token()
+        {
+            if (m_callback)
+            {
+                m_callback();
+            }
+        }
+
         void cancel()
         {
-            std::lock_guard<std::mutex> lock(m_lock);
-            m_signaled = true;
-            m_condition.notify_all();
+            std::function<void()> callback;
+            {
+                std::lock_guard<std::mutex> lock(m_lock);
+                m_signaled = true;
+                m_condition.notify_all();
+                callback = m_callback;
+                m_callback = nullptr;
+            } // unlock
+
+            if (callback)
+            {
+                callback();
+            }
         }
 
         void reset()
         {
             std::lock_guard<std::mutex> lock(m_lock);
+            assert(m_callback == nullptr);
             m_signaled = false;
+            m_callback = nullptr;
         }
 
         bool is_canceled() const
@@ -75,9 +95,33 @@ namespace signalr
                 throw canceled_exception();
             }
         }
+
+        void register_callback(std::function<void()> callback)
+        {
+            bool run_callback = false;
+            {
+                std::lock_guard<std::mutex> lock(m_lock);
+                assert(m_callback == nullptr);
+                if (m_signaled)
+                {
+                    run_callback = m_signaled;
+                }
+                else
+                {
+                    m_callback = callback;
+                }
+            } // unlock
+
+            if (run_callback)
+            {
+                callback();
+            }
+        }
+
     private:
         std::mutex m_lock;
         std::condition_variable m_condition;
         bool m_signaled;
+        std::function<void()> m_callback;
     };
 }

src/signalrclient/connection_impl.cpp

@@ -3,7 +3,6 @@
 // See the LICENSE file in the project root for more information.
 
 #include "stdafx.h"
-#include <thread>
 #include <algorithm>
 #include "constants.h"
 #include "connection_impl.h"
@@ -17,6 +16,7 @@
 #include <assert.h>
 #include "signalrclient/websocket_client.h"
 #include "default_websocket_client.h"
+#include "signalr_default_scheduler.h"
 
 namespace signalr
 {
@@ -26,42 +26,25 @@ namespace signalr
     }
 
     std::shared_ptr<connection_impl> connection_impl::create(const std::string& url, trace_level trace_level, const std::shared_ptr<log_writer>& log_writer,
-        std::shared_ptr<http_client> http_client, std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> websocket_factory, const bool skip_negotiation)
+        std::function<std::shared_ptr<http_client>(const signalr_client_config&)> http_client_factory, std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> websocket_factory, const bool skip_negotiation)
     {
         return std::shared_ptr<connection_impl>(new connection_impl(url, trace_level,
-            log_writer ? log_writer : std::make_shared<trace_log_writer>(), http_client, websocket_factory, skip_negotiation));
+            log_writer ? log_writer : std::make_shared<trace_log_writer>(), http_client_factory, websocket_factory, skip_negotiation));
     }
 
     connection_impl::connection_impl(const std::string& url, trace_level trace_level, const std::shared_ptr<log_writer>& log_writer,
-        std::unique_ptr<http_client> http_client, std::unique_ptr<transport_factory> transport_factory, const bool skip_negotiation)
-        : m_base_url(url), m_connection_state(connection_state::disconnected), m_logger(log_writer, trace_level), m_transport(nullptr),
-        m_transport_factory(std::move(transport_factory)), m_skip_negotiation(skip_negotiation), m_message_received([](const std::string&) noexcept {}), m_disconnected([]() noexcept {})
-    {
-        if (http_client != nullptr)
-        {
-            m_http_client = std::move(http_client);
-        }
-        else
-        {
-#ifdef USE_CPPRESTSDK
-            m_http_client = std::unique_ptr<class http_client>(new default_http_client());
-#endif
-        }
-    }
-
-    connection_impl::connection_impl(const std::string& url, trace_level trace_level, const std::shared_ptr<log_writer>& log_writer,
-        std::shared_ptr<http_client> http_client, std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> websocket_factory, const bool skip_negotiation)
+        std::function<std::shared_ptr<http_client>(const signalr_client_config&)> http_client_factory, std::function<std::shared_ptr<websocket_client>(const signalr_client_config&)> websocket_factory, const bool skip_negotiation)
         : m_base_url(url), m_connection_state(connection_state::disconnected), m_logger(log_writer, trace_level), m_transport(nullptr), m_skip_negotiation(skip_negotiation),
-        m_message_received([](const std::string&) noexcept {}), m_disconnected([]() noexcept {})
+        m_message_received([](const std::string&) noexcept {}), m_disconnected([]() noexcept {}), m_disconnect_cts(std::make_shared<cancellation_token>())
     {
-        if (http_client != nullptr)
+        if (http_client_factory != nullptr)
         {
-            m_http_client = std::move(http_client);
+            m_http_client_factory = std::move(http_client_factory);
         }
         else
         {
 #ifdef USE_CPPRESTSDK
-            m_http_client = std::unique_ptr<class http_client>(new default_http_client());
+            m_http_client_factory = [](const signalr_client_config&) { return std::unique_ptr<class http_client>(new default_http_client()); };
 #endif
         }
 
@@ -72,7 +55,7 @@ namespace signalr
 #endif
         }
 
-        m_transport_factory = std::unique_ptr<transport_factory>(new transport_factory(m_http_client, websocket_factory));
+        m_transport_factory = std::unique_ptr<transport_factory>(new transport_factory(m_http_client_factory, websocket_factory));
     }
 
     connection_impl::~connection_impl()
@@ -138,11 +121,18 @@ namespace signalr
             // there should not be any active transport at this point
             assert(!m_transport);
 
-            m_disconnect_cts = std::make_shared<cancellation_token>();
+            m_disconnect_cts->reset();
             m_start_completed_event.reset();
             m_connection_id = "";
         }
 
+        m_scheduler = m_signalr_client_config.get_scheduler();
+        if (!m_scheduler)
+        {
+            m_scheduler = std::make_shared<signalr_default_scheduler>();
+            m_signalr_client_config.set_scheduler(m_scheduler);
+        }
+
         start_negotiate(m_base_url, 0, callback);
     }
 
@@ -157,7 +147,7 @@ namespace signalr
         }
 
         std::weak_ptr<connection_impl> weak_connection = shared_from_this();
-        const auto& token = m_disconnect_cts;
+        const auto token = m_disconnect_cts;
 
         const auto transport_started = [weak_connection, callback, token](std::shared_ptr<transport> transport, std::exception_ptr exception)
         {
@@ -225,7 +215,8 @@ namespace signalr
             return start_transport(url, transport_started);
         }
 
-        negotiate::negotiate(*m_http_client, url, m_signalr_client_config,
+        auto http_client = m_http_client_factory(m_signalr_client_config);
+        negotiate::negotiate(http_client, url, m_signalr_client_config,
             [callback, weak_connection, redirect_count, token, url, transport_started](negotiation_response&& response, std::exception_ptr exception)
             {
                 auto connection = weak_connection.lock();
@@ -320,7 +311,7 @@ namespace signalr
         std::shared_ptr<std::mutex> connect_request_lock = std::make_shared<std::mutex>();
 
         auto weak_connection = std::weak_ptr<connection_impl>(connection);
-        const auto& disconnect_cts = m_disconnect_cts;
+        const auto disconnect_cts = m_disconnect_cts;
         const auto& logger = m_logger;
 
         auto transport = connection->m_transport_factory->create_transport(
@@ -406,39 +397,51 @@ namespace signalr
                 }
             });
 
-        std::thread([disconnect_cts, connect_request_done, connect_request_lock, callback, weak_connection]()
-        {
-            disconnect_cts->wait(5000);
+        disconnect_cts->register_callback([connect_request_done, connect_request_lock, callback]()
+            {
+                bool run_callback = false;
+                {
+                    std::lock_guard<std::mutex> lock(*connect_request_lock);
 
+                    // no op after connection started successfully
+                    if (*connect_request_done == false)
+                    {
+                        *connect_request_done = true;
+                        run_callback = true;
+                    }
+                } // unlock
+
+                if (run_callback)
+                {
+                    // The callback checks the disconnect_cts token and will handle it appropriately
+                    callback({}, nullptr);
+                }
+            });
+
+        timer(m_scheduler, [connect_request_done, connect_request_lock, callback](std::chrono::milliseconds duration) {
             bool run_callback = false;
             {
                 std::lock_guard<std::mutex> lock(*connect_request_lock);
+
                 // no op after connection started successfully
                 if (*connect_request_done == false)
                 {
+                    if (duration < std::chrono::seconds(5))
+                    {
+                        return false;
+                    }
                     *connect_request_done = true;
                     run_callback = true;
                 }
-            }
+            } // unlock
 
-            // if the disconnect_cts is canceled it means that the connection has been stopped or went out of scope in
-            // which case we should not throw due to timeout.
-            if (disconnect_cts->is_canceled())
+            if (run_callback)
             {
-                if (run_callback)
-                {
-                    // The callback checks the disconnect_cts token and will handle it appropriately
-                    callback({}, nullptr);
-                }
-            }
-            else
-            {
-                if (run_callback)
-                {
-                    callback({}, std::make_exception_ptr(signalr_exception("transport timed out when trying to connect")));
-                }
+                callback({}, std::make_exception_ptr(signalr_exception("transport timed out when trying to connect")));
             }
-        }).detach();
+
+            return true;
+        });
 
         connection->send_connect_request(transport, url, [callback, connect_request_done, connect_request_lock, transport](std::exception_ptr exception)
             {
@@ -597,6 +600,7 @@ namespace signalr
             const auto current_state = get_connection_state();
             if (current_state == connection_state::disconnected)
             {
+                m_disconnect_cts->cancel();
                 callback(nullptr);
                 return;
             }