From 5f0640aa34d002dc701c819e5a677c90ea806926 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Wed, 4 Sep 2024 22:00:53 +0300
Subject: [PATCH 01/10] try to fix

---
 src/tests/rppgrpc/test_async_server.cpp | 9 ++++++---
 src/tests/utils/rpp_trompeloil.hpp      | 2 +-
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/src/tests/rppgrpc/test_async_server.cpp b/src/tests/rppgrpc/test_async_server.cpp
index 7d139c6be..9bfc97668 100644
--- a/src/tests/rppgrpc/test_async_server.cpp
+++ b/src/tests/rppgrpc/test_async_server.cpp
@@ -55,7 +55,8 @@ TEST_CASE("Async server")
             if constexpr (requires { reactor->get_observer(); })
                 reactor->get_observer().on_completed();
 
-            CHECK(writer->Finish().ok());
+            const auto res = writer->Finish().ok();
+            CHECK(res);
             wait(last);
         }
 
@@ -92,7 +93,8 @@ TEST_CASE("Async server")
             if constexpr (requires { reactor->get_observer(); })
                 reactor->get_observer().on_completed();
 
-            REQUIRE(writer->Finish().ok());
+            const auto res = writer->Finish().ok();
+            REQUIRE(res);
 
             wait(last);
         }
@@ -119,7 +121,8 @@ TEST_CASE("Async server")
                 REQUIRE(response.value() == i);
             }
             REQUIRE(!writer->Read(&response));
-            REQUIRE(writer->Finish().ok());
+            const auto res = writer->Finish().ok();
+            REQUIRE(res);
             wait(last);
         }
     };
diff --git a/src/tests/utils/rpp_trompeloil.hpp b/src/tests/utils/rpp_trompeloil.hpp
index 124e15f10..e8508899c 100644
--- a/src/tests/utils/rpp_trompeloil.hpp
+++ b/src/tests/utils/rpp_trompeloil.hpp
@@ -30,7 +30,7 @@ namespace trompeloeil
     inline void reporter<specialized>::sendOk(
         const char* trompeloeil_mock_calls_done_correctly)
     {
-        REQUIRE(trompeloeil_mock_calls_done_correctly != 0);
+        REQUIRE(trompeloeil_mock_calls_done_correctly);
     }
 } // namespace trompeloeil
 

From 76a535c08b16ef54abb638fce1c2ce7124042108 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Wed, 4 Sep 2024 23:16:53 +0300
Subject: [PATCH 02/10] improve simple operators

---
 src/rpp/rpp/observables/connectable_observable.hpp | 8 ++++----
 src/rpp/rpp/observables/fwd.hpp                    | 7 -------
 src/rpp/rpp/observables/observable.hpp             | 8 ++++----
 3 files changed, 8 insertions(+), 15 deletions(-)

diff --git a/src/rpp/rpp/observables/connectable_observable.hpp b/src/rpp/rpp/observables/connectable_observable.hpp
index 567afeebe..218522894 100644
--- a/src/rpp/rpp/observables/connectable_observable.hpp
+++ b/src/rpp/rpp/observables/connectable_observable.hpp
@@ -130,14 +130,14 @@ namespace rpp
                                    details::ref_count_on_subscribe_t<connectable_observable<OriginalObservable, Subject>>>{*this};
         }
 
-        template<rpp::constraint::operator_observable_transform<const connectable_observable&> Op>
-        auto operator|(Op&& op) const &
+        template<typename Op>
+        rpp::constraint::observable auto operator|(Op&& op) const &
         {
             return std::forward<Op>(op)(*this);
         }
 
-        template<rpp::constraint::operator_observable_transform<connectable_observable&&> Op>
-        auto operator|(Op&& op)&&
+        template<typename Op>
+        rpp::constraint::observable auto operator|(Op&& op)&&
         {
             return std::forward<Op>(op)(std::move(*this));
         }
diff --git a/src/rpp/rpp/observables/fwd.hpp b/src/rpp/rpp/observables/fwd.hpp
index 8b4eabe15..5a450aa0b 100644
--- a/src/rpp/rpp/observables/fwd.hpp
+++ b/src/rpp/rpp/observables/fwd.hpp
@@ -79,13 +79,6 @@ namespace rpp::constraint
     template<typename T, typename Type>
     concept observable_of_type = observable<T> && std::same_as<utils::extract_observable_type_t<T>, std::decay_t<Type>>;
 
