Add public cancellation_token and pass to http_client (#64)

Author: BrennanConroy

include/signalrclient/cancellation_token.h

@@ -0,0 +1,36 @@
+// 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 <memory>
+
+namespace signalr
+{
+    class cancellation_token_source;
+
+    class cancellation_token
+    {
+    public:
+        /**
+         * Register a callback to run when this token is canceled or destructed.
+         * If the token is already canceled, the callback will run immediately.
+         */
+        void register_callback(std::function<void()> callback);
+
+        /**
+         * Check if the token has been canceled already.
+         */
+        bool is_canceled() const;
+    private:
+        friend cancellation_token get_cancellation_token(std::weak_ptr<cancellation_token_source> s);
+
+        cancellation_token(std::weak_ptr<cancellation_token_source> parent)
+            : mParent(parent)
+        {
+        }
+
+        std::weak_ptr<cancellation_token_source> mParent;
+    };
+}

include/signalrclient/http_client.h

@@ -8,6 +8,7 @@
 #include <functional>
 #include <map>
 #include <chrono>
+#include "cancellation_token.h"
 #include <exception>
 
 namespace signalr
@@ -62,7 +63,8 @@ namespace signalr
     class http_client
     {
     public:
-        virtual void send(const std::string& url, const http_request& request, std::function<void(const http_response&, std::exception_ptr)> callback) = 0;
+        virtual void send(const std::string& url, http_request& request,
+            std::function<void(const http_response&, std::exception_ptr)> callback, cancellation_token token) = 0;
 
         virtual ~http_client() {}
     };

src/signalrclient/CMakeLists.txt

@@ -1,5 +1,7 @@
 set (SOURCES
   callback_manager.cpp
+  cancellation_token.cpp
+  cancellation_token_source.cpp
   connection.cpp
   connection_impl.cpp
   default_http_client.cpp

src/signalrclient/cancellation_token.cpp

@@ -0,0 +1,33 @@
+// 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.
+
+#include "stdafx.h"
+#include "cancellation_token_source.h"
+#include "signalrclient/cancellation_token.h"
+
+namespace signalr
+{
+    void cancellation_token::register_callback(std::function<void()> callback)
+    {
+        auto obj = mParent.lock();
+        if (obj)
+        {
+            obj->register_callback(callback);
+        }
+        else
+        {
+            callback();
+        }
+    }
+
+    bool cancellation_token::is_canceled() const
+    {
+        auto obj = mParent.lock();
+        if (obj)
+        {
+            return obj->is_canceled();
+        }
+        return true;
+    }
+}

src/signalrclient/cancellation_token_source.cpp

@@ -0,0 +1,16 @@
+// 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.
+
+#include "stdafx.h"
+#include "cancellation_token_source.h"
+
+namespace signalr
+{
+    const unsigned int cancellation_token_source::timeout_infinite = 0xFFFFFFFF;
+
+    cancellation_token get_cancellation_token(std::weak_ptr<cancellation_token_source> s)
+    {
+        return cancellation_token(s);
+    }
+}

src/signalrclient/cancellation_token.h renamed to src/signalrclient/cancellation_token_source.h

@@ -7,57 +7,91 @@
 #include <assert.h>
 #include <condition_variable>
 #include <mutex>
+#include <vector>
+#include "signalrclient/cancellation_token.h"
 
 namespace signalr
 {
     class canceled_exception : public std::exception
     {
+        virtual char const* what() const noexcept
+        {
+            return "an operation was canceled";
+        }
     };
 
-    class cancellation_token
+    class aggregate_exception : public std::exception
     {
     public:
-        static const unsigned int timeout_infinite = 0xFFFFFFFF;
+        void add_exception(const std::exception& ex)
+        {
+            if (!m_message.empty())
+            {
+                m_message += '\n';
+            }
+            m_message += ex.what();
+        }
 
-        cancellation_token() noexcept
-            : m_signaled(false)
+        virtual char const* what() const noexcept
         {
+            return m_message.c_str();
         }
+    private:
+        std::string m_message;
+    };
 
-        cancellation_token(const cancellation_token&) = delete;
-        cancellation_token& operator=(const cancellation_token&) = delete;
+    class cancellation_token_source
+    {
+    public:
+        static const unsigned int timeout_infinite;
 
-        ~cancellation_token()
+        cancellation_token_source() noexcept
+            : m_signaled(false)
         {
-            if (m_callback)
-            {
-                m_callback();
-            }
         }
 
+        cancellation_token_source(const cancellation_token_source&) = delete;
+        cancellation_token_source& operator=(const cancellation_token_source&) = delete;
+
         void cancel()
         {
-            std::function<void()> callback;
+            std::vector<std::function<void()>> callbacks;
             {
                 std::lock_guard<std::mutex> lock(m_lock);
                 m_signaled = true;
                 m_condition.notify_all();
-                callback = m_callback;
-                m_callback = nullptr;
+                callbacks = std::move(m_callbacks);
+                m_callbacks = std::vector<std::function<void()>>();
             } // unlock
 
-            if (callback)
+            if (!callbacks.empty())
             {
-                callback();
+                aggregate_exception errors;
+                for (auto& func : callbacks)
+                {
+                    try
+                    {
+                        func();
+                    }
+                    catch (const std::exception& ex)
+                    {
+                        errors.add_exception(ex);
+                    }
+                }
+
+                if (errors.what()[0] != 0)
+                {
+                    throw errors;
+                }
             }
         }
 
         void reset()
         {
             std::lock_guard<std::mutex> lock(m_lock);
-            assert(m_callback == nullptr);
+            assert(m_callbacks.empty());
             m_signaled = false;
-            m_callback = nullptr;
+            m_callbacks.clear();
         }
 
         bool is_canceled() const
@@ -68,7 +102,7 @@ namespace signalr
         unsigned int wait(unsigned int timeout)
         {
             std::unique_lock<std::mutex> lock(m_lock);
-            if (timeout == cancellation_token::timeout_infinite)
+            if (timeout == cancellation_token_source::timeout_infinite)
             {
                 m_condition.wait(lock, [this]() noexcept { return m_signaled; });
                 return 0;
@@ -79,13 +113,13 @@ namespace signalr
                 const auto status = m_condition.wait_for(lock, period, [this]() noexcept { return m_signaled; });
                 assert(status == m_signaled);
                 // Return 0 if the wait completed as a result of signaling the event. Otherwise, return timeout_infinite
-                return status ? 0 : cancellation_token::timeout_infinite;
+                return status ? 0 : cancellation_token_source::timeout_infinite;
             }
         }
 
         unsigned int wait()
         {
-            return wait(cancellation_token::timeout_infinite);
+            return wait(cancellation_token_source::timeout_infinite);
         }
 
         void throw_if_cancellation_requested() const
@@ -101,14 +135,13 @@ namespace signalr
             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;
+                    m_callbacks.push_back(callback);
                 }
             } // unlock
 
@@ -122,6 +155,8 @@ namespace signalr
         std::mutex m_lock;
         std::condition_variable m_condition;
         bool m_signaled;
-        std::function<void()> m_callback;
+        std::vector<std::function<void()>> m_callbacks;
     };
+
+    cancellation_token get_cancellation_token(std::weak_ptr<cancellation_token_source> s);
 }

