Refactor disposables

Doxyfile

@@ -230,7 +230,7 @@ JAVADOC_BANNER         = NO
 # If the QT_AUTOBRIEF tag is set to YES then Doxygen will interpret the first
 # line (until the first dot) of a Qt-style comment as the brief description. If
 # set to NO, the Qt-style will behave just like regular Qt-style comments (thus
-# requiring an explicit \brief command for a brief description.)
+# requiring an explicit @brief command for a brief description.)
 # The default value is: NO.
 
 QT_AUTOBRIEF           = NO
@@ -432,7 +432,7 @@ DISTRIBUTE_GROUP_DOC   = NO
 
 # If one adds a struct or class to a group and this option is enabled, then also
 # any nested class or struct is added to the same group. By default this option
-# is disabled and one has to add nested compounds explicitly via \ingroup.
+# is disabled and one has to add nested compounds explicitly via @ingroup.
 # The default value is: NO.
 
 GROUP_NESTED_COMPOUNDS = NO
@@ -447,7 +447,7 @@ GROUP_NESTED_COMPOUNDS = NO
 SUBGROUPING            = YES
 
 # When the INLINE_GROUPED_CLASSES tag is set to YES, classes, structs and unions
-# are shown inside the group in which they are included (e.g. using \ingroup)
+# are shown inside the group in which they are included (e.g. using @ingroup)
 # instead of on a separate page (for HTML and Man pages) or section (for LaTeX
 # and RTF).
 #

docs/readme.md

@@ -131,98 +131,15 @@ In such an way it is not powerful enough, so Reactive Programming provides a lis
 
 \copydoc observables
 
-### Operators
-
-\copydoc operators
-
-#### How Operators Work?
-
-Example:
-
-```cpp
-rpp::source::create<int>([](const auto& observer){
-  observer.on_next(1);
-  observer.on_completed();
-});
-```
-
-This example shows next: we create observble of `int` via operator `create`. This observable just emits to observer value `1` and then completes. Type of this observable is `rpp::observable<int, ...>` where `...` implementation defined type. So, actually it is `observable of ints`. Let's say we want to convert `int` to `std::string`. We could subscribe and then convert it or use `map` operator (also known as `transform`) to transform some original value to some another value:
+### Observers:
 
-```cpp
-rpp::source::create<int>([](const auto& observer){
-  observer.on_next(1);
-  observer.on_completed();
-})
-| rpp::operators::map([](int v){ return std::string{v}; });
-```
-
-For now it is `observable of strings` due to it is `rpp::observable<std::string, ...>`. But what is `rpp::operators::map` then? Actually it is functor-adaptor - just functor accepting observable and returning another observable. It accepts original observable and converts it to observable of "final type". "final type" is result of invocation of passed function against original observable's type. In our case it is `decltype([](int v){ return std::string{v}; }(int{}))` is it is `std::string`. So, `map` can be implemented in the following way:
-
-```cpp
-template<typename Fn>
-struct map
-{
-  Fn fn{};
+\copydoc observers
 
-  template<typename Type, typename Internal>
-  auto operator()(const rpp::observable<Type, Internal>& observable) const {
-    using FinalType = std::invoke_result_t<Fn, Type>;
-    return rpp::source::create<FinalType>([observable, fn](const rpp::dynamic_observer<FinalType>& observer)
-    {
-      observable.subscribe([observer, fn](const auto& v) { observer.on_next(fn(v)); },
-                           [observer](const std::exception_ptr& err) { observer.on_error(err); },
-                           [observer]() { observer.on_completed(); });
-    };);
-  }
-}
-```
-
-It is template for such an functor-adaptor. Provided example - is simplest possible way to implement new operators - just provide function for transformation of observable. For example, it is fully valid example:
-```cpp
-rpp::source::just(1)
-    | [](const auto& observable) { return rpp::source::concat(observable, rpp::source::just(2)); };
-```
-
-There we convert observable to concatenation of original observable and `just(2)`.
-
-One more posible but a bit more advanced way to implement operators - is to lift observer. To do this, your functor-adapter must to satisfy `rpp::constraint::operator_lift` concept. Actually, your class must to have:
-- member function `lift` accepting downstream observer and returning new upstream observer
-- inner `template<rpp::constraint::decayed_type T> struct traits` struct accepting typename of upstream and providing:
-  - `using result_type =` with typename of new resulting type for new observable
-  - (optionally) `struct requirements` with static_asserts over passed type
-
-Example:
-```cpp
-template<typename Fn>
-struct map
-{
-  template<rpp::constraint::decayed_type T>
-  struct traits
-  {
-    struct requirements
-    {
-      static_assert(std::invocable<Fn, T>, "Fn is not invocable with T");
-    };
-
-    using result_type = std::invoke_result_t<Fn, T>;
-  };
-
-  Fn fn{};
-
-  template<typename Upstream, typename Downstream>
-  auto lift(const rpp::dynamic_observer<Downstream>& observer) const
-  {
-      return rpp::make_lambda_observer<Upstream>([observer, fn](const auto& v){ observer.on_next(fn(v)); },
-                                                 [observer](const std::exception_ptr& err) { observer.on_error(err); },
-                                                 [observer]() { observer.on_completed(); });
-  }
-}
+### Operators
 
-```
-In this case you providing logic how to convert downstream observer to upstream observer. Actually this implementation is equal to previous one, but without handling of observable - you are expressing your operator in terms of observers
+\copydoc operators
 
