Qt scheduler

CMakePresets.json

@@ -224,6 +224,9 @@
     {
       "name": "ci-tests",
       "configuration": "Release",
+      "environment": {
+        "QT_QPA_PLATFORM":"offscreen"
+      },
       "output": {
         "outputOnFailure": true
       },

cmake/dependencies.cmake

@@ -7,10 +7,11 @@ endif()
 
 # ==================== QT ==========================
 if (RPP_BUILD_QT_CODE AND (RPP_BUILD_TESTS OR RPP_BUILD_EXAMPLES))
-  find_package(Qt6 COMPONENTS Widgets)
+  find_package(Qt6 COMPONENTS Widgets QUIET)
   if (Qt6_FOUND)
     SET(RPP_QT_TARGET Qt6)
   else()
+    message("-- RPP: Can't find Qt6, searching for Qt5...")
     find_package(Qt5 REQUIRED COMPONENTS Widgets)
     SET(RPP_QT_TARGET Qt5)
   endif()

src/rpp/rpp/schedulers/immediate_scheduler.hpp

@@ -17,7 +17,6 @@
 
 #include <chrono>
 #include <concepts>
-#include <thread>
 
 namespace rpp::schedulers
 {

src/rppqt/rppqt/fwd.hpp

@@ -21,4 +21,12 @@
 * \ingroup rppqt
 */
 
-#include <rppqt/sources/fwd.hpp>
+#include <rppqt/sources/fwd.hpp>
+
+/**
+ * \defgroup schedulers Schedulers
+ * \brief Scheduler is the way to introduce multi-threading in your application via RPP
+ * \see https://reactivex.io/documentation/scheduler.html
+ * \ingroup rpp
+ */
+#include <rppqt/schedulers/fwd.hpp>

src/rppqt/rppqt/rppqt.hpp

@@ -11,4 +11,5 @@
 #pragma once
 
 #include <rppqt/fwd.hpp>
-#include <rppqt/sources.hpp>
+#include <rppqt/sources.hpp>
+#include <rppqt/schedulers.hpp>

src/rppqt/rppqt/schedulers.hpp

@@ -0,0 +1,20 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2022 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+/**
+ * \defgroup qt_schedulers QT Schedulers
+ * \brief Scheduler is the way to introduce multi-threading in your application via RPP
+ * \see https://reactivex.io/documentation/scheduler.html
+ * \ingroup rppqt
+ */
+
+#include <rppqt/schedulers/main_thread_scheduler.hpp>

src/rppqt/rppqt/schedulers/fwd.hpp

@@ -0,0 +1,16 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2022 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+namespace rppqt::schedulers
+{
+class main_thread_scheduler;
+} // namespace rppqt::schedulers

src/rppqt/rppqt/schedulers/main_thread_scheduler.hpp

@@ -0,0 +1,74 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2022 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+#include <rppqt/schedulers/fwd.hpp>                 // own forwarding
+#include <rpp/schedulers/details/worker.hpp>        // worker
+#include <rpp/subscriptions/subscription_base.hpp>  // lifetime
+#include <rppqt/utils/exceptions.hpp>
+
+#include <chrono>
+#include <concepts>
+
+#include <QCoreApplication>
+#include <QTimer>
+
+namespace rppqt::schedulers
+{
+/**
+ * \brief Schedule provided schedulables to main GUI QT thread (where QApplication placed)
+ * \ingroup qt_schedulers
+ */
+class main_thread_scheduler final : public rpp::schedulers::details::scheduler_tag
+{
+private:
+    class worker_strategy;
+    using main_thread_schedulable = rpp::schedulers::schedulable_wrapper<worker_strategy>;
+
+    class worker_strategy
+    {
+    public:
+        worker_strategy(const rpp::subscription_base& sub)
+            : m_sub{sub} {}
+
+        bool is_subscribed() const { return m_sub.is_subscribed(); }
+
+        void defer_at(rpp::schedulers::time_point time_point, rpp::schedulers::constraint::schedulable_fn auto&& fn) const
+        {
+            defer_at(time_point, main_thread_schedulable{*this, time_point, std::forward<decltype(fn)>(fn)});
+        }
+
+        void defer_at(rpp::schedulers::time_point time_point, main_thread_schedulable&& fn) const
+        {
+            if (!m_sub.is_subscribed())
+                return;
+
+            const auto application = QCoreApplication::instance();
+            if (!application)
+                throw utils::no_active_qapplication{
+                    "Pointer to application is null. Create QApplication before using main_thread_scheduler!"};
+
+            const auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(now() - time_point).count();
+            QTimer::singleShot(duration, application, std::move(fn));
+        }
+
+        static rpp::schedulers::time_point now() { return rpp::schedulers::clock_type::now(); }
+    private:
+        rpp::subscription_base m_sub;
+    };
+
+public:
+    static auto create_worker(const rpp::subscription_base& sub = {})
+    {
+        return rpp::schedulers::worker<worker_strategy>{sub};
+    }
+};
+} // namespace rppqt::schedulers

src/rppqt/rppqt/utils/exceptions.hpp

@@ -0,0 +1,21 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2022 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+#include <stdexcept>
+
+namespace rppqt::utils
+{
+struct no_active_qapplication : std::runtime_error
+{
+    using std::runtime_error::runtime_error;
+};
+}

src/tests/rppqt/test_main_thread_scheduler.cpp

@@ -0,0 +1,66 @@
+//                   ReactivePlusPlus library
+// 
+//           Copyright Aleksey Loginov 2022 - present.
+//  Distributed under the Boost Software License, Version 1.0.
+//     (See accompanying file LICENSE_1_0.txt or copy at
+//           https://www.boost.org/LICENSE_1_0.txt)
+// 
+//  Project home: https://github.com/victimsnino/ReactivePlusPlus
+
+#include <catch2/catch_test_macros.hpp>
+#include <rpp/subscriptions/composite_subscription.hpp>
+
+#include <rppqt/schedulers/main_thread_scheduler.hpp>
+
+#include <future>
+#include <QApplication>
+
+SCENARIO("main_thread_scheduler schedules actions to main thread", "[schedulers]")
+{
+    GIVEN("qapplication and scheduler")
+    {
+        int argc{};
+        QCoreApplication application{argc, nullptr};
+        WHEN("submitting action to main scheduler from another thread")
+        {
+            std::promise<std::thread::id> execution_thread{};
+            std::thread{[&]
+            {
+                rppqt::schedulers::main_thread_scheduler{}.create_worker().schedule([&]()->rpp::schedulers::optional_duration
+                {
+                    execution_thread.set_value(std::this_thread::get_id());
+                    return {};
+                });
+            }}.join();
+            QTimer::singleShot(10, &application, [&]{application.exit();});
+            application.exec();
+            THEN("thread of exectuion of schedulable should be same as thread of application")
+            {
+                auto future = execution_thread.get_future();
+                REQUIRE(future.wait_for(std::chrono::seconds{3}) == std::future_status::ready);
+                CHECK(future.get() == std::this_thread::get_id());
+            }
+        }
+        WHEN("submitting action to main scheduler from another thread but with unsubscribed subscription after schedule")
+        {
+            std::promise<std::thread::id> execution_thread{};
+            std::thread{[&]
+            {
+                rpp::composite_subscription sub{};
+                rppqt::schedulers::main_thread_scheduler{}.create_worker(sub).schedule([&]()->rpp::schedulers::optional_duration
+                {
+                    execution_thread.set_value(std::this_thread::get_id());
+                    return {};
+                });
+                sub.unsubscribe();
+            }}.join();
+            QTimer::singleShot(10, &application, [&]{application.exit();});
+            application.exec();
+            THEN("no schedule execution")
+            {
+                auto future = execution_thread.get_future();
+                REQUIRE(future.wait_for(std::chrono::seconds{1}) == std::future_status::timeout);
+            }
+        }
+    }
+}