rx-observe_on.hpp

Go to the documentation of this file.

// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.

#pragma once

#if !defined(RXCPP_OPERATORS_RX_OBSERVE_ON_HPP)
#define RXCPP_OPERATORS_RX_OBSERVE_ON_HPP

#include "../rx-includes.hpp"

namespace rxcpp {

namespace operators {

namespace detail {

template<class... AN>
struct observe_on_invalid_arguments {};

template<class... AN>
struct observe_on_invalid : public rxo::operator_base<observe_on_invalid_arguments<AN...>> {
    using type = observable<observe_on_invalid_arguments<AN...>, observe_on_invalid<AN...>>;
};
template<class... AN>
using observe_on_invalid_t = typename observe_on_invalid<AN...>::type;

template<class T, class Coordination>
struct observe_on
{
    typedef rxu::decay_t<T> source_value_type;

    typedef rxu::decay_t<Coordination> coordination_type;
    typedef typename coordination_type::coordinator_type coordinator_type;

    coordination_type coordination;

    observe_on(coordination_type cn)
        : coordination(std::move(cn))
    {
    }

    template<class Subscriber>
    struct observe_on_observer
    {
        typedef observe_on_observer<Subscriber> this_type;
        typedef source_value_type value_type;
        typedef rxu::decay_t<Subscriber> dest_type;
        typedef observer<value_type, this_type> observer_type;

        typedef rxn::notification<T> notification_type;
        typedef typename notification_type::type base_notification_type;
        typedef std::deque<base_notification_type> queue_type;

        struct mode
        {
            enum type {
                Invalid = 0,
                Processing,
                Empty,
                Disposed,
                Errored
            };
        };
        struct observe_on_state : std::enable_shared_from_this<observe_on_state>
        {
            mutable std::mutex lock;
            mutable queue_type fill_queue;
            mutable queue_type drain_queue;
            composite_subscription lifetime;
            mutable typename mode::type current;
            coordinator_type coordinator;
            dest_type destination;

            observe_on_state(dest_type d, coordinator_type coor, composite_subscription cs)
                : lifetime(std::move(cs))
                , current(mode::Empty)
                , coordinator(std::move(coor))
                , destination(std::move(d))
            {
            }

            void finish(std::unique_lock<std::mutex>& guard, typename mode::type end) const {
                if (!guard.owns_lock()) {
                    std::terminate();
                }
                if (current == mode::Errored || current == mode::Disposed) {return;}
                current = end;
                queue_type fill_expired;
                swap(fill_expired, fill_queue);
                queue_type drain_expired;
                swap(drain_expired, drain_queue);
                RXCPP_UNWIND_AUTO([&](){guard.lock();});
                guard.unlock();
                lifetime.unsubscribe();
                destination.unsubscribe();
            }

            void ensure_processing(std::unique_lock<std::mutex>& guard) const {
                if (!guard.owns_lock()) {
                    std::terminate();
                }
                if (current == mode::Empty) {
                    current = mode::Processing;

                    if (!lifetime.is_subscribed() && fill_queue.empty() && drain_queue.empty()) {
                        finish(guard, mode::Disposed);
                    }

                    auto keepAlive = this->shared_from_this();

                    auto drain = [keepAlive, this](const rxsc::schedulable& self){
                        using std::swap;
                        try {
                            for (;;) {
                                if (drain_queue.empty() || !destination.is_subscribed()) {
                                    std::unique_lock<std::mutex> guard(lock);
                                    if (!destination.is_subscribed() ||
                                        (!lifetime.is_subscribed() && fill_queue.empty() && drain_queue.empty())) {
                                        finish(guard, mode::Disposed);
                                        return;
                                    }
                                    if (drain_queue.empty()) {
                                        if (fill_queue.empty()) {
                                            current = mode::Empty;
                                            return;
                                        }
                                        swap(fill_queue, drain_queue);
                                    }
                                }
                                auto notification = std::move(drain_queue.front());
                                drain_queue.pop_front();
                                notification->accept(destination);
                                std::unique_lock<std::mutex> guard(lock);
                                self();
                                if (lifetime.is_subscribed()) break;
                            }
                        } catch(...) {
                            destination.on_error(std::current_exception());
                            std::unique_lock<std::mutex> guard(lock);
                            finish(guard, mode::Errored);
                        }
                    };

                    auto selectedDrain = on_exception(
                        [&](){return coordinator.act(drain);},
                        destination);
                    if (selectedDrain.empty()) {
                        finish(guard, mode::Errored);
                        return;
                    }

                    auto processor = coordinator.get_worker();

                    RXCPP_UNWIND_AUTO([&](){guard.lock();});
                    guard.unlock();

                    processor.schedule(selectedDrain.get());
                }
            }
        };
        std::shared_ptr<observe_on_state> state;

