diff --git a/docs/Implementation Status.md b/docs/Implementation Status.md
index d96a6f08f..af1e3f46d 100644
--- a/docs/Implementation Status.md	
+++ b/docs/Implementation Status.md	
@@ -106,13 +106,13 @@
 
 ### Aggregate
 
-- [ ] average
+- [x] average
 - [x] concat
-- [ ] count
-- [ ] max 
-- [ ] min
-- [ ] reduce
-- [ ] sum
+- [x] count
+- [x] max 
+- [x] min
+- [x] reduce
+- [x] sum
 
 ### Backpressure
 
diff --git a/src/examples/rpp/doxygen/reduce.cpp b/src/examples/rpp/doxygen/reduce.cpp
new file mode 100644
index 000000000..db3e9dcce
--- /dev/null
+++ b/src/examples/rpp/doxygen/reduce.cpp
@@ -0,0 +1,70 @@
+#include <rpp/rpp.hpp>
+
+#include <iostream>
+
+/**
+ * \example reduce.cpp
+ **/
+
+int main()
+{
+    //! [reduce]
+    rpp::source::just(1,2,3)
+            .reduce(0, std::plus<int>{})
+            .subscribe([](int v) { std::cout << v << std::endl; });
+    // Output: 6
+    //! [reduce]
+
+     //! [reduce_vector]
+    rpp::source::just(1,2,3)
+            .reduce(std::vector<int>{}, [](std::vector<int>&& seed, int new_value)
+            {
+                seed.push_back(new_value);
+                return std::move(seed);
+            })
+            .subscribe([](const std::vector<int>& v)
+            {
+                std::cout << "vector: ";
+                for(int val : v)
+                    std::cout << val << " ";
+                std::cout << std::endl;
+            });
+    // Output: vector: 1 2 3
+    //! [reduce_vector]
+
+    //! [average]
+    rpp::source::just(1,2,3)
+            .average()
+            .subscribe([](int v) { std::cout << v << std::endl; });
+    // Output: 2
+    //! [average]
+
+    //! [sum]
+    rpp::source::just(1,2,3)
+            .sum()
+            .subscribe([](int v) { std::cout << v << std::endl; });
+    // Output: 6
+    //! [sum]
+
+    //! [count]
+    rpp::source::just(1,2,3)
+            .count()
+            .subscribe([](int v) { std::cout << v << std::endl; });
+    // Output: 3
+    //! [count]
+
+    //! [min]
+    rpp::source::just(5,1,2,3)
+            .min()
+            .subscribe([](int v) { std::cout << v << std::endl; });
+    // Output: 1
+    //! [min]
+
+    //! [max]
+    rpp::source::just(5,1,2,3)
+            .max()
+            .subscribe([](int v) { std::cout << v << std::endl; });
+    // Output: 5
+    //! [max]
+    return 0;
+}
diff --git a/src/rpp/rpp/observables/interface_observable.hpp b/src/rpp/rpp/observables/interface_observable.hpp
index 1cebbeb72..bd5ec0dff 100644
--- a/src/rpp/rpp/observables/interface_observable.hpp
+++ b/src/rpp/rpp/observables/interface_observable.hpp
@@ -54,6 +54,7 @@ struct RPP_EMPTY_BASES interface_observable
     , details::member_overload<Type, SpecificObservable, details::observe_on_tag>
     , details::member_overload<Type, SpecificObservable, details::on_error_resume_next_tag>
     , details::member_overload<Type, SpecificObservable, details::publish_tag>
+    , details::member_overload<Type, SpecificObservable, details::reduce_tag>
     , details::member_overload<Type, SpecificObservable, details::repeat_tag>
     , details::member_overload<Type, SpecificObservable, details::sample_tag>
     , details::member_overload<Type, SpecificObservable, details::scan_tag>
diff --git a/src/rpp/rpp/operators.hpp b/src/rpp/rpp/operators.hpp
index da700d740..1479add18 100644
--- a/src/rpp/rpp/operators.hpp
+++ b/src/rpp/rpp/operators.hpp
@@ -80,6 +80,7 @@
 */
 
 #include <rpp/operators/concat.hpp>
