rx-reduce.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_REDUCE_HPP)
#define RXCPP_OPERATORS_RX_REDUCE_HPP

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

namespace rxcpp {

namespace operators {

namespace detail {

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

template<class... AN>
struct reduce_invalid : public rxo::operator_base<reduce_invalid_arguments<AN...>> {
    using type = observable<reduce_invalid_arguments<AN...>, reduce_invalid<AN...>>;
};
template<class... AN>
using reduce_invalid_t = typename reduce_invalid<AN...>::type;

template<class Seed, class ResultSelector>
struct is_result_function_for {

    typedef rxu::decay_t<ResultSelector> result_selector_type;
    typedef rxu::decay_t<Seed> seed_type;

    struct tag_not_valid {};

    template<class CS, class CRS>
    static auto check(int) -> decltype((*(CRS*)nullptr)(*(CS*)nullptr));
    template<class CS, class CRS>
    static tag_not_valid check(...);

    typedef rxu::decay_t<decltype(check<seed_type, result_selector_type>(0))> type;
    static const bool value = !std::is_same<type, tag_not_valid>::value;
};

template<class T, class Observable, class Accumulator, class ResultSelector, class Seed>
struct reduce_traits
{
    typedef rxu::decay_t<Observable> source_type;
    typedef rxu::decay_t<Accumulator> accumulator_type;
    typedef rxu::decay_t<ResultSelector> result_selector_type;
    typedef rxu::decay_t<Seed> seed_type;

    typedef T source_value_type;

    typedef typename is_result_function_for<seed_type, result_selector_type>::type value_type;
};

template<class T, class Observable, class Accumulator, class ResultSelector, class Seed>
struct reduce : public operator_base<rxu::value_type_t<reduce_traits<T, Observable, Accumulator, ResultSelector, Seed>>>
{
    typedef reduce<T, Observable, Accumulator, ResultSelector, Seed> this_type;
    typedef reduce_traits<T, Observable, Accumulator, ResultSelector, Seed> traits;

    typedef typename traits::source_type source_type;
    typedef typename traits::accumulator_type accumulator_type;
    typedef typename traits::result_selector_type result_selector_type;
    typedef typename traits::seed_type seed_type;

    typedef typename traits::source_value_type source_value_type;
    typedef typename traits::value_type value_type;

    struct reduce_initial_type
    {
        ~reduce_initial_type()
        {
        }
        reduce_initial_type(source_type o, accumulator_type a, result_selector_type rs, seed_type s)
            : source(std::move(o))
            , accumulator(std::move(a))
            , result_selector(std::move(rs))
            , seed(std::move(s))
        {
        }
        source_type source;
        accumulator_type accumulator;
        result_selector_type result_selector;
        seed_type seed;

    private:
        reduce_initial_type& operator=(reduce_initial_type o) RXCPP_DELETE;
    };
    reduce_initial_type initial;