        observe_on_observer(dest_type d, coordinator_type coor, composite_subscription cs)
            : state(std::make_shared<observe_on_state>(std::move(d), std::move(coor), std::move(cs)))
        {
        }

        void on_next(source_value_type v) const {
            std::unique_lock<std::mutex> guard(state->lock);
            if (state->current == mode::Errored || state->current == mode::Disposed) { return; }
            state->fill_queue.push_back(notification_type::on_next(std::move(v)));
            state->ensure_processing(guard);
        }
        void on_error(std::exception_ptr e) const {
            std::unique_lock<std::mutex> guard(state->lock);
            if (state->current == mode::Errored || state->current == mode::Disposed) { return; }
            state->fill_queue.push_back(notification_type::on_error(e));
            state->ensure_processing(guard);
        }
        void on_completed() const {
            std::unique_lock<std::mutex> guard(state->lock);
            if (state->current == mode::Errored || state->current == mode::Disposed) { return; }
            state->fill_queue.push_back(notification_type::on_completed());
            state->ensure_processing(guard);
        }

        static subscriber<value_type, observer<value_type, this_type>> make(dest_type d, coordination_type cn, composite_subscription cs = composite_subscription()) {
            auto coor = cn.create_coordinator(d.get_subscription());
            d.add(cs);

            this_type o(d, std::move(coor), cs);
            auto keepAlive = o.state;
            cs.add([=](){
                std::unique_lock<std::mutex> guard(keepAlive->lock);
                keepAlive->ensure_processing(guard);
            });

            return make_subscriber<value_type>(d, cs, make_observer<value_type>(std::move(o)));
        }
    };

    template<class Subscriber>
    auto operator()(Subscriber dest) const
        -> decltype(observe_on_observer<decltype(dest.as_dynamic())>::make(dest.as_dynamic(), coordination)) {
        return      observe_on_observer<decltype(dest.as_dynamic())>::make(dest.as_dynamic(), coordination);
    }
};

}

template<class... AN>
auto observe_on(AN&&... an)
    ->      operator_factory<observe_on_tag, AN...> {
     return operator_factory<observe_on_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
}

}

template<>
struct member_overload<observe_on_tag>
{
    template<class Observable, class Coordination,
        class Enabled = rxu::enable_if_all_true_type_t<
            is_observable<Observable>,
            is_coordination<Coordination>>,
        class SourceValue = rxu::value_type_t<Observable>,
        class ObserveOn = rxo::detail::observe_on<SourceValue, rxu::decay_t<Coordination>>>
    static auto member(Observable&& o, Coordination&& cn)
        -> decltype(o.template lift<SourceValue>(ObserveOn(std::forward<Coordination>(cn)))) {
        return      o.template lift<SourceValue>(ObserveOn(std::forward<Coordination>(cn)));
    }

    template<class... AN>
    static operators::detail::observe_on_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "observe_on takes (Coordination)");
    }
};

class observe_on_one_worker : public coordination_base
{
    rxsc::scheduler factory;

    class input_type
    {
        rxsc::worker controller;
        rxsc::scheduler factory;
        identity_one_worker coordination;
    public:
        explicit input_type(rxsc::worker w)
            : controller(w)
            , factory(rxsc::make_same_worker(w))
            , coordination(factory)
        {
        }
        inline rxsc::worker get_worker() const {
            return controller;
        }
        inline rxsc::scheduler get_scheduler() const {
            return factory;
        }
        inline rxsc::scheduler::clock_type::time_point now() const {
            return factory.now();
        }
        template<class Observable>
        auto in(Observable o) const
            -> decltype(o.observe_on(coordination)) {
            return      o.observe_on(coordination);
        }
        template<class Subscriber>
        auto out(Subscriber s) const
            -> Subscriber {
            return s;
        }
        template<class F>
        auto act(F f) const
            -> F {
            return f;
        }
    };

public:

    explicit observe_on_one_worker(rxsc::scheduler sc) : factory(sc) {}

    typedef coordinator<input_type> coordinator_type;

    inline rxsc::scheduler::clock_type::time_point now() const {
        return factory.now();
    }

    inline coordinator_type create_coordinator(composite_subscription cs = composite_subscription()) const {
        auto w = factory.create_worker(std::move(cs));
        return coordinator_type(input_type(std::move(w)));
    }
};

inline observe_on_one_worker observe_on_run_loop(const rxsc::run_loop& rl) {
    static observe_on_one_worker r(rxsc::make_run_loop(rl));
    return r;
}

inline observe_on_one_worker observe_on_event_loop() {
    static observe_on_one_worker r(rxsc::make_event_loop());
    return r;
}

inline observe_on_one_worker observe_on_new_thread() {
    static observe_on_one_worker r(rxsc::make_new_thread());
    return r;
}

}

#endif