+#include <rpp/operators/reduce.hpp>
 
 /**
  * \defgroup utility_operators Utility Operators
diff --git a/src/rpp/rpp/operators/fwd.hpp b/src/rpp/rpp/operators/fwd.hpp
index 0c72e24bb..556bd6d55 100644
--- a/src/rpp/rpp/operators/fwd.hpp
+++ b/src/rpp/rpp/operators/fwd.hpp
@@ -29,6 +29,7 @@
 #include <rpp/operators/fwd/observe_on.hpp>
 #include <rpp/operators/fwd/on_error_resume_next.hpp>
 #include <rpp/operators/fwd/publish.hpp>
+#include <rpp/operators/fwd/reduce.hpp>
 #include <rpp/operators/fwd/ref_count.hpp>
 #include <rpp/operators/fwd/repeat.hpp>
 #include <rpp/operators/fwd/sample.hpp>
diff --git a/src/rpp/rpp/operators/fwd/reduce.hpp b/src/rpp/rpp/operators/fwd/reduce.hpp
new file mode 100644
index 000000000..fdbd0dc60
--- /dev/null
+++ b/src/rpp/rpp/operators/fwd/reduce.hpp
@@ -0,0 +1,263 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2022 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+#include <rpp/observables/details/member_overload.hpp>
+
+namespace rpp::details
+{
+struct reduce_tag;
+}
+
+namespace rpp::details
+{
+// accepts Result and Type and returns Result
+template<typename Fn, typename Result, typename Type>
+concept reduce_accumulator = std::is_invocable_r_v<std::decay_t<Result>, Fn, std::decay_t<Result>, std::decay_t<Type>>;
+
+template<typename T>
+concept is_can_be_summed = requires(T t)
+{
+    { t + t } -> std::convertible_to<T>;
+};
+
+template<typename T, typename CastBeforeDrop>
+concept is_can_be_averaged = is_can_be_summed<T> && requires(CastBeforeDrop nt)
+{
+    { nt / size_t{} };
+};
+
+template<constraint::decayed_type Type, constraint::decayed_type Seed, reduce_accumulator<Seed, Type> AccumulatorFn, std::invocable<Seed&&> ResultSelectorFn>
+struct reduce_impl;
+
+template<constraint::decayed_type CastBeforeDivide, constraint::observable TObs>
+auto average_impl(TObs&& observable);
+
+template<constraint::observable TObs>
+auto sum_impl(TObs&& observable);
+
+template<constraint::observable TObs>
+auto count_impl(TObs&& observable);
+
+template<constraint::observable TObs, typename Comparator>
+auto min_impl(TObs&& observable, Comparator&& comparator);
+
+template<constraint::observable TObs, typename Comparator>
+auto max_impl(TObs&& observable, Comparator&& comparator);
+
+template<constraint::decayed_type Type, typename SpecificObservable>
+struct member_overload<Type, SpecificObservable, reduce_tag>
+{
+    /**
+     * \brief Applies accumulator function to each emission from observable and result of accumulator from previous step and emits final value
+     * 
+     * \marble reduce
+        {
+            source observable                  : +--1-2-3-|
+            operator "reduce: s=1, (s,x)=>s+x" : +--------7|
+        }
+     *
+     * \param initial_seed initial value for seed which will be applied for first value from observable. Then it will be replaced with result and etc. 
+     * \param accumulator function which accepts seed value and new value from observable and return new value of seed. Can accept seed by move-reference.
+     *
+     * \return new specific_observable with the reduce operator as most recent operator.
+     * \warning #include <rpp/operators/reduce.hpp>
+     * 
+     * \par Example
+     * \snippet reduce.cpp reduce
+     * \snippet reduce.cpp reduce_vector
+	 *
+     * \ingroup transforming_operators
+     * \see https://reactivex.io/documentation/operators/reduce.html
+     */
+    template<typename Seed, reduce_accumulator<Seed, Type> AccumulatorFn, std::invocable<Seed&&> ResultSelectorFn = std::identity>
+    auto reduce(Seed&& initial_seed, AccumulatorFn&& accumulator, ResultSelectorFn&& result_selector = {}) const & requires is_header_included<reduce_tag, Seed, AccumulatorFn, ResultSelectorFn>
+    {
+        return static_cast<const SpecificObservable*>(this)->template lift<utils::decayed_invoke_result_t<ResultSelectorFn, std::decay_t<Seed>>>(
+            reduce_impl<Type, std::decay_t<Seed>, std::decay_t<AccumulatorFn>, std::decay_t<ResultSelectorFn>>{
+                                         std::forward<Seed>(initial_seed),
+                                         std::forward<AccumulatorFn>(accumulator),
+                                         std::forward<ResultSelectorFn>(result_selector)});
+    }
+
+    template<typename Seed, reduce_accumulator<Seed, Type> AccumulatorFn, std::invocable<Seed&&> ResultSelectorFn = std::identity>
+    auto reduce(Seed&& initial_seed, AccumulatorFn&& accumulator, ResultSelectorFn&& result_selector = {}) && requires is_header_included<reduce_tag, Seed, AccumulatorFn, ResultSelectorFn>
+    {
+        return std::move(*static_cast<SpecificObservable*>(this)).template lift<utils::decayed_invoke_result_t<ResultSelectorFn, std::decay_t<Seed>>>(
+            reduce_impl<Type, std::decay_t<Seed>, std::decay_t<AccumulatorFn>, std::decay_t<ResultSelectorFn>>{
+                                         std::forward<Seed>(initial_seed),
+                                         std::forward<AccumulatorFn>(accumulator),
+                                         std::forward<ResultSelectorFn>(result_selector)});
+    }
+
+    /**
+     * \brief Calculates the average of emissions and emits final value
+     * 
+     * \marble average
+        {
+            source observable  : +--1-2-3-|
+            operator "average" : +--------2|
+        }
+     *
+     * \tparam CastBeforeDivide cast accumulated value to this type before division
+     * \return new specific_observable with the average operator as most recent operator.
+     * \throws rpp::utils::not_enough_emissions in case of no any emissions from original observable
+     *
+     * \warning #include <rpp/operators/reduce.hpp>
+     * 
+     * \par Example
+     * \snippet reduce.cpp average
+	 *
+     * \ingroup transforming_operators
+     * \see https://reactivex.io/documentation/operators/average.html
+     */
+    template<typename CastBeforeDivide = Type, typename ...Args>
+    auto average() const & requires (is_header_included<reduce_tag, CastBeforeDivide, Args...> && is_can_be_averaged<Type, CastBeforeDivide>)
+    {
+        return average_impl<CastBeforeDivide>(*static_cast<const SpecificObservable*>(this));
+    }
+
+    template<typename CastBeforeDivide = Type, typename ...Args>
+    auto average() && requires (is_header_included<reduce_tag, CastBeforeDivide, Args...> && is_can_be_averaged<Type, CastBeforeDivide>)
+    {
+        return average_impl<CastBeforeDivide>(std::move(*static_cast<SpecificObservable*>(this)));
+    }
+
+    
+    /**
+     * \brief Calculates the sum of emissions and emits final value
+     * 
+     * \marble sum
+        {
+            source observable  : +--1-2-3-|
+            operator "sum"     : +--------6|
+        }
+     *
+     * \return new specific_observable with the sum operator as most recent operator.
+     * \throws rpp::utils::not_enough_emissions in case of no any emissions from original observable
+     *
+     * \warning #include <rpp/operators/reduce.hpp>
+     * 
+     * \par Example
+     * \snippet reduce.cpp sum
+	 *
+     * \ingroup transforming_operators
+     * \see https://reactivex.io/documentation/operators/sum.html
+     */
+    template<typename ...Args>
+    auto sum() const & requires (is_header_included<reduce_tag, Args...> && is_can_be_summed<Type>)
+    {
+        return sum_impl(*static_cast<const SpecificObservable*>(this));
+    }
+
+    template<typename ...Args>
+    auto sum() && requires (is_header_included<reduce_tag, Args...> && is_can_be_summed<Type>)
+    {
+        return sum_impl(std::move(*static_cast<SpecificObservable*>(this)));
+    }
+
+    /**
+     * \brief Calculates the amount of emitted emissions and emits this count
+     * 
+     * \marble count
+        {
+            source observable  : +--1-2-3-|
+            operator "count"   : +--------3|
+        }
+     *
+     * \return new specific_observable with the count operator as most recent operator.
+     * \warning #include <rpp/operators/reduce.hpp>
+     * 
+     * \par Example
+     * \snippet reduce.cpp count
+	 *
+     * \ingroup transforming_operators
+     * \see https://reactivex.io/documentation/operators/count.html
+     */
+    template<typename ...Args>
+    auto count() const & requires is_header_included<reduce_tag, Args...>
+    {
+        return count_impl(*static_cast<const SpecificObservable*>(this));
+    }
+
+    template<typename ...Args>
+    auto count() && requires is_header_included<reduce_tag, Args...>
+    {
+        return count_impl(std::move(*static_cast<SpecificObservable*>(this)));
+    }
+
+    /**
+     * \brief Emits the emission which has minimal value from the whole observable
+     * 
+     * \marble min
+        {
+            source observable  : +-6-1-2-3-|
+            operator "min"     : +---------1|
+        }
+     *
+     * \param comparator is function to deduce if left value is less than right
+     * \return new specific_observable with the min operator as most recent operator.
+     * \throws rpp::utils::not_enough_emissions in case of no any emissions from original observable
+     *
+     * \warning #include <rpp/operators/reduce.hpp>
+     * 
+     * \par Example
+     * \snippet reduce.cpp min
+	 *
+     * \ingroup transforming_operators
+     * \see https://reactivex.io/documentation/operators/min.html
+     */
+    template<std::strict_weak_order<Type, Type> Comparator = std::less<Type>, typename ...Args>
+    auto min(Comparator&& comparator = {}) const & requires is_header_included<reduce_tag, Comparator, Args...>
+    {
+        return min_impl(*static_cast<const SpecificObservable*>(this), std::forward<Comparator>(comparator));
+    }
+
+    template<std::strict_weak_order<Type, Type> Comparator = std::less<Type>, typename ...Args>
+    auto min(Comparator&& comparator = {}) && requires is_header_included<reduce_tag, Comparator, Args...>
+    {
+        return min_impl(std::move(*static_cast<SpecificObservable*>(this)), std::forward<Comparator>(comparator));
+    }
+
+    /**
+     * \brief Emits the emission which has maximal value from the whole observable
+     * 
+     * \marble max
+        {
+            source observable  : +-6-1-2-3-|
+            operator "max"     : +---------6|
+        }
+     *
+     * \param comparator is function to deduce if left value is less than right
+     * \return new specific_observable with the max operator as most recent operator.
+     * \throws rpp::utils::not_enough_emissions in case of no any emissions from original observable
+     *
+     * \warning #include <rpp/operators/reduce.hpp>
+     * 
+     * \par Example
+     * \snippet reduce.cpp max
+	 *
+     * \ingroup transforming_operators
+     * \see https://reactivex.io/documentation/operators/max.html
+     */
+    template<std::strict_weak_order<Type, Type> Comparator = std::less<Type>, typename ...Args>
+    auto max(Comparator&& comparator = {}) const & requires is_header_included<reduce_tag, Comparator, Args...>
+    {
+        return max_impl(*static_cast<const SpecificObservable*>(this), std::forward<Comparator>(comparator));
+    }
+
+    template<std::strict_weak_order<Type, Type> Comparator = std::less<Type>, typename ...Args>
+    auto max(Comparator&& comparator = {}) && requires is_header_included<reduce_tag, Comparator, Args...>
+    {
+        return max_impl(std::move(*static_cast<SpecificObservable*>(this)), std::forward<Comparator>(comparator));
+    }
+};
+} // namespace rpp::details
diff --git a/src/rpp/rpp/operators/fwd/scan.hpp b/src/rpp/rpp/operators/fwd/scan.hpp
index 270b38600..c6fff707e 100644
--- a/src/rpp/rpp/operators/fwd/scan.hpp
+++ b/src/rpp/rpp/operators/fwd/scan.hpp
@@ -21,7 +21,7 @@ namespace rpp::details
 {
 // accepts Result and Type and returns Result
 template<typename Fn, typename Result, typename Type>
-concept scan_accumulator = std::is_invocable_r_v<std::decay_t<Result>, Fn, std::decay_t<Result>, std::decay_t<Type>>;
+concept scan_accumulator = reduce_accumulator<Fn,Result,Type>;
 
 template<constraint::decayed_type Type, constraint::decayed_type Result, scan_accumulator<Result, Type> AccumulatorFn>
 struct scan_impl;
diff --git a/src/rpp/rpp/operators/reduce.hpp b/src/rpp/rpp/operators/reduce.hpp
new file mode 100644
index 000000000..8a636315d
--- /dev/null
+++ b/src/rpp/rpp/operators/reduce.hpp
@@ -0,0 +1,178 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2022 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+#include <rpp/defs.hpp>                                    // RPP_NO_UNIQUE_ADDRESS
+#include <rpp/observables/constraints.hpp>                 
+#include <rpp/operators/lift.hpp>                          // required due to operator uses lift
+#include <rpp/operators/details/subscriber_with_state.hpp> // create_subscriber_with_state
+#include <rpp/operators/fwd/reduce.hpp>                      // own forwarding
+#include <rpp/subscribers/constraints.hpp>                 // constraint::subscriber
+#include <rpp/utils/functors.hpp>                          // forwarding_on_error
+#include <rpp/utils/utilities.hpp>                         // utils::as_const
+
+
+IMPLEMENTATION_FILE(reduce_tag);
+
+namespace rpp::details
+{
+template<constraint::decayed_type Seed, typename AccumulatorFn, std::invocable<Seed&&> SelectorFn = std::identity>
+struct reduce_state
+{
+    mutable Seed                        seed;
+    RPP_NO_UNIQUE_ADDRESS AccumulatorFn accumulator;
+    RPP_NO_UNIQUE_ADDRESS SelectorFn    selector{};
+};
+
+struct reduce_on_next
+{
+    template<constraint::decayed_type Result, typename AccumulatorFn, typename SelectorFn>
+    void operator()(auto&&                                   value,
+                    const constraint::subscriber auto&,
+                    const reduce_state<Result, AccumulatorFn, SelectorFn>& state) const
+    {
+        state.seed = state.accumulator(std::move(state.seed), std::forward<decltype(value)>(value));
+    }
+};
+
+struct reduce_on_completed
+{
+    template<constraint::decayed_type Result, typename AccumulatorFn, typename SelectorFn>
+    void operator()(const constraint::subscriber auto&                     sub,
+                    const reduce_state<Result, AccumulatorFn, SelectorFn>& state) const
+    {
+        try
+        {
+            sub.on_next(state.selector(std::move(state.seed)));
+        }
+        catch (...)
+        {
+            sub.on_error(std::current_exception());
+            return;
+        }
+        sub.on_completed();
+    }
+};
+
+template<constraint::decayed_type Type, constraint::decayed_type Seed, reduce_accumulator<Seed, Type> AccumulatorFn, std::invocable<Seed&&> ResultSelectorFn>
+struct reduce_impl
+{
+    Seed                                   initial_value;
+    RPP_NO_UNIQUE_ADDRESS AccumulatorFn    accumulator;
+    RPP_NO_UNIQUE_ADDRESS ResultSelectorFn selector;
+
+    template<constraint::subscriber_of_type<utils::decayed_invoke_result_t<ResultSelectorFn, Seed>> TSub>
+    auto operator()(TSub&& subscriber) const
+    {
+        auto subscription = subscriber.get_subscription();
+        // dynamic_state there to make shared_ptr for observer instead of making shared_ptr for state
+        return create_subscriber_with_dynamic_state<Type>(std::move(subscription),
+                                                          reduce_on_next{},
+                                                          utils::forwarding_on_error{},
+                                                          reduce_on_completed{},
+                                                          std::forward<TSub>(subscriber),
+                                                          reduce_state<Seed, AccumulatorFn, ResultSelectorFn>{initial_value, accumulator, selector});
+    }
+};
+
+template<constraint::decayed_type CastBeforeDivide, constraint::observable TObs>
+auto average_impl(TObs&& observable)
+{
+    using Type = utils::extract_observable_type_t<std::decay_t<TObs>>;
+    using Pair = std::pair<std::optional<Type>, int32_t>;
+    return std::forward<TObs>(observable).reduce(Pair{},
+                                                 [](Pair&& seed, auto&& val)
+                                                 {
+                                                     if (seed.first)
+                                                        seed.first.value() += std::forward<decltype(val)>(val);
+                                                     else
+                                                         seed.first = std::forward<decltype(val)>(val);
+                                                     ++seed.second;
+                                                     return std::move(seed);
+                                                 },
+                                                 [](Pair&& seed)
+                                                 {
+                                                     if (!seed.first)
+                                                        throw utils::not_enough_emissions{"`average` operator requires at least one emission to calculate average"};
+
+                                                     return static_cast<CastBeforeDivide>(std::move(seed.first).value()) / seed.second;
+                                                 });
+}
+
+template<constraint::observable TObs>
+auto sum_impl(TObs&& observable)
+{
+    using Type = utils::extract_observable_type_t<std::decay_t<TObs>>;
+    return std::forward<TObs>(observable).reduce(std::optional<Type>{},
+                                                 [](std::optional<Type>&& seed, auto&& val)
+                                                 {
+                                                     if (!seed)
+                                                         seed = std::forward<decltype(val)>(val);
+                                                     else
+                                                        seed.value() += std::forward<decltype(val)>(val);
+                                                     return std::move(seed);
+                                                 },
+                                                 [](std::optional<Type>&& seed)
+                                                 {
+                                                     if (!seed)
+                                                         throw utils::not_enough_emissions{"`sum` operator requires at least one emission to calculate sum"};
+
+                                                     return std::move(seed.value());
+                                                 });
+}
+
+template<constraint::observable TObs>
+auto count_impl(TObs&& observable)
+{
+    return std::forward<TObs>(observable).reduce(size_t{}, [](size_t seed, auto&&) { return ++seed; });
+}
+
+template<constraint::observable TObs, typename Comparator>
+auto min_impl(TObs&& observable, Comparator&& comparator)
+{
+    using Type = utils::extract_observable_type_t<std::decay_t<TObs>>;
+    return std::forward<TObs>(observable).reduce(std::optional<Type>{},
+                                                 [comparator](std::optional<Type>&& seed, auto&& val)
+                                                 {
+                                                     if (!seed || comparator(utils::as_const(val), seed.value()))
+                                                         seed = std::forward<decltype(val)>(val);
+                                                     return std::move(seed);
+                                                 },
+                                                 [](std::optional<Type>&& seed)
+                                                 {
+                                                     if (!seed)
+                                                         throw utils::not_enough_emissions{"`min` operator requires at least one emission to calculate min"};
+
+                                                     return std::move(seed.value());
+                                                 });
+}
+
+template<constraint::observable TObs, typename Comparator>
+auto max_impl(TObs&& observable, Comparator&& comparator)
+{
+    using Type = utils::extract_observable_type_t<std::decay_t<TObs>>;
+    return std::forward<TObs>(observable).reduce(std::optional<Type>{},
+                                                 [comparator](std::optional<Type>&& seed, auto&& val)
+                                                 {
+                                                     if (!seed || comparator(seed.value(), utils::as_const(val)))
+                                                         seed = std::forward<decltype(val)>(val);
+                                                     return std::move(seed);
+                                                 },
+                                                 [](std::optional<Type>&& seed)
+                                                 {
+                                                     if (!seed)
+                                                         throw utils::not_enough_emissions{"`max` operator requires at least one emission to calculate min"};
+
+                                                     return std::move(seed.value());
+                                                 });
+}
+} // namespace rpp::details
+
diff --git a/src/rpp/rpp/operators/scan.hpp b/src/rpp/rpp/operators/scan.hpp
index a9d8e25e2..e24d6b325 100644
--- a/src/rpp/rpp/operators/scan.hpp
+++ b/src/rpp/rpp/operators/scan.hpp
@@ -14,6 +14,7 @@
 #include <rpp/operators/lift.hpp>                          // required due to operator uses lift
 #include <rpp/operators/details/subscriber_with_state.hpp> // create_subscriber_with_state
 #include <rpp/operators/fwd/scan.hpp>                      // own forwarding
+#include <rpp/operators/reduce.hpp>                        // reduce to re-use
 #include <rpp/subscribers/constraints.hpp>                 // constraint::subscriber
 #include <rpp/utils/functors.hpp>                          // forwarding_on_error
 #include <rpp/utils/utilities.hpp>                         // utils::as_const
@@ -23,21 +24,14 @@ IMPLEMENTATION_FILE(scan_tag);
 
 namespace rpp::details
 {
-template<constraint::decayed_type Result, typename AccumulatorFn>
-struct scan_state
-{
-    mutable Result                      seed;
-    RPP_NO_UNIQUE_ADDRESS AccumulatorFn accumulator;
-};
-
-struct scan_on_next
+struct scan_on_next : private reduce_on_next
 {
     template<constraint::decayed_type Result, typename AccumulatorFn>
     void operator()(auto&&                                   value,
                     const constraint::subscriber auto&       sub,
-                    const scan_state<Result, AccumulatorFn>& state) const
+                    const reduce_state<Result, AccumulatorFn>& state) const
     {
-        state.seed = state.accumulator(std::move(state.seed), std::forward<decltype(value)>(value));
+        reduce_on_next::operator()(std::forward<decltype(value)>(value), sub, state);
         sub.on_next(utils::as_const(state.seed));
     }
 };
@@ -58,7 +52,7 @@ struct scan_impl
                                                           utils::forwarding_on_error{},
                                                           utils::forwarding_on_completed{},
                                                           std::forward<TSub>(subscriber),
-                                                          scan_state<Result, AccumulatorFn>{initial_value, accumulator});
+                                                          reduce_state<Result, AccumulatorFn>{initial_value, accumulator});
     }
 };
 } // namespace rpp::details