-    template<typename Op, typename TObs>
-    concept operator_observable_transform = requires(const Op& op, TObs obs) {
-        {
-            op(static_cast<TObs>(obs))
-        } -> rpp::constraint::observable;
-    };
-
     template<typename Op, typename Type>
     concept operator_base = requires(const Op& op) { typename std::decay_t<Op>::template operator_traits<Type>; } && details::observables::constraint::disposable_strategy<details::observables::deduce_updated_disposable_strategy<std::decay_t<Op>, typename details::observables::chain<details::observables::fake_strategy<Type>>::expected_disposable_strategy>>;
 
diff --git a/src/rpp/rpp/observables/observable.hpp b/src/rpp/rpp/observables/observable.hpp
index 6d3d6ca29..5932e6b3f 100644
--- a/src/rpp/rpp/observables/observable.hpp
+++ b/src/rpp/rpp/observables/observable.hpp
@@ -324,14 +324,14 @@ namespace rpp
             return std::move(*this).inner_make_chain_operator(std::forward<Op>(op));
         }
 
-        template<constraint::operator_observable_transform<const observable&> Op>
-        auto operator|(Op&& op) const &
+        template<typename Op>
+        rpp::constraint::observable auto operator|(Op&& op) const &
         {
             return std::forward<Op>(op)(*this);
         }
 
-        template<constraint::operator_observable_transform<observable&&> Op>
-        auto operator|(Op&& op) &&
+        template<typename Op>
+        rpp::constraint::observable auto operator|(Op&& op) &&
         {
             return std::forward<Op>(op)(std::move(*this));
         }

From 3b16d9bac8127b293cbd243eca9b3fce6bac9181 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 6 Sep 2024 00:01:08 +0300
Subject: [PATCH 03/10] a bit better

---
 src/examples/rpp/doxygen/map.cpp              |  1 +
 .../observables/connectable_observable.hpp    |  4 +-
 .../observables/details/chain_strategy.hpp    |  2 +
 src/rpp/rpp/observables/fwd.hpp               | 28 +++++------
 src/rpp/rpp/observables/observable.hpp        | 49 ++++++++-----------
 src/rpp/rpp/operators/flat_map.hpp            |  6 +--
 src/rpp/rpp/operators/multicast.hpp           |  5 +-
 src/rpp/rpp/operators/start_with.hpp          |  2 +-
 src/rpp/rpp/operators/subscribe.hpp           | 16 +++---
 9 files changed, 52 insertions(+), 61 deletions(-)

diff --git a/src/examples/rpp/doxygen/map.cpp b/src/examples/rpp/doxygen/map.cpp
index 4e22e1332..01d869ea3 100644
--- a/src/examples/rpp/doxygen/map.cpp
+++ b/src/examples/rpp/doxygen/map.cpp
@@ -11,6 +11,7 @@ int main() // NOLINT(bugprone-exception-escape)
     //! [Same type]
     rpp::source::just(42)
         | rpp::operators::map([](int value) { return value + 10; })
+        | rpp::ops::start_with(std::string{})
         | rpp::operators::subscribe([](int v) { std::cout << v << std::endl; });
     // Output: 52
     //! [Same type]
diff --git a/src/rpp/rpp/observables/connectable_observable.hpp b/src/rpp/rpp/observables/connectable_observable.hpp
index 218522894..1326ca050 100644
--- a/src/rpp/rpp/observables/connectable_observable.hpp
+++ b/src/rpp/rpp/observables/connectable_observable.hpp
@@ -130,13 +130,13 @@ namespace rpp
                                    details::ref_count_on_subscribe_t<connectable_observable<OriginalObservable, Subject>>>{*this};
         }
 
-        template<typename Op>
+        template<std::invocable<const connectable_observable&> Op>
         rpp::constraint::observable auto operator|(Op&& op) const &
         {
             return std::forward<Op>(op)(*this);
         }
 
