path: root/Rx/v2/src/rxcpp/operators/rx-observe_on.hpp

// 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 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 observer_base<source_value_type> base_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::queue<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 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 ensure_processing(std::unique_lock<std::mutex>& guard) const {
                if (!guard.owns_lock()) {
                    abort();
                }
                if (current == mode::Empty) {
                    current = mode::Processing;

                    auto keepAlive = this->shared_from_this();

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

                    auto selectedDrain = on_exception(
                        [&](){return coordinator.act(drain);},
                        destination);
                    if (selectedDrain.empty()) {
                        current = mode::Errored;
                        using std::swap;
                        queue_type expired;
                        swap(expired, queue);
                        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);
            state->queue.push(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);
            state->queue.push(notification_type::on_error(e));
            state->ensure_processing(guard);
        }
        void on_completed() const {
            std::unique_lock<std::mutex> guard(state->lock);
            state->queue.push(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([keepAlive](){
                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 Coordination>
class observe_on_factory
{
    typedef rxu::decay_t<Coordination> coordination_type;
    coordination_type coordination;
public:
    observe_on_factory(coordination_type cn) : coordination(std::move(cn)) {}
    template<class Observable>
    auto operator()(Observable&& source)
        -> decltype(source.template lift<rxu::value_type_t<rxu::decay_t<Observable>>>(observe_on<rxu::value_type_t<rxu::decay_t<Observable>>, coordination_type>(coordination))) {
        return      source.template lift<rxu::value_type_t<rxu::decay_t<Observable>>>(observe_on<rxu::value_type_t<rxu::decay_t<Observable>>, coordination_type>(coordination));
    }
};

}

template<class Coordination>
auto observe_on(Coordination cn)
    ->      detail::observe_on_factory<Coordination> {
    return  detail::observe_on_factory<Coordination>(std::move(cn));
}


}

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_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