add run-loop scheduler

Author: Kirk Shoop

Rx/v2/src/rxcpp/operators/rx-repeat.hpp

@@ -53,12 +53,16 @@ struct repeat : public operator_base<T>
             }
             composite_subscription source_lifetime;
             output_type out;
+            composite_subscription::weak_subscription lifetime_token;
 
             void do_subscribe() {
                 auto state = this->shared_from_this();
+                
+                state->out.remove(state->lifetime_token);
+                state->source_lifetime.unsubscribe();
 
                 state->source_lifetime = composite_subscription();
-                state->out.add(state->source_lifetime);
+                state->lifetime_token = state->out.add(state->source_lifetime);
 
                 state->source.subscribe(
                     state->out,

Rx/v2/src/rxcpp/schedulers/rx-runloop.hpp

@@ -0,0 +1,202 @@
+// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
+
+#pragma once
+
+#if !defined(RXCPP_RX_SCHEDULER_RUN_LOOP_HPP)
+#define RXCPP_RX_SCHEDULER_RUN_LOOP_HPP
+
+#include "../rx-includes.hpp"
+
+namespace rxcpp {
+
+namespace schedulers {
+
+namespace detail {
+
+struct run_loop_state : public std::enable_shared_from_this<run_loop_state>
+{
+    typedef scheduler::clock_type clock_type;
+
+    typedef detail::schedulable_queue<
+        clock_type::time_point> queue_item_time;
+
+    typedef queue_item_time::item_type item_type;
+    typedef queue_item_time::const_reference const_reference_item_type;
+
+    virtual ~run_loop_state()
+    {
+    }
+
+    run_loop_state()
+    {
+    }
+
+    composite_subscription lifetime;
+    mutable std::mutex lock;
+    mutable queue_item_time q;
+    recursion r;
+};
+
+}
+
+
+struct run_loop_scheduler : public scheduler_interface
+{
+private:
+    typedef run_loop_scheduler this_type;
+    run_loop_scheduler(const this_type&);
+
+    struct run_loop_worker : public worker_interface
+    {
+    private:
+        typedef run_loop_worker this_type;
+
+        run_loop_worker(const this_type&);
+
+    public:
+        std::weak_ptr<detail::run_loop_state> state;
+
+        virtual ~run_loop_worker()
+        {
+        }
+
+        explicit run_loop_worker(std::weak_ptr<detail::run_loop_state> ws)
+            : state(ws)
+        {
+        }
+
+        virtual clock_type::time_point now() const {
+            return clock_type::now();
+        }
+
+        virtual void schedule(const schedulable& scbl) const {
+            schedule(now(), scbl);
+        }
+
+        virtual void schedule(clock_type::time_point when, const schedulable& scbl) const {
+            if (scbl.is_subscribed()) {
+                auto st = state.lock();
+                std::unique_lock<std::mutex> guard(st->lock);
+                st->q.push(detail::run_loop_state::item_type(when, scbl));
+                st->r.reset(false);
+            }
+        }
+    };
+
+    std::weak_ptr<detail::run_loop_state> state;
+
+public:
+    explicit run_loop_scheduler(std::weak_ptr<detail::run_loop_state> ws)
+        : state(ws)
+    {
+    }
+    virtual ~run_loop_scheduler()
+    {
+    }
+
+    virtual clock_type::time_point now() const {
+        return clock_type::now();
+    }
+
+    virtual worker create_worker(composite_subscription cs) const {
+        auto lifetime = state.lock()->lifetime;
+        auto token = lifetime.add(cs);
+        cs.add([=](){lifetime.remove(token);});
+        return worker(cs, create_worker_interface());
+    }
+
+    std::shared_ptr<worker_interface> create_worker_interface() const {
+        return std::make_shared<run_loop_worker>(state);
+    }
+};
+
+class run_loop
+{
+private:
+    typedef run_loop this_type;
+    // don't allow this instance to copy/move since it owns current_thread queue
+    // for the thread it is constructed on.
+    run_loop(const this_type&);
+    run_loop(this_type&&);
+
+    typedef scheduler::clock_type clock_type;
+    typedef detail::action_queue queue_type;
+
+    typedef detail::run_loop_state::item_type item_type;
+    typedef detail::run_loop_state::const_reference_item_type const_reference_item_type;
+
+    std::shared_ptr<detail::run_loop_state> state;
+    std::shared_ptr<run_loop_scheduler> sc;
+
+public:
+    run_loop()
+        : state(std::make_shared<detail::run_loop_state>())
+        , sc(std::make_shared<run_loop_scheduler>(state))
+    {
+        // take ownership so that the current_thread scheduler
+        // uses the same queue on this thread
+        queue_type::ensure(sc->create_worker_interface());
+    }
+    ~run_loop()
+    {
+        state->lifetime.unsubscribe();
+
+        std::unique_lock<std::mutex> guard(state->lock);
+
+        // release ownership
+        queue_type::destroy();
+
+        auto expired = std::move(state->q);
+        if (!state->q.empty()) abort();
+    }
+
+    clock_type::time_point now() const {
+        return clock_type::now();
+    }
+    
+    composite_subscription get_subscription() const {
+        return state->lifetime;
+    }
+    
+    bool empty() const {
+        return state->q.empty();
+    }
+
+    const_reference_item_type peek() const {
+        return state->q.top();
+    }
+
+    void dispatch() const {
+        std::unique_lock<std::mutex> guard(state->lock);
+        if (state->q.empty()) {
+            return;
+        }
+        auto& peek = state->q.top();
+        if (!peek.what.is_subscribed()) {
+            state->q.pop();
+            return;
+        }
+        if (clock_type::now() < peek.when) {
+            return;
+        }
+        auto what = peek.what;
+        state->q.pop();
+        state->r.reset(state->q.empty());
+        guard.unlock();
+        what(state->r.get_recurse());
+    }
+
+    scheduler get_scheduler() const {
+        return make_scheduler(sc);
+    }
+};
+
+inline scheduler make_run_loop(const run_loop& r) {
+    return r.get_scheduler();
+}
+
+}
+
+}
+
+#endif

projects/CMake/CMakeLists.txt

@@ -85,6 +85,7 @@ set(RX_SOURCES
    ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-eventloop.hpp
    ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-immediate.hpp
    ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-newthread.hpp
+   ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-runloop.hpp
    ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-sameworker.hpp
    ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-test.hpp
    ${RXCPP_DIR}/Rx/v2/src/rxcpp/schedulers/rx-virtualtime.hpp
@@ -107,4 +108,4 @@ set(RX_SOURCES
 source_group("src" FILES ${RX_SOURCES})
 
 add_library(RxCpp SHARED ${RX_SOURCES})
-SET_TARGET_PROPERTIES(RxCpp PROPERTIES LINKER_LANGUAGE CXX)
+SET_TARGET_PROPERTIES(RxCpp PROPERTIES LINKER_LANGUAGE CXX)