    ~reduce()
    {
    }
    reduce(source_type o, accumulator_type a, result_selector_type rs, seed_type s)
        : initial(std::move(o), std::move(a), std::move(rs), std::move(s))
    {
    }
    template<class Subscriber>
    void on_subscribe(Subscriber o) const {
        struct reduce_state_type
            : public reduce_initial_type
            , public std::enable_shared_from_this<reduce_state_type>
        {
            reduce_state_type(reduce_initial_type i, Subscriber scrbr)
                : reduce_initial_type(i)
                , source(i.source)
                , current(reduce_initial_type::seed)
                , out(std::move(scrbr))
            {
            }
            source_type source;
            seed_type current;
            Subscriber out;

        private:
            reduce_state_type& operator=(reduce_state_type o) RXCPP_DELETE;
        };
        auto state = std::make_shared<reduce_state_type>(initial, std::move(o));
        state->source.subscribe(
            state->out,
        // on_next
            [state](T t) {
                seed_type next = state->accumulator(std::move(state->current), std::move(t));
                state->current = std::move(next);
            },
        // on_error
            [state](std::exception_ptr e) {
                state->out.on_error(e);
            },
        // on_completed
            [state]() {
                auto result = on_exception(
                    [&](){return state->result_selector(std::move(state->current));},
                    state->out);
                if (result.empty()) {
                    return;
                }
                state->out.on_next(std::move(result.get()));
                state->out.on_completed();
            }
        );
    }
private:
    reduce& operator=(reduce o) RXCPP_DELETE;
};

template<class T>
struct initialize_seeder {
    typedef T seed_type;
    static seed_type seed() {
        return seed_type{};
    }
};

template<class T>
struct average {
    struct seed_type
    {
        seed_type()
            : value()
            , count(0)
        {
        }
        rxu::maybe<T> value;
        int count;
        rxu::detail::maybe<double> stage;
    };
    static seed_type seed() {
        return seed_type{};
    }
    template<class U>
    seed_type operator()(seed_type a, U&& v) {
        if (a.count != 0 &&
            (a.count == std::numeric_limits<int>::max() ||
            ((v > 0) && (*(a.value) > (std::numeric_limits<T>::max() - v))) ||
            ((v < 0) && (*(a.value) < (std::numeric_limits<T>::min() - v))))) {
            // would overflow, calc existing and reset for next batch
            // this will add error to the final result, but the alternative
            // is to fail on overflow
            double avg = static_cast<double>(*(a.value)) / a.count;
            if (!a.stage.empty()) {
                a.stage.reset((*a.stage + avg) / 2);
            } else {
                a.stage.reset(avg);
            }
            a.value.reset(std::forward<U>(v));
            a.count = 1;
        } else if (a.value.empty()) {
            a.value.reset(std::forward<U>(v));
            a.count = 1;
        } else {
            *(a.value) += v;
            ++a.count;
        }
        return a;
    }
    double operator()(seed_type a) {
        if (!a.value.empty()) {
            double avg = static_cast<double>(*(a.value)) / a.count;
            if (!a.stage.empty()) {
                avg = (*a.stage + avg) / 2;
            }
            return avg;
        }
        throw rxcpp::empty_error("average() requires a stream with at least one value");
    }
};

template<class T>
struct sum {
    typedef rxu::maybe<T> seed_type;
    static seed_type seed() {
        return seed_type();
    }
    template<class U>
    seed_type operator()(seed_type a, U&& v) const {
        if (a.empty())
            a.reset(std::forward<U>(v));
        else
            *a = *a + v;
        return a;
    }
    T operator()(seed_type a) const {
        if (a.empty())
            throw rxcpp::empty_error("sum() requires a stream with at least one value");
        return *a;
    }
};

template<class T>
struct max {
    typedef rxu::maybe<T> seed_type;
    static seed_type seed() {
        return seed_type();
    }
    template<class U>
    seed_type operator()(seed_type a, U&& v) {
        if (a.empty() || *a < v)
            a.reset(std::forward<U>(v));
        return a;
    }
    T operator()(seed_type a) {
        if (a.empty())
            throw rxcpp::empty_error("max() requires a stream with at least one value");
        return *a;
    }
};

template<class T>
struct min {
    typedef rxu::maybe<T> seed_type;
    static seed_type seed() {
        return seed_type();
    }
    template<class U>
    seed_type operator()(seed_type a, U&& v) {
        if (a.empty() || v < *a)
            a.reset(std::forward<U>(v));
        return a;
    }
    T operator()(seed_type a) {
        if (a.empty())
            throw rxcpp::empty_error("min() requires a stream with at least one value");
        return *a;
    }
};

template<class T>
struct first {
    using seed_type = rxu::maybe<T>;
    static seed_type seed() {
        return seed_type();
    }
    template<class U>
    seed_type operator()(seed_type a, U&& v) {
        a.reset(std::forward<U>(v));
        return a;
    }
    T operator()(seed_type a) {
        if (a.empty()) {
            throw rxcpp::empty_error("first() requires a stream with at least one value");
        }
        return *a;
    }
};

template<class T>
struct last {
    using seed_type = rxu::maybe<T>;
    static seed_type seed() {
        return seed_type();
    }
    template<class U>
    seed_type operator()(seed_type a, U&& v) {
        a.reset(std::forward<U>(v));
        return a;
    }
    T operator()(seed_type a) {
        if (a.empty()) {
            throw rxcpp::empty_error("last() requires a stream with at least one value");
        }
        return *a;
    }
};

}

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

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

inline auto first()
    ->     operator_factory<first_tag> {
    return operator_factory<first_tag>(std::tuple<>{});
}

inline auto last()
    ->     operator_factory<last_tag> {
    return operator_factory<last_tag>(std::tuple<>{});
}

inline auto count()
    ->     operator_factory<reduce_tag, int, rxu::count, rxu::detail::take_at<0>> {
    return operator_factory<reduce_tag, int, rxu::count, rxu::detail::take_at<0>>(std::make_tuple(0, rxu::count(), rxu::take_at<0>()));
}

inline auto average()
    ->     operator_factory<average_tag> {
    return operator_factory<average_tag>(std::tuple<>{});
}

inline auto sum()
    ->     operator_factory<sum_tag> {
    return operator_factory<sum_tag>(std::tuple<>{});
}

inline auto min()
    ->     operator_factory<min_tag> {
    return operator_factory<min_tag>(std::tuple<>{});
}

inline auto max()
    ->     operator_factory<max_tag> {
    return operator_factory<max_tag>(std::tuple<>{});
}

}

template<>
struct member_overload<reduce_tag>
{

    template<class Observable, class Seed, class Accumulator, class ResultSelector,
        class Reduce = rxo::detail::reduce<rxu::value_type_t<Observable>, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class Value = rxu::value_type_t<Reduce>,
        class Result = observable<Value, Reduce>>
    static Result member(Observable&& o, Seed&& s, Accumulator&& a, ResultSelector&& r)
    {
        return Result(Reduce(std::forward<Observable>(o), std::forward<Accumulator>(a), std::forward<ResultSelector>(r), std::forward<Seed>(s)));
    }

