<commit_before>// 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 typename std::decay<T>::type source_value_type;

    typedef typename std::decay<Coordination>::type 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 : public observer_base<source_value_type>
    {
        typedef observe_on_observer<Subscriber> this_type;
        typedef observer_base<source_value_type> base_type;
        typedef typename base_type::value_type value_type;
        typedef typename std::decay<Subscriber>::type 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
            };
        };
        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;
            rxsc::worker processor;
            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){
                        try {
                            if (drain_queue.empty() || !destination.is_subscribed()) {
                                std::unique_lock<std::mutex> guard(lock);
                                if (!destination.is_subscribed() ||
                                    (!lifetime.is_subscribed() && queue.empty())) {
                                    current = mode::Disposed;
                                    using std::swap;
                                    queue_type expired;
                                    swap(expired, queue);
                                    guard.unlock();
                                    lifetime.unsubscribe();
                                    destination.unsubscribe();
                                    return;
                                }
                                if (queue.empty()) {
                                    current = mode::Empty;
                                    return;
                                }
                                using std::swap;
                                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::Empty;
                        }
                    };

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

                    auto processor = coordinator.get_worker();
                    processor.schedule(lifetime, 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 {
            if (state->lifetime.is_subscribed()) {
                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 {
            if (state->lifetime.is_subscribed()) {
                std::unique_lock<std::mutex> guard(state->lock);
                state->queue.push(notification_type::on_error(e));
                state->ensure_processing(guard);
            }
        }
        void on_completed() const {
            if (state->lifetime.is_subscribed()) {
                std::unique_lock<std::mutex> guard(state->lock);
                state->queue.push(notification_type::on_completed());
                state->ensure_processing(guard);
            }
        }

        static subscriber<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>(cs, 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);
    }<commit_msg>fix deadlock, error and dispose<commit_after>// 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 typename std::decay<T>::type source_value_type;

    typedef typename std::decay<Coordination>::type 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 : public observer_base<source_value_type>
    {
        typedef observe_on_observer<Subscriber> this_type;
        typedef observer_base<source_value_type> base_type;
        typedef typename base_type::value_type value_type;
        typedef typename std::decay<Subscriber>::type 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;
            rxsc::worker processor;
            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()) {
                        std::unique_lock<std::mutex> guard(lock);
                        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, 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);
                if (keepAlive->queue.empty() && keepAlive->current == mode::Empty && keepAlive->destination.is_subscribed()) {
                    keepAlive->current = mode::Disposed;
                    keepAlive->destination.unsubscribe();
                }
            });

            return make_subscriber<value_type>(cs, 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);
    }
{'fraction_non_alphanumeric': 0.09792870226856418, 'fraction_numerical': 0.0001409046075806679, 'mean_word_length': 1.5262502224595125, 'pattern_counts': {'":': 0, '<': 56, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7578396', 'n_tokens_mistral': 3597, 'n_tokens_neox': 3396, 'n_words': 602}