-        template<typename Op>
+        template<std::invocable<connectable_observable&&> Op>
         rpp::constraint::observable auto operator|(Op&& op)&&
         {
             return std::forward<Op>(op)(std::move(*this));
diff --git a/src/rpp/rpp/observables/details/chain_strategy.hpp b/src/rpp/rpp/observables/details/chain_strategy.hpp
index bf6eb3b6b..0d2d1948f 100644
--- a/src/rpp/rpp/observables/details/chain_strategy.hpp
+++ b/src/rpp/rpp/observables/details/chain_strategy.hpp
@@ -24,6 +24,8 @@ namespace rpp::details::observables
 
         using operator_traits = typename TStrategy::template operator_traits<typename base::value_type>;
 
+        static_assert(rpp::constraint::operator_chain<TStrategy, typename base::value_type, typename base::expected_disposable_strategy>);
+
     public:
         using expected_disposable_strategy = deduce_updated_disposable_strategy<TStrategy, typename base::expected_disposable_strategy>;
         using value_type                   = typename operator_traits::result_type;
diff --git a/src/rpp/rpp/observables/fwd.hpp b/src/rpp/rpp/observables/fwd.hpp
index 5a450aa0b..6cd897baf 100644
--- a/src/rpp/rpp/observables/fwd.hpp
+++ b/src/rpp/rpp/observables/fwd.hpp
@@ -79,37 +79,40 @@ namespace rpp::constraint
     template<typename T, typename Type>
     concept observable_of_type = observable<T> && std::same_as<utils::extract_observable_type_t<T>, std::decay_t<Type>>;
 
-    template<typename Op, typename Type>
-    concept operator_base = requires(const Op& op) { typename std::decay_t<Op>::template operator_traits<Type>; } && details::observables::constraint::disposable_strategy<details::observables::deduce_updated_disposable_strategy<std::decay_t<Op>, typename details::observables::chain<details::observables::fake_strategy<Type>>::expected_disposable_strategy>>;
+    template<typename TObservable, typename... TObservables>
+    concept observables_of_same_type = rpp::constraint::observable<TObservable> && (rpp::constraint::observable<TObservables> && ...) && (std::same_as<rpp::utils::extract_observable_type_t<TObservable>, rpp::utils::extract_observable_type_t<TObservables>> && ...);
 
     template<typename Op, typename Type>
-    concept operator_subscribe = operator_base<Op, Type> && requires(const Op& op, rpp::details::observers::fake_observer<typename std::decay_t<Op>::template operator_traits<Type>::result_type>&& observer, const details::observables::chain<details::observables::fake_strategy<Type>>& chain) {
+    concept operator_subscribe = requires(const Op& op, rpp::details::observers::fake_observer<typename std::decay_t<Op>::template operator_traits<Type>::result_type>&& observer, const details::observables::chain<details::observables::fake_strategy<Type>>& chain) {
         {
             op.subscribe(std::move(observer), chain)
         };
     };
 
     template<typename Op, typename Type>
-    concept operator_lift = operator_base<Op, Type> && requires(const Op& op, rpp::details::observers::fake_observer<typename std::decay_t<Op>::template operator_traits<Type>::result_type>&& observer) {
+    concept operator_lift = requires(const Op& op, rpp::details::observers::fake_observer<typename std::decay_t<Op>::template operator_traits<Type>::result_type>&& observer) {
         {
             op.template lift<Type>(std::move(observer))
         } -> rpp::constraint::observer_of_type<Type>;
     };
 
     template<typename Op, typename Type, typename DisposableStrategy>
-    concept operator_lift_with_disposable_strategy = operator_base<Op, Type> && requires(const Op& op, rpp::details::observers::fake_observer<typename std::decay_t<Op>::template operator_traits<Type>::result_type>&& observer) {
+    concept operator_lift_with_disposable_strategy = requires(const Op& op, rpp::details::observers::fake_observer<typename std::decay_t<Op>::template operator_traits<Type>::result_type>&& observer) {
         {
             op.template lift_with_disposable_strategy<Type, DisposableStrategy>(std::move(observer))
         } -> rpp::constraint::observer_of_type<Type>;
     };
 
     template<typename Op, typename Type, typename DisposableStrategy>
-    concept operator_chain = operator_base<std::decay_t<Op>, Type>
-                          && requires { typename std::decay_t<Op>::template operator_traits<Type>::result_type; }
-                          && (operator_subscribe<std::decay_t<Op>, Type> || operator_lift<std::decay_t<Op>, Type> || operator_lift_with_disposable_strategy<std::decay_t<Op>, Type, DisposableStrategy>);
+    concept operator_chain =
+        requires() {
+            typename std::decay_t<Op>::template operator_traits<Type>;
+            typename std::decay_t<Op>::template operator_traits<Type>::result_type;
+        }
+        && details::observables::constraint::disposable_strategy<details::observables::deduce_updated_disposable_strategy<std::decay_t<Op>,
+                                                                                                                          typename details::observables::chain<details::observables::fake_strategy<Type>>::expected_disposable_strategy>>
+        && (operator_subscribe<std::decay_t<Op>, Type> || operator_lift<std::decay_t<Op>, Type> || operator_lift_with_disposable_strategy<std::decay_t<Op>, Type, DisposableStrategy>);
 
-    template<typename TObservable, typename... TObservables>
-    concept observables_of_same_type = rpp::constraint::observable<TObservable> && (rpp::constraint::observable<TObservables> && ...) && (std::same_as<rpp::utils::extract_observable_type_t<TObservable>, rpp::utils::extract_observable_type_t<TObservables>> && ...);
 } // namespace rpp::constraint
 
 namespace rpp
@@ -126,8 +129,3 @@ namespace rpp
     template<constraint::decayed_type Type, rpp::constraint::observable_of_type<Type>... Observables>
     class variant_observable;
 } // namespace rpp
-
-
-#define RPP_CHECK_IF_TRAIT_ASSERTS_SATISFIED(Op, Type)                            \
-    /* operator_traits can be instantiated if all inner static_asserts are fine*/ \
-    if constexpr (requires { { typename std::decay_t<Op>::template operator_traits<Type>{}}; })
diff --git a/src/rpp/rpp/observables/observable.hpp b/src/rpp/rpp/observables/observable.hpp
index 5932e6b3f..af80ac68b 100644
--- a/src/rpp/rpp/observables/observable.hpp
+++ b/src/rpp/rpp/observables/observable.hpp
@@ -310,42 +310,46 @@ namespace rpp
          */
         auto as_dynamic() && { return rpp::dynamic_observable<Type>{std::move(*this)}; }
 
-        template<constraint::operator_base<Type> Op>
-        auto operator|(Op&& op) const &
+        template<typename... Args>
+        auto operator|(const rpp::operators::details::subscribe_t<Args...>& op) const
         {
-            RPP_CHECK_IF_TRAIT_ASSERTS_SATISFIED(Op, Type)
-            return inner_make_chain_operator(std::forward<Op>(op));
+            return op(*this);
         }
 
-        template<constraint::operator_base<Type> Op>
-        auto operator|(Op&& op) &&
+        template<typename... Args>
+        auto operator|(rpp::operators::details::subscribe_t<Args...>&& op) const
         {
-            RPP_CHECK_IF_TRAIT_ASSERTS_SATISFIED(Op, Type)
-            return std::move(*this).inner_make_chain_operator(std::forward<Op>(op));
+            return std::move(op)(*this);
         }
 
-        template<typename Op>
+        template<std::invocable<const observable&> Op>
+            requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) const &
         {
             return std::forward<Op>(op)(*this);
         }
 
-        template<typename Op>
+        template<std::invocable<observable&&> Op>
+            requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) &&
         {
             return std::forward<Op>(op)(std::move(*this));
         }
 