    template<class Observable, class Seed, class Accumulator,
        class ResultSelector=rxu::detail::take_at<0>,
        class Reduce = rxo::detail::reduce<rxu::value_type_t<Observable>, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class Value = rxu::value_type_t<Reduce>,
        class Result = observable<Value, Reduce>>
    static Result member(Observable&& o, Seed&& s, Accumulator&& a)
    {
        return Result(Reduce(std::forward<Observable>(o), std::forward<Accumulator>(a), rxu::detail::take_at<0>(), std::forward<Seed>(s)));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "reduce takes (Seed, Accumulator, optional ResultSelector), Accumulator takes (Seed, Observable::value_type) -> Seed, ResultSelector takes (Observable::value_type) -> ResultValue");
    }
};

template<>
struct member_overload<first_tag>
{
    template<class Observable,
        class SValue = rxu::value_type_t<Observable>,
        class Operation = operators::detail::first<SValue>,
        class Seed = decltype(Operation::seed()),
        class Accumulator = Operation,
        class ResultSelector = Operation,
        class TakeOne = decltype(((rxu::decay_t<Observable>*)nullptr)->take(1)),
        class Reduce = rxo::detail::reduce<SValue, rxu::decay_t<TakeOne>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class RValue = rxu::value_type_t<Reduce>,
        class Result = observable<RValue, Reduce>>
    static Result member(Observable&& o)
    {
        return Result(Reduce(o.take(1), Operation{}, Operation{}, Operation::seed()));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "first does not support Observable::value_type");
    }
};

template<>
struct member_overload<last_tag>
{
    template<class Observable,
        class SValue = rxu::value_type_t<Observable>,
        class Operation = operators::detail::last<SValue>,
        class Seed = decltype(Operation::seed()),
        class Accumulator = Operation,
        class ResultSelector = Operation,
        class Reduce = rxo::detail::reduce<SValue, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class RValue = rxu::value_type_t<Reduce>,
        class Result = observable<RValue, Reduce>>
    static Result member(Observable&& o)
    {
        return Result(Reduce(std::forward<Observable>(o), Operation{}, Operation{}, Operation::seed()));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "last does not support Observable::value_type");
    }
};

template<>
struct member_overload<sum_tag>
{
    template<class Observable,
        class SValue = rxu::value_type_t<Observable>,
        class Operation = operators::detail::sum<SValue>,
        class Seed = decltype(Operation::seed()),
        class Accumulator = Operation,
        class ResultSelector = Operation,
        class Reduce = rxo::detail::reduce<SValue, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class RValue = rxu::value_type_t<Reduce>,
        class Result = observable<RValue, Reduce>>
    static Result member(Observable&& o)
    {
        return Result(Reduce(std::forward<Observable>(o), Operation{}, Operation{}, Operation::seed()));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "sum does not support Observable::value_type");
    }
};

template<>
struct member_overload<average_tag>
{
    template<class Observable,
        class SValue = rxu::value_type_t<Observable>,
        class Operation = operators::detail::average<SValue>,
        class Seed = decltype(Operation::seed()),
        class Accumulator = Operation,
        class ResultSelector = Operation,
        class Reduce = rxo::detail::reduce<SValue, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class RValue = rxu::value_type_t<Reduce>,
        class Result = observable<RValue, Reduce>>
    static Result member(Observable&& o)
    {
        return Result(Reduce(std::forward<Observable>(o), Operation{}, Operation{}, Operation::seed()));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "average does not support Observable::value_type");
    }
};

template<>
struct member_overload<max_tag>
{
    template<class Observable,
        class SValue = rxu::value_type_t<Observable>,
        class Operation = operators::detail::max<SValue>,
        class Seed = decltype(Operation::seed()),
        class Accumulator = Operation,
        class ResultSelector = Operation,
        class Reduce = rxo::detail::reduce<SValue, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class RValue = rxu::value_type_t<Reduce>,
        class Result = observable<RValue, Reduce>>
    static Result member(Observable&& o)
    {
        return Result(Reduce(std::forward<Observable>(o), Operation{}, Operation{}, Operation::seed()));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "max does not support Observable::value_type");
    }
};

template<>
struct member_overload<min_tag>
{
    template<class Observable,
        class SValue = rxu::value_type_t<Observable>,
        class Operation = operators::detail::min<SValue>,
        class Seed = decltype(Operation::seed()),
        class Accumulator = Operation,
        class ResultSelector = Operation,
        class Reduce = rxo::detail::reduce<SValue, rxu::decay_t<Observable>, rxu::decay_t<Accumulator>, rxu::decay_t<ResultSelector>, rxu::decay_t<Seed>>,
        class RValue = rxu::value_type_t<Reduce>,
        class Result = observable<RValue, Reduce>>
    static Result member(Observable&& o)
    {
        return Result(Reduce(std::forward<Observable>(o), Operation{}, Operation{}, Operation::seed()));
    }

    template<class... AN>
    static operators::detail::reduce_invalid_t<AN...> member(AN...) {
        std::terminate();
        return {};
        static_assert(sizeof...(AN) == 10000, "min does not support Observable::value_type");
    }
};

}

#endif