src/signalrclient/connection_impl.cpp

@@ -35,7 +35,7 @@ namespace signalr
     connection_impl::connection_impl(const std::string& url, trace_level trace_level, const std::shared_ptr<log_writer>& log_writer,
         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([](std::exception_ptr) noexcept {}), m_disconnect_cts(std::make_shared<cancellation_token>())
+        m_message_received([](const std::string&) noexcept {}), m_disconnected([](std::exception_ptr) noexcept {}), m_disconnect_cts(std::make_shared<cancellation_token_source>())
     {
         if (http_client_factory != nullptr)
         {
@@ -65,7 +65,18 @@ namespace signalr
             // Signaling the event is safe here. We are in the dtor so noone is using this instance. There might be some
             // outstanding threads that hold on to the connection via a weak pointer but they won't be able to acquire
             // the instance since it is being destroyed. Note that the event may actually be in non-signaled state here.
-            m_start_completed_event.cancel();
+            try
+            {
+                m_start_completed_event.cancel();
+            }
+            catch (const std::exception& ex)
+            {
+                if (m_logger.is_enabled(trace_level::warning))
+                {
+                    m_logger.log(trace_level::warning, std::string("start completed event threw an exception in the destructor: ")
+                        .append(ex.what()));
+                }
+            }
             completion_event completion;
             auto logger = m_logger;
             shutdown([completion, logger](std::exception_ptr exception) mutable
@@ -194,7 +205,18 @@ namespace signalr
 
                 connection->m_transport = nullptr;
                 connection->change_state(connection_state::disconnected);
-                connection->m_start_completed_event.cancel();
+                try
+                {
+                    connection->m_start_completed_event.cancel();
+                }
+                catch (const std::exception& ex)
+                {
+                    if (connection->m_logger.is_enabled(trace_level::warning))
+                    {
+                        connection->m_logger.log(trace_level::warning, std::string("start completed event threw an exception in start negotiate: ")
+                            .append(ex.what()));
+                    }
+                }
                 callback(std::current_exception());
                 return;
             }
@@ -213,7 +235,18 @@ namespace signalr
                 assert(false);
             }
 
-            connection->m_start_completed_event.cancel();
+            try
+            {
+                connection->m_start_completed_event.cancel();
+            }
+            catch (const std::exception& ex)
+            {
+                if (connection->m_logger.is_enabled(trace_level::warning))
+                {
+                    connection->m_logger.log(trace_level::warning, std::string("start completed event threw an exception in start negotiate: ")
+                        .append(ex.what()));
+                }
+            }
             callback(nullptr);
         };
 
@@ -328,7 +361,7 @@ namespace signalr
                 }
 
                 connection->start_transport(url, transport_started);
-            });
+            }, get_cancellation_token(m_disconnect_cts));
     }
 
     void connection_impl::start_transport(const std::string& url, std::function<void(std::shared_ptr<transport>, std::exception_ptr)> transport_started)
@@ -552,7 +585,18 @@ namespace signalr
             const auto current_state = get_connection_state();
             if (current_state == connection_state::disconnected)
             {
-                m_disconnect_cts->cancel();
+                try
+                {
+                    m_disconnect_cts->cancel();
+                }
+                catch (const std::exception& ex)
+                {
+                    if (m_logger.is_enabled(trace_level::warning))
+                    {
+                        m_logger.log(trace_level::warning, std::string("disconnect event threw an exception in shutdown: ")
+                            .append(ex.what()));
+                    }
+                }
                 callback(m_stop_error);
                 return;
             }
@@ -567,7 +611,18 @@ namespace signalr
             }
 
             // we request a cancellation of the ongoing start (if any) and wait until it is canceled
-            m_disconnect_cts->cancel();
+            try
+            {
+                m_disconnect_cts->cancel();
+            }
+            catch (const std::exception& ex)
+            {
+                if (m_logger.is_enabled(trace_level::warning))
+                {
+                    m_logger.log(trace_level::warning, std::string("disconnect event threw an exception in shutdown: ")
+                        .append(ex.what()));
+                }
+            }
 
             while (m_start_completed_event.wait(60000) != 0)
             {