-        template<typename... Args>
-        auto operator|(const rpp::operators::details::subscribe_t<Args...>& op) const
+        template<typename Op>
+            requires requires { typename std::decay_t<Op>::template operator_traits<Type>; }
+        rpp::constraint::observable auto operator|(Op&& op) const &
         {
-            return op(*this);
+            using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
+            return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), m_strategy};
         }
 
-        template<typename... Args>
-        auto operator|(rpp::operators::details::subscribe_t<Args...>&& op) const
+        template<typename Op>
+            requires requires { typename std::decay_t<Op>::template operator_traits<Type>; }
+        rpp::constraint::observable auto operator|(Op&& op) &&
         {
-            return std::move(op)(*this);
+            using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
+            return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), std::move(m_strategy)};
         }
 
         template<typename Op>
@@ -360,19 +364,6 @@ namespace rpp
             return std::move(*this) | std::forward<Op>(op);
         }
 
-    private:
-        template<constraint::operator_chain<Type, expected_disposable_strategy> Op>
-        auto inner_make_chain_operator(Op&& op) const &
-        {
-            return observable<typename std::decay_t<Op>::template operator_traits<Type>::result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), m_strategy};
-        }
-
-        template<constraint::operator_chain<Type, expected_disposable_strategy> Op>
-        auto inner_make_chain_operator(Op&& op) &&
-        {
-            return observable<typename std::decay_t<Op>::template operator_traits<Type>::result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), std::move(m_strategy)};
-        }
-
     private:
         RPP_NO_UNIQUE_ADDRESS Strategy m_strategy;
     };
diff --git a/src/rpp/rpp/operators/flat_map.hpp b/src/rpp/rpp/operators/flat_map.hpp
index 1f7b5266d..8044cd752 100644
--- a/src/rpp/rpp/operators/flat_map.hpp
+++ b/src/rpp/rpp/operators/flat_map.hpp
@@ -22,20 +22,18 @@ namespace rpp::operators::details
         RPP_NO_UNIQUE_ADDRESS Fn m_fn;
 
         template<rpp::constraint::observable TObservable>
-            requires (std::invocable<Fn, rpp::utils::extract_observable_type_t<TObservable>>
-                      && rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::extract_observable_type_t<TObservable>>>)
         auto operator()(TObservable&& observable) const &
         {
+            static_assert(std::invocable<Fn, rpp::utils::extract_observable_type_t<TObservable>> && rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::extract_observable_type_t<TObservable>>>, "fn should return observable");
             return std::forward<TObservable>(observable)
                  | rpp::ops::map(m_fn)
                  | rpp::ops::merge();
         }
 
         template<rpp::constraint::observable TObservable>
