add example for set_error

fbshipit-source-id: f8fb9b36c34d11ffee5649291a7d09a796733567

folly/experimental/pushmi/buildSingleHeader.cmake

@@ -47,11 +47,13 @@ set(header_files
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/subject.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/empty.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/just.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/error.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/defer.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/on.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/tap.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/transform.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/filter.h"
+    "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/switch_on_error.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/via.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/request_via.h"
     "${CMAKE_CURRENT_SOURCE_DIR}/include/pushmi/o/share.h"

folly/experimental/pushmi/examples/CMakeLists.txt

@@ -10,3 +10,4 @@ add_subdirectory(composition)
 add_subdirectory(for_each)
 add_subdirectory(reduce)
 add_subdirectory(set_done)
+add_subdirectory(set_error)

folly/experimental/pushmi/examples/include/no_fail.h

+#pragma once
+// Copyright (c) 2018-present, Facebook, Inc.
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+#include <pushmi/single_deferred.h>
+#include <pushmi/o/submit.h>
+
+namespace pushmi {
+
+namespace detail {
+
+struct no_fail_fn {
+  auto operator()() const {
+    return constrain(lazy::Sender<_1>, [](auto in) {
+      using In = decltype(in);
+      return ::pushmi::detail::deferred_from<In, single<>>(
+        std::move(in),
+        ::pushmi::detail::submit_transform_out<In>(
+          constrain(lazy::Receiver<_1>, [](auto out) {
+            using Out = decltype(out);
+            return ::pushmi::detail::out_from_fn<In>()(
+              std::move(out),
+              ::pushmi::on_error([](auto&, auto&&) noexcept {
+                std::abort();
+              })
+            );
+          })
+        )
+      );
+    });
+  }
+};
+
+} // namespace detail
+
+namespace operators {
+PUSHMI_INLINE_VAR constexpr detail::no_fail_fn no_fail{};
+} // namespace operators
+
+} // namespace pushmi

folly/experimental/pushmi/examples/set_done/set_done_2.cpp

@@ -28,13 +28,14 @@ auto get_setting() {
 auto println = [](auto v){std::cout << v << std::endl;};
 
 // concat not yet implemented
+template<class T, class E = std::exception_ptr>
 auto concat =
   [](auto in){
     return mi::make_single_deferred(
       [in](auto out) mutable {
         ::pushmi::submit(in, mi::make_single(out,
         [](auto out, auto v){
-          // ::pushmi::submit(v, out);
+          ::pushmi::submit(v, mi::any_single<T, E>(out));
         }));
       });
   };
@@ -57,7 +58,7 @@ int main()
         }
         return mi::any_single_deferred<std::string>{op::just(std::to_string(i))};
       }) |
-      concat |
+      concat<std::string> |
       op::submit(println);
 
   std::cout << "OK" << std::endl;

folly/experimental/pushmi/examples/set_error/CMakeLists.txt

+
+
+add_executable(set_error_2 set_error_2.cpp)
+target_link_libraries(set_error_2
+  pushmi
+  examples
+  Threads::Threads)

folly/experimental/pushmi/examples/set_error/set_error_2.cpp