diff --git a/src/tests/rpp/test_reduce.cpp b/src/tests/rpp/test_reduce.cpp
new file mode 100644
index 000000000..7e357ff54
--- /dev/null
+++ b/src/tests/rpp/test_reduce.cpp
@@ -0,0 +1,337 @@
+//                   ReactivePlusPlus library
+// 
+//           Copyright Aleksey Loginov 2022 - present.
+//  Distributed under the Boost Software License, Version 1.0.
+//     (See accompanying file LICENSE_1_0.txt or copy at
+//           https://www.boost.org/LICENSE_1_0.txt)
+// 
+//  Project home: https://github.com/victimsnino/ReactivePlusPlus
+
+#include "copy_count_tracker.hpp"
+#include "mock_observer.hpp"
+
+#include <catch2/catch_test_macros.hpp>
+
+#include <rpp/operators/reduce.hpp>
+
+#include <rpp/sources/just.hpp>
+#include <rpp/sources/empty.hpp>
+
+SCENARIO("reduce reduces values and store state", "[reduce]")
+{
+    GIVEN("observable")
+    {
+        auto obs = rpp::observable::just(1, 2, 3);
+        WHEN("subscribe on it via reduce with plus")
+        {
+            auto mock = mock_observer<int>{};
+
+            obs.reduce(0, std::plus<int>{}).subscribe(mock);
+            THEN("observer obtains final sum")
+            {
+                CHECK(mock.get_received_values() == std::vector{6});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+        WHEN("subscribe on it via reduce with aggregating in vector")
+        {
+            auto mock = mock_observer<std::vector<int>>{};
+
+            obs.reduce(std::vector<int>{},
+                     [](std::vector<int>&& seed, int new_val)
+                     {
+                         seed.push_back(new_val);
+                         return std::move(seed);
+                     }).subscribe(mock);
+
+            THEN("observer obtains final vector")
+            {
+                CHECK(mock.get_received_values() == std::vector{{std::vector{1,2,3}}});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+        WHEN("subscribe on it via reduce with exception")
+        {
+            auto mock = mock_observer<int>{};
+
+            volatile bool none{};
+            obs.reduce(0,
+                     [&](int, int)-> int
+                     {
+                         if (none)
+                             return 0;
+                         throw std::runtime_error{""};
+                     }).subscribe(mock);
+
+            THEN("observer obtains only on_error")
+            {
+                CHECK(mock.get_total_on_next_count() == 0);
+                CHECK(mock.get_on_error_count() == 1);
+                CHECK(mock.get_on_completed_count() == 0);
+            }
+        }
+    }
+    GIVEN("observable 1-2->")
+    {
+        auto obs = rpp::observable::create<int>([](const auto& sub)
+        {
+            sub.on_next(1);
+            sub.on_next(2);
+        });
+        WHEN("subscribe on it via reduce with plus")
+        {
+            auto mock = mock_observer<int>{};
+
+            obs.reduce(0, std::plus<int>{}).subscribe(mock);
+            THEN("no any values")
+            {
+                CHECK(mock.get_received_values() == std::vector<int>{});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 0);
+            }
+        }
+    }
+}
+
+SCENARIO("average calculates average", "[reduce]")
+{
+    GIVEN("-1-2| observable")
+    {
+        auto obs = rpp::source::just(1,2);
+        WHEN("subscribe on it via average")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.average().subscribe(mock);
+
+            THEN("observer obtained value as average int")
+            {
+                CHECK(mock.get_received_values() == std::vector{static_cast<int>(1+2)/2});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+        WHEN("subscribe on it via average<double>")
+        {
+            auto mock = mock_observer<double>{};
+            obs.average<double>().subscribe(mock);
+
+            THEN("observer obtained value as average double")
+            {
+                CHECK(mock.get_received_values() == std::vector<double>{1.5});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    GIVEN("-| observable")
+    {
+        auto obs = rpp::source::empty<int>();
+        WHEN("subscribe on it via average")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.average().subscribe(mock);
+
+            THEN("observer obtained on_error")
+            {
+                CHECK(mock.get_received_values() == std::vector<int>{});
+                CHECK(mock.get_on_error_count() == 1);
+                CHECK(mock.get_on_completed_count() == 0);
+            }
+        }
+    }
+}
+
+SCENARIO("sum calculates sum", "[reduce]")
+{
+    GIVEN("-1-2| observable")
+    {
+        auto obs = rpp::source::just(1,2);
+        WHEN("subscribe on it via sum")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.sum().subscribe(mock);
+
+            THEN("observer obtained value as sum int")
+            {
+                CHECK(mock.get_received_values() == std::vector{3});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    GIVEN("-| observable")
+    {
+        auto obs = rpp::source::empty<int>();
+        WHEN("subscribe on it via sum")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.sum().subscribe(mock);
+
+            THEN("observer obtained on_error")
+            {
+                CHECK(mock.get_received_values() == std::vector<int>{});
+                CHECK(mock.get_on_error_count() == 1);
+                CHECK(mock.get_on_completed_count() == 0);
+            }
+        }
+    }
+}
+
+SCENARIO("count calculates count", "[reduce]")
+{
+    GIVEN("-1-2| observable")
+    {
+        auto obs = rpp::source::just(1,2);
+        WHEN("subscribe on it via count")
+        {
+            auto mock = mock_observer<size_t>{};
+            auto r= obs.count().subscribe(mock);
+
+            THEN("observer obtained value as count size_t")
+            {
+                CHECK(mock.get_received_values() == std::vector{size_t{2}});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    GIVEN("-| observable")
+    {
+        auto obs = rpp::source::empty<int>();
+        WHEN("subscribe on it via count")
+        {
+            auto mock = mock_observer<size_t>{};
+            auto r= obs.count().subscribe(mock);
+
+            THEN("observer obtained zero")
+            {
+                CHECK(mock.get_received_values() == std::vector<size_t>{0});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+}
+
+SCENARIO("min calculates min", "[reduce]")
+{
+    GIVEN("-3-1-2| observable")
+    {
+        auto obs = rpp::source::just(3,1,2);
+        WHEN("subscribe on it via min")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.min().subscribe(mock);
+
+            THEN("observer obtained min value")
+            {
+                CHECK(mock.get_received_values() == std::vector{1});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    GIVEN("-| observable")
+    {
+        auto obs = rpp::source::empty<int>();
+        WHEN("subscribe on it via min")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.min().subscribe(mock);
+
+            THEN("observer obtained on_error")
+            {
+                CHECK(mock.get_received_values() == std::vector<int>{});
+                CHECK(mock.get_on_error_count() == 1);
+                CHECK(mock.get_on_completed_count() == 0);
+            }
+        }
+    }
+}
+
+SCENARIO("max calculates min", "[reduce]")
+{
+    GIVEN("-3-1-2| observable")
+    {
+        auto obs = rpp::source::just(3,1,2);
+        WHEN("subscribe on it via max")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.max().subscribe(mock);
+
+            THEN("observer obtained max value")
+            {
+                CHECK(mock.get_received_values() == std::vector{3});
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    GIVEN("-| observable")
+    {
+        auto obs = rpp::source::empty<int>();
+        WHEN("subscribe on it via max")
+        {
+            auto mock = mock_observer<int>{};
+            auto r= obs.max().subscribe(mock);
+
+            THEN("observer obtained on_error")
+            {
+                CHECK(mock.get_received_values() == std::vector<int>{});
+                CHECK(mock.get_on_error_count() == 1);
+                CHECK(mock.get_on_completed_count() == 0);
+            }
+        }
+    }
+}
+
+SCENARIO("reduce keeps state for copies", "[reduce]")
+{
+    auto mock = mock_observer<int>{};
+    GIVEN("observable which sends values via copy")
+    {
+        auto obs = rpp::source::create<int>([](const auto& sub)
+        {
+            for (size_t i = 0; i < 10; ++i)
+            {
+                auto copy = sub;
+                copy.on_next(1);
+            }
+            sub.on_completed();
+        });
+        WHEN("subscribe on it via reduce")
+        {
+            obs.reduce(int{}, [](int seed, int new_v) { return seed + new_v; }).subscribe(mock);
+            THEN("observer obtains only last value")
+            {
+                CHECK(mock.get_received_values() == std::vector{ 10 });
+            }
+        }
+    }
+}
+
+SCENARIO("reduce doesn't produce extra copies", "[reduce][track_copy]")
+{
+    GIVEN("observable and subscriber")
+    {
+        copy_count_tracker verifier{};
+        auto               obs = rpp::source::just(1).reduce(verifier, [](copy_count_tracker&& seed, int) { return std::move(seed); });
+        WHEN("subscribe")
+        {
+            obs.subscribe([](const auto&) {});
+            THEN("no extra copies")
+            {
+                REQUIRE(verifier.get_copy_count() == 2); // 1 copy to reduce state + 1 copy for provided subscriber to shared_state
+                REQUIRE(verifier.get_move_count() == 5);
+                // 1 move to observable state + 1 move to subscriber + 1 move from lambda + 1 move to new_state + 1 move to final lambda
+            }
+        }
+    }
+}