-            requires (std::invocable<Fn, rpp::utils::extract_observable_type_t<TObservable>>
-                      && rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::extract_observable_type_t<TObservable>>>)
         auto operator()(TObservable&& observable) &&
         {
+            static_assert(std::invocable<Fn, rpp::utils::extract_observable_type_t<TObservable>> && rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::extract_observable_type_t<TObservable>>>, "fn should return observable");
             return std::forward<TObservable>(observable)
                  | rpp::ops::map(std::move(m_fn))
                  | rpp::ops::merge();
diff --git a/src/rpp/rpp/operators/multicast.hpp b/src/rpp/rpp/operators/multicast.hpp
index 91adc7729..a8526dfd0 100644
--- a/src/rpp/rpp/operators/multicast.hpp
+++ b/src/rpp/rpp/operators/multicast.hpp
@@ -22,9 +22,9 @@ namespace rpp::operators::details
         RPP_NO_UNIQUE_ADDRESS Subject m_subject;
 
         template<rpp::constraint::observable TObservable>
-            requires std::same_as<rpp::utils::extract_observable_type_t<TObservable>, rpp::subjects::utils::extract_subject_type_t<Subject>>
         auto operator()(TObservable&& observable) const
         {
+            static_assert(std::same_as<rpp::utils::extract_observable_type_t<TObservable>, rpp::subjects::utils::extract_subject_type_t<Subject>>, "observable and subject should be of same type");
             return rpp::connectable_observable<std::decay_t<TObservable>, Subject>{std::forward<TObservable>(observable), m_subject};
         }
     };
@@ -33,9 +33,10 @@ namespace rpp::operators::details
     struct template_multicast_t
     {
         template<rpp::constraint::observable TObservable>
-            requires rpp::constraint::subject<Subject<rpp::utils::extract_observable_type_t<TObservable>>>
         auto operator()(TObservable&& observable) const
         {
+            static_assert(rpp::constraint::subject<Subject<rpp::utils::extract_observable_type_t<TObservable>>>, "subject should be constructible with type of observable");
+
             return rpp::connectable_observable<std::decay_t<TObservable>,
                                                Subject<rpp::utils::extract_observable_type_t<TObservable>>>{std::forward<TObservable>(observable),
                                                                                                             Subject<rpp::utils::extract_observable_type_t<TObservable>>{}};
diff --git a/src/rpp/rpp/operators/start_with.hpp b/src/rpp/rpp/operators/start_with.hpp
index 87a72047a..2f7fbb70a 100644
--- a/src/rpp/rpp/operators/start_with.hpp
+++ b/src/rpp/rpp/operators/start_with.hpp
@@ -23,9 +23,9 @@ namespace rpp::operators::details
         rpp::utils::tuple<TObservables...> observables{};
 
         template<rpp::constraint::observable TObservable>
-            requires rpp::constraint::observables_of_same_type<TObservable, TObservables...>
         auto operator()(TObservable&& observable) const
         {
+            static_assert(rpp::constraint::observables_of_same_type<TObservable, TObservables...>, "observables should be of same type");
             return observables.apply(&apply<TObservable>, std::forward<TObservable>(observable));
         }
 
diff --git a/src/rpp/rpp/operators/subscribe.hpp b/src/rpp/rpp/operators/subscribe.hpp
index 84ca2933d..9f9c75813 100644
--- a/src/rpp/rpp/operators/subscribe.hpp
+++ b/src/rpp/rpp/operators/subscribe.hpp
@@ -60,16 +60,16 @@ namespace rpp::operators::details
         }
 
         template<rpp::constraint::observable Observable>
-            requires rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>
         void operator()(const Observable& observable) const &
         {
+            static_assert(rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>, "observer and observable should be of same type");
             observable.subscribe(m_observer_strategy);
         }
 
         template<rpp::constraint::observable Observable>
-            requires rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>
         void operator()(const Observable& observable) &&
         {
+            static_assert(rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>, "observer and observable should be of same type");
             observable.subscribe(std::move(m_observer_strategy));
         }
 
@@ -116,17 +116,17 @@ namespace rpp::operators::details
         }
 
         template<rpp::constraint::observable Observable>
-            requires rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>
         rpp::composite_disposable_wrapper operator()(const Observable& observable) const &
         {
+            static_assert(rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>, "observer and observable should be of same type");
             observable.subscribe(m_disposable, m_observer_strategy);
             return m_disposable;
         }
 
         template<rpp::constraint::observable Observable>
-            requires rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>
         rpp::composite_disposable_wrapper operator()(const Observable& observable) &&
         {
+            static_assert(rpp::constraint::observer_strategy<ObserverStrategy, rpp::utils::extract_observable_type_t<Observable>>, "observer and observable should be of same type");
             observable.subscribe(m_disposable, std::move(m_observer_strategy));
             return m_disposable;
         }