+#include <vector>
+#include <algorithm>
+#include <cassert>
+#include <iostream>
+
+#include <pushmi/o/just.h>
+#include <pushmi/o/error.h>
+#include <pushmi/o/empty.h>
+#include <pushmi/o/transform.h>
+#include <pushmi/o/switch_on_error.h>
+#include <no_fail.h>
+
+using namespace pushmi::aliases;
+
+// concat not yet implemented
+template<class T, class E = std::exception_ptr>
+auto concat =
+  [](auto in){
+    return mi::make_single_deferred(
+      [in](auto out) mutable {
+        ::pushmi::submit(in, mi::make_single(out,
+        [](auto out, auto v){
+          ::pushmi::submit(v, mi::any_single<T, E>(out));
+        }));
+      });
+  };
+
+int main()
+{
+  // support all error value types
+
+  op::error(std::exception_ptr{}) |
+    op::submit();
+
+  op::error(std::errc::argument_list_too_long) |
+    op::submit();
+
+// transform an error
+
+  op::error(std::errc::argument_list_too_long) |
+    op::switch_on_error([](auto e) noexcept { return op::error(std::exception_ptr{}); }) |
+    op::submit();
+
+// use default value if an error occurs
+
+  op::just(42) |
+    op::switch_on_error([](auto e) noexcept { return op::just(0); }) |
+    op::submit();
+
+// suppress if an error occurs
+
+  op::error(std::errc::argument_list_too_long) |
+    op::switch_on_error([](auto e) noexcept { return op::empty(); }) |
+    op::submit();
+
+// abort if an error occurs
+
+  op::just(42) |
+    op::no_fail() |
+    op::submit();
+
+// transform value to error_
+
+  op::just(42) |
+    op::transform([](auto v) {
+      using r_t = mi::any_single_deferred<int>;
+      if (v < 40) {
+        return r_t{op::error<int>(std::exception_ptr{})};
+      } else {
+        return r_t{op::just(v)};
+      }
+    }) |
+    concat<int> |
+    op::submit();
+
+// retry on error
+
+  // http.get(ex) |
+  //   op::timeout(ex, 1s) |
+  //   op::switch_on_error([](auto e) noexcept { return op::timer(ex, 1s); }) |
+  //   op::repeat() |
+  //   op::timeout(ex, 10s) |
+  //   op::submit();
+
+  std::cout << "OK" << std::endl;
+}

folly/experimental/pushmi/include/pushmi.h