-**(Advanced)**
-In case of implementing operator via `lift` you can control disposable strategy via `updated_disposable_strategy` parameter. It accepts disposable strategy of upstream and returns disposable strategy for downstream. It needed only for optimization and reducing disposables handling cost and it is purely advanced thing. Not sure if anyone is going to use it by its own for now =)
+Check the @link operators @endlink for more details about operators.
 
 ### Schedulers
 
@@ -419,14 +336,14 @@ Below you can find list of extensions for RPP with adaption to external framewor
 ### rppqt
 
 \copydoc rppqt
-Check API reference of \link rppqt \endlink for more details
+Check API reference of @link rppqt @endlink for more details
 
 ### rppgrpc
 
 \copydoc rppgrpc
-Check API reference of \link rppgrpc \endlink for more details
+Check API reference of @link rppgrpc @endlink for more details
 
 ### rppasio
 
 \copydoc rppasio
-Check API reference of \link rppasio \endlink for more details
+Check API reference of @link rppasio @endlink for more details

src/examples/rpp/doxygen/as_blocking.cpp

@@ -5,7 +5,7 @@
 #include <iostream>
 
 /**
- * \example as_blocking.cpp
+ * @example as_blocking.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/concat.cpp

@@ -5,7 +5,7 @@
 #include <iostream>
 
 /**
- * \example concat.cpp
+ * @example concat.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/connect.cpp

@@ -5,7 +5,7 @@
 #include <iostream>
 
 /**
- * \example connect.cpp
+ * @example connect.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/create.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example create.cpp
+ * @example create.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/defer.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example defer.cpp
+ * @example defer.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/delay.cpp

@@ -5,7 +5,7 @@
 #include <iostream>
 
 /**
- * \example delay.cpp
+ * @example delay.cpp
  **/
 int main() // NOLINT(bugprone-exception-escape)
 {

src/examples/rpp/doxygen/distinct_until_changed.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example distinct_until_changed.cpp
+ * @example distinct_until_changed.cpp
  **/
 
 int main()

src/examples/rpp/doxygen/from.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example from.cpp
+ * @example from.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/group_by.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example group_by.cpp
+ * @example group_by.cpp
  **/
 int main()
 {

src/examples/rpp/doxygen/interval.cpp

@@ -4,7 +4,7 @@
 #include <iostream>
 
 /**
- * \example interval.cpp
+ * @example interval.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/just.cpp

@@ -4,7 +4,7 @@
 #include <iostream>
 
 /**
- * \example just.cpp
+ * @example just.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/last.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example last.cpp
+ * @example last.cpp
  **/
 int main()
 {

src/examples/rpp/doxygen/map.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example map.cpp
+ * @example map.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rpp/doxygen/observe_on.cpp

@@ -5,7 +5,7 @@
 #include <iostream>
 
 /**
- * \example observe_on.cpp
+ * @example observe_on.cpp
  **/
 int main() // NOLINT(bugprone-exception-escape)
 {

src/examples/rpp/doxygen/readme.cpp

@@ -7,6 +7,54 @@
  * @example readme.cpp
  */
 
+
+// ![simple_custom_map]
+template<typename Fn>
+struct simple_map
+{
+    simple_map(const Fn& fn)
+        : fn(fn)
+    {
+    }
+
+    Fn fn{};
+
+    // 1: define traits for the operator with upstream (previous type) type
+    template<rpp::constraint::decayed_type T>
+    struct operator_traits
+    {
+        // 1.1: it could have static asserts to be sure T is applicable for this operator
+        static_assert(std::invocable<Fn, T>, "Fn is not invocable with T");
+
+        // 1.2: it should have `result_type` is type of new observable after applying this operator
+        using result_type = std::invoke_result_t<Fn, T>;
+    };
+
+    // 2: define updated optimal disposables strategy. Set to `rpp::details::observables::default_disposables_strategy` if you don't know what is that.
+    template<rpp::details::observables::constraint::disposables_strategy Prev>
+    using updated_optimal_disposables_strategy = Prev;
+
+
+    // 3: implement core logic of operator: accept downstream observer (of result_type) and convert it to upstream observer (of T).
+    template<typename Upstream, rpp::constraint::observer Observer>
+    auto lift(Observer&& observer) const
+    {
+        const auto dynamic_observer = std::forward<Observer>(observer).as_dynamic();
+        return rpp::make_lambda_observer<Upstream>([dynamic_observer, fn = fn](const auto& v) { dynamic_observer.on_next(fn(v)); },
+                                                   [dynamic_observer](const std::exception_ptr& err) { dynamic_observer.on_error(err); },
+                                                   [dynamic_observer]() { dynamic_observer.on_completed(); });
+    }
+};
+
+template<typename Fn>
+simple_map(const Fn& fn) -> simple_map<Fn>;
+
+void test()
+{
+    rpp::source::just(1) | simple_map([](int v) { return std::to_string(v); }) | rpp::ops::subscribe();
+}
+// ![simple_custom_map]
+
 int main() // NOLINT(bugprone-exception-escape)
 {
     // ![readme]
@@ -18,5 +66,7 @@ int main() // NOLINT(bugprone-exception-escape)
         | rpp::operators::subscribe([](char v) { std::cout << v; });
     // ![readme]
 
+    rpp::source::just(1) | simple_map([](int v) { return std::to_string(v); }) | rpp::ops::subscribe();
+
     return 0;
 }

src/examples/rpp/doxygen/skip.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example skip.cpp
+ * @example skip.cpp
  **/
 int main()
 {

src/examples/rpp/doxygen/switch_on_next.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example switch_on_next.cpp
+ * @example switch_on_next.cpp
  **/
 int main()
 {

src/examples/rpp/doxygen/take.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example take.cpp
+ * @example take.cpp
  **/
 int main() // NOLINT(bugprone-exception-escape)
 {

src/examples/rpp/doxygen/take_while.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example take_while.cpp
+ * @example take_while.cpp
  **/
 int main() // NOLINT(bugprone-exception-escape)
 {

src/examples/rpp/doxygen/thread_pool.cpp

@@ -3,7 +3,7 @@
 #include <iostream>
 
 /**
- * \example thread_pool.cpp
+ * @example thread_pool.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)

src/examples/rppgrpc/doxygen/client_reactor.cpp

@@ -5,7 +5,7 @@
 
 #include "protocol.grpc.pb.h"
 /**
- * \example client_reactor.cpp
+ * @example client_reactor.cpp
  **/
 
 int main() // NOLINT(bugprone-exception-escape)