@@ -150,16 +150,16 @@ namespace rpp::operators::details
         }
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observable_strategy<Type> Strategy>
-            requires std::invocable<OnNext, Type>
         void operator()(const rpp::observable<Type, Strategy>& observable) &&
         {
+            static_assert(std::invocable<OnNext, Type>, "OnNext should be suitable for type of observable");
             observable.subscribe(std::move(m_on_next), std::move(m_on_error), std::move(m_on_completed));
         }
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observable_strategy<Type> Strategy>
-            requires std::invocable<OnNext, Type>
         void operator()(const rpp::observable<Type, Strategy>& observable) const &
         {
+            static_assert(std::invocable<OnNext, Type>, "OnNext should be suitable for type of observable");
             observable.subscribe(m_on_next, m_on_error, m_on_completed);
         }
 
@@ -183,17 +183,17 @@ namespace rpp::operators::details
         }
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observable_strategy<Type> Strategy>
-            requires std::invocable<OnNext, Type>
         rpp::composite_disposable_wrapper operator()(const rpp::observable<Type, Strategy>& observable) &&
         {
+            static_assert(std::invocable<OnNext, Type>, "OnNext should be suitable for type of observable");
             observable.subscribe(m_disposable, std::move(m_on_next), std::move(m_on_error), std::move(m_on_completed));
             return std::move(m_disposable);
         }
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observable_strategy<Type> Strategy>
-            requires std::invocable<OnNext, Type>
         rpp::composite_disposable_wrapper operator()(const rpp::observable<Type, Strategy>& observable) const &
         {
+            static_assert(std::invocable<OnNext, Type>, "OnNext should be suitable for type of observable");
             observable.subscribe(m_disposable, m_on_next, m_on_error, m_on_completed);
             return m_disposable;
         }

From 65cd1422d8dca5a99645119861ff7f167b47f988 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 6 Sep 2024 17:13:41 +0300
Subject: [PATCH 04/10] fix

---
 src/rpp/rpp/observables/observable.hpp | 33 +++++++++++---------------
 src/rpp/rpp/operators/start_with.hpp   |  3 ++-
 2 files changed, 16 insertions(+), 20 deletions(-)

diff --git a/src/rpp/rpp/observables/observable.hpp b/src/rpp/rpp/observables/observable.hpp
index af80ac68b..769ebcc0e 100644
--- a/src/rpp/rpp/observables/observable.hpp
+++ b/src/rpp/rpp/observables/observable.hpp
@@ -322,34 +322,29 @@ namespace rpp
             return std::move(op)(*this);
         }
 
-        template<std::invocable<const observable&> Op>
-            requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
-        rpp::constraint::observable auto operator|(Op&& op) const &
-        {
-            return std::forward<Op>(op)(*this);
-        }
-
-        template<std::invocable<observable&&> Op>
-            requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
-        rpp::constraint::observable auto operator|(Op&& op) &&
-        {
-            return std::forward<Op>(op)(std::move(*this));
-        }
 
         template<typename Op>
-            requires requires { typename std::decay_t<Op>::template operator_traits<Type>; }
+            requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) const &
         {
-            using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
-            return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), m_strategy};
+            if constexpr(requires { typename std::decay_t<Op>::template operator_traits<Type>; }) {
+                using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
+                return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), m_strategy};
+            } else {
+                return std::forward<Op>(op)(*this);
+            }
         }
 
         template<typename Op>
-            requires requires { typename std::decay_t<Op>::template operator_traits<Type>; }
+            requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) &&
         {
-            using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
-            return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), std::move(m_strategy)};
+            if constexpr(requires { typename std::decay_t<Op>::template operator_traits<Type>; }) {
+                using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
+                return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), std::move(m_strategy)};
+            } else {
+                return std::forward<Op>(op)(std::move(*this));
+            }
         }
 
         template<typename Op>
diff --git a/src/rpp/rpp/operators/start_with.hpp b/src/rpp/rpp/operators/start_with.hpp
index 2f7fbb70a..773c72068 100644
--- a/src/rpp/rpp/operators/start_with.hpp
+++ b/src/rpp/rpp/operators/start_with.hpp
@@ -50,9 +50,10 @@ namespace rpp::operators::details
         {
         }
 
-        template<rpp::constraint::observable_of_type<rpp::utils::iterable_value_t<PackedContainer>> TObservable>
+        template<rpp::constraint::observable TObservable>
         auto operator()(TObservable&& observable) const
         {
+            static_assert(rpp::constraint::observable_of_type<TObservable, rpp::utils::iterable_value_t<PackedContainer>>, "observables should be of same type");
             return rpp::source::concat(rpp::source::from_iterable(container, scheduler), std::forward<TObservable>(observable));
         }
     };