@@ -5667,6 +5667,49 @@ auto just(V v) {
 // This source code is licensed under the MIT license found in the
 // LICENSE file in the root directory of this source tree.
 
+//#include "../deferred.h"
+//#include "submit.h"
+//#include "extension_operators.h"
+
+namespace pushmi {
+
+namespace operators {
+
+PUSHMI_TEMPLATE(class E)
+  (requires SemiMovable<E>)
+auto error(E e) {
+  return make_deferred(
+    constrain(lazy::NoneReceiver<_1, E>,
+      [e = std::move(e)](auto out) mutable {
+        ::pushmi::set_error(out, std::move(e));
+      }
+    )
+  );
+}
+
+PUSHMI_TEMPLATE(class V, class E)
+  (requires SemiMovable<V> && SemiMovable<E>)
+auto error(E e) {
+  return make_single_deferred(
+    constrain(lazy::SingleReceiver<_1, V, E>,
+      [e = std::move(e)](auto out) mutable {
+        ::pushmi::set_error(out, std::move(e));
+      }
+    )
+  );
+}
+
+} // namespace operators
+
+} // namespace pushmi
+// clang-format off
+// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
+//#pragma once
+// Copyright (c) 2018-present, Facebook, Inc.
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
 //#include "../single.h"
 //#include "../single_deferred.h"
 //#include "submit.h"
@@ -6164,6 +6207,54 @@ PUSHMI_INLINE_VAR constexpr detail::filter_fn filter{};
 // This source code is licensed under the MIT license found in the
 // LICENSE file in the root directory of this source tree.
 
+//#include "../piping.h"
+//#include "extension_operators.h"
+
+namespace pushmi {
+
+namespace detail {
+
+struct switch_on_error_fn {
+  PUSHMI_TEMPLATE(class ErrorSelector)
+    (requires SemiMovable<ErrorSelector>)
+  auto operator()(ErrorSelector es) const {
+    return constrain(lazy::Sender<_1>, [es = std::move(es)](auto in) {
+      using In = decltype(in);
+      return ::pushmi::detail::deferred_from<In, single<>>(
+        std::move(in),
+        ::pushmi::detail::submit_transform_out<In>(
+          constrain(lazy::Receiver<_1>, [es](auto out) {
+            using Out = decltype(out);
+            return ::pushmi::detail::out_from_fn<In>()(
+              std::move(out),
+              // copy 'es' to allow multiple calls to submit
+              ::pushmi::on_error([es](auto& out, auto&& e) noexcept {
+                static_assert(::pushmi::NothrowInvocable<ErrorSelector, decltype(e)>,
+                  "switch_on_error - error selector function must be noexcept");
+                auto next = es(std::move(e));
+                ::pushmi::submit(next, std::move(out));
+              })
+            );
+          })
+        )
+      );
+    });
+  }
+};
+
+} // namespace detail
+
+namespace operators {
+PUSHMI_INLINE_VAR constexpr detail::switch_on_error_fn switch_on_error{};
+} // namespace operators
+
+} // namespace pushmi
+//#pragma once
+// Copyright (c) 2018-present, Facebook, Inc.
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
 //#include "../piping.h"
 //#include "../executor.h"
 //#include "extension_operators.h"

folly/experimental/pushmi/include/pushmi/o/error.h

+// clang-format off
+// clang format does not support the '<>' in the lambda syntax yet.. []<>()->{}
+#pragma once
+// Copyright (c) 2018-present, Facebook, Inc.
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+#include "../deferred.h"
+#include "submit.h"
+#include "extension_operators.h"
+
+namespace pushmi {
+
+namespace operators {
+
+PUSHMI_TEMPLATE(class E)
+  (requires SemiMovable<E>)
+auto error(E e) {
+  return make_deferred(
+    constrain(lazy::NoneReceiver<_1, E>,
+      [e = std::move(e)](auto out) mutable {
+        ::pushmi::set_error(out, std::move(e));
+      }
+    )
+  );
+}
+
+PUSHMI_TEMPLATE(class V, class E)
+  (requires SemiMovable<V> && SemiMovable<E>)
+auto error(E e) {
+  return make_single_deferred(
+    constrain(lazy::SingleReceiver<_1, V, E>,
+      [e = std::move(e)](auto out) mutable {
+        ::pushmi::set_error(out, std::move(e));
+      }
+    )
+  );
+}
+
+} // namespace operators
+
+} // namespace pushmi

folly/experimental/pushmi/include/pushmi/o/switch_on_error.h

+#pragma once
+// Copyright (c) 2018-present, Facebook, Inc.
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+#include "../piping.h"
+#include "extension_operators.h"
+
+namespace pushmi {
+
+namespace detail {
+
+struct switch_on_error_fn {
+  PUSHMI_TEMPLATE(class ErrorSelector)
+    (requires SemiMovable<ErrorSelector>)
+  auto operator()(ErrorSelector es) const {
+    return constrain(lazy::Sender<_1>, [es = std::move(es)](auto in) {
+      using In = decltype(in);
+      return ::pushmi::detail::deferred_from<In, single<>>(
+        std::move(in),
+        ::pushmi::detail::submit_transform_out<In>(
+          constrain(lazy::Receiver<_1>, [es](auto out) {
+            using Out = decltype(out);
+            return ::pushmi::detail::out_from_fn<In>()(
+              std::move(out),
+              // copy 'es' to allow multiple calls to submit
+              ::pushmi::on_error([es](auto& out, auto&& e) noexcept {
+                static_assert(::pushmi::NothrowInvocable<ErrorSelector, decltype(e)>,
+                  "switch_on_error - error selector function must be noexcept");
+                auto next = es(std::move(e));
+                ::pushmi::submit(next, std::move(out));
+              })
+            );
+          })
+        )
+      );
+    });
+  }
+};
+
+} // namespace detail
+
+namespace operators {
+PUSHMI_INLINE_VAR constexpr detail::switch_on_error_fn switch_on_error{};
+} // namespace operators
+
+} // namespace pushmi