From 65f8c337979aa2e6ea54602711b585a4a7438844 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 6 Sep 2024 17:13:48 +0300
Subject: [PATCH 05/10] revert

---
 src/examples/rpp/doxygen/map.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/examples/rpp/doxygen/map.cpp b/src/examples/rpp/doxygen/map.cpp
index 01d869ea3..4e22e1332 100644
--- a/src/examples/rpp/doxygen/map.cpp
+++ b/src/examples/rpp/doxygen/map.cpp
@@ -11,7 +11,6 @@ int main() // NOLINT(bugprone-exception-escape)
     //! [Same type]
     rpp::source::just(42)
         | rpp::operators::map([](int value) { return value + 10; })
-        | rpp::ops::start_with(std::string{})
         | rpp::operators::subscribe([](int v) { std::cout << v << std::endl; });
     // Output: 52
     //! [Same type]

From 1a264c47fa261b7da0c20f755b21f7e9c015cd81 Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Fri, 6 Sep 2024 14:14:01 +0000
Subject: [PATCH 06/10] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 src/rpp/rpp/observables/observable.hpp | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/src/rpp/rpp/observables/observable.hpp b/src/rpp/rpp/observables/observable.hpp
index 769ebcc0e..94ad8a385 100644
--- a/src/rpp/rpp/observables/observable.hpp
+++ b/src/rpp/rpp/observables/observable.hpp
@@ -327,10 +327,13 @@ namespace rpp
             requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) const &
         {
-            if constexpr(requires { typename std::decay_t<Op>::template operator_traits<Type>; }) {
+            if constexpr (requires { typename std::decay_t<Op>::template operator_traits<Type>; })
+            {
                 using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
                 return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), m_strategy};
-            } else {
+            }
+            else
+            {
                 return std::forward<Op>(op)(*this);
             }
         }
@@ -339,10 +342,13 @@ namespace rpp
             requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) &&
         {
-            if constexpr(requires { typename std::decay_t<Op>::template operator_traits<Type>; }) {
+            if constexpr (requires { typename std::decay_t<Op>::template operator_traits<Type>; })
+            {
                 using result_type = typename std::decay_t<Op>::template operator_traits<Type>::result_type;
                 return observable<result_type, details::observables::make_chain_t<std::decay_t<Op>, Strategy>>{std::forward<Op>(op), std::move(m_strategy)};
-            } else {
+            }
+            else
+            {
                 return std::forward<Op>(op)(std::move(*this));
             }
         }

From 7b1174e275d9c242e709023f566eeeb9f621f995 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 6 Sep 2024 18:08:45 +0300
Subject: [PATCH 07/10] compile fixes

---
 .../observables/connectable_observable.hpp    | 26 +++++++------------
 1 file changed, 10 insertions(+), 16 deletions(-)

diff --git a/src/rpp/rpp/observables/connectable_observable.hpp b/src/rpp/rpp/observables/connectable_observable.hpp
index 1326ca050..31ed3d830 100644
--- a/src/rpp/rpp/observables/connectable_observable.hpp
+++ b/src/rpp/rpp/observables/connectable_observable.hpp
@@ -130,28 +130,22 @@ namespace rpp
                                    details::ref_count_on_subscribe_t<connectable_observable<OriginalObservable, Subject>>>{*this};
         }
 
-        template<std::invocable<const connectable_observable&> Op>
-        rpp::constraint::observable auto operator|(Op&& op) const &
-        {
-            return std::forward<Op>(op)(*this);
-        }
-
-        template<std::invocable<connectable_observable&&> Op>
-        rpp::constraint::observable auto operator|(Op&& op)&&
-        {
-            return std::forward<Op>(op)(std::move(*this));
-        }
-
         template<typename Op>
-        decltype(std::declval<const base>() | std::declval<Op>()) operator|(Op&& op) const &
+        auto operator|(Op&& op) const &
         {
-            return static_cast<const base&>(*this) | std::forward<Op>(op);
+            if constexpr (std::invocable<std::decay_t<Op>, const connectable_observable&>)
+                return std::forward<Op>(op)(*this);
+            else
+                return static_cast<const base&>(*this) | std::forward<Op>(op);
         }
 
         template<typename Op>
-        decltype(std::declval<base>() | std::declval<Op>()) operator|(Op&& op)&&
+        auto operator|(Op&& op)&&
         {
-            return static_cast<base&&>(*this) | std::forward<Op>(op);
+            if constexpr (std::invocable<std::decay_t<Op>, connectable_observable&&>)
+                return std::forward<Op>(op)(std::move(*this));
+            else
+                return static_cast<base&&>(*this) | std::forward<Op>(op);
         }
 
     private:

From 66e5a5eb64e6e0d00b144f848661d64758240b96 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 6 Sep 2024 18:15:54 +0300
Subject: [PATCH 08/10] fix a bit

---
 .../rpp/observables/connectable_observable.hpp | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/src/rpp/rpp/observables/connectable_observable.hpp b/src/rpp/rpp/observables/connectable_observable.hpp
index 31ed3d830..ae4471469 100644
--- a/src/rpp/rpp/observables/connectable_observable.hpp
+++ b/src/rpp/rpp/observables/connectable_observable.hpp
@@ -134,7 +134,10 @@ namespace rpp
         auto operator|(Op&& op) const &
         {
             if constexpr (std::invocable<std::decay_t<Op>, const connectable_observable&>)
+            {
+                static_assert(rpp::constraint::observable<std::invoke_result_t<std::decay_t<Op>, const connectable_observable&>>, "Result of Op should be observable");
                 return std::forward<Op>(op)(*this);
+            }
             else
                 return static_cast<const base&>(*this) | std::forward<Op>(op);
         }
@@ -143,11 +146,26 @@ namespace rpp
         auto operator|(Op&& op)&&
         {
             if constexpr (std::invocable<std::decay_t<Op>, connectable_observable&&>)
+            {
+                static_assert(rpp::constraint::observable<std::invoke_result_t<std::decay_t<Op>, connectable_observable&&>>, "Result of Op should be observable");
                 return std::forward<Op>(op)(std::move(*this));
+            }
             else
                 return static_cast<base&&>(*this) | std::forward<Op>(op);
         }
 
+        template<typename Op>
+        auto pipe(Op&& op) const &
+        {
+            return *this | std::forward<Op>(op);
+        }
+
+        template<typename Op>
+        auto pipe(Op&& op) &&
+        {
+            return std::move(*this) | std::forward<Op>(op);
+        }
+
     private:
         RPP_NO_UNIQUE_ADDRESS OriginalObservable m_original_observable;
         Subject                                  m_subject;

From 0526a5988fc5ddeb71c0a804794cae0098c04b33 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 6 Sep 2024 18:17:18 +0300
Subject: [PATCH 09/10] one more

---
 src/rpp/rpp/observables/observable.hpp | 14 ++++----------
 1 file changed, 4 insertions(+), 10 deletions(-)

diff --git a/src/rpp/rpp/observables/observable.hpp b/src/rpp/rpp/observables/observable.hpp
index 94ad8a385..a0a0daea7 100644
--- a/src/rpp/rpp/observables/observable.hpp
+++ b/src/rpp/rpp/observables/observable.hpp
@@ -310,19 +310,13 @@ namespace rpp
          */
         auto as_dynamic() && { return rpp::dynamic_observable<Type>{std::move(*this)}; }
 
-        template<typename... Args>
-        auto operator|(const rpp::operators::details::subscribe_t<Args...>& op) const
-        {
-            return op(*this);
-        }
-
-        template<typename... Args>
-        auto operator|(rpp::operators::details::subscribe_t<Args...>&& op) const
+        template<typename Subscribe>
+            requires rpp::utils::is_base_of_v<std::decay_t<Subscribe>, rpp::operators::details::subscribe_t>
+        auto operator|(Subscribe&& op) const
         {
-            return std::move(op)(*this);
+            return std::forward<Subscribe>(op)(*this);
         }
 
-
         template<typename Op>
             requires (!rpp::utils::is_base_of_v<std::decay_t<Op>, rpp::operators::details::subscribe_t>)
         rpp::constraint::observable auto operator|(Op&& op) const &

From 85c5c02adb9428a93731d9e84fe928f94a50a8e3 Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Fri, 6 Sep 2024 15:20:45 +0000
Subject: [PATCH 10/10] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 src/rpp/rpp/observables/connectable_observable.hpp | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/rpp/rpp/observables/connectable_observable.hpp b/src/rpp/rpp/observables/connectable_observable.hpp
index ae4471469..557699e4b 100644
--- a/src/rpp/rpp/observables/connectable_observable.hpp
+++ b/src/rpp/rpp/observables/connectable_observable.hpp
@@ -155,13 +155,13 @@ namespace rpp
         }
 
         template<typename Op>
-        auto pipe(Op&& op) const &
+        auto pipe(Op && op) const &
         {
             return *this | std::forward<Op>(op);
         }
 
         template<typename Op>
-        auto pipe(Op&& op) &&
+        auto pipe(Op && op)&&
         {
             return std::move(*this) | std::forward<Op>(op);
         }