Merge pull request #264 from elbeno/multi-then

✨ Add multi-channel `then`

Author: elbeno

include/async/then.hpp

@@ -26,7 +26,7 @@
 
 namespace async {
 namespace _then {
-template <stdx::ct_string Name, typename HandleTag, typename CompleteTag,
+template <stdx::ct_string Name, typename HandleTags, typename CompleteTag,
           typename S, typename R, typename... Fs>
 struct receiver {
     using is_receiver = void;
@@ -53,7 +53,7 @@ struct receiver {
   private:
     template <typename T, typename... Args>
     auto handle(Args &&...args) -> void {
-        if constexpr (std::same_as<HandleTag, T>) {
+        if constexpr (boost::mp11::mp_contains<HandleTags, T>::value) {
             auto results = stdx::call_by_need(
                 std::move(fs),
                 stdx::tuple<Args &&...>{std::forward<Args>(args)...});
@@ -76,15 +76,15 @@ template <typename Tag> struct as_signature {
 };
 } // namespace detail
 
-template <stdx::ct_string Name, typename HandleTag, typename CompleteTag,
+template <stdx::ct_string Name, typename HandleTags, typename CompleteTag,
           typename S, typename... Fs>
 struct sender {
     template <async::receiver R>
     [[nodiscard]] constexpr auto connect(R &&r) && {
         check_connect<sender &&, R>();
         return async::connect(
             std::move(s),
-            receiver<Name, HandleTag, CompleteTag, S, std::remove_cvref_t<R>,
+            receiver<Name, HandleTags, CompleteTag, S, std::remove_cvref_t<R>,
                      Fs...>{std::forward<R>(r), std::move(fs)});
     }
 
@@ -95,8 +95,8 @@ struct sender {
     [[nodiscard]] constexpr auto connect(R &&r) const & {
         check_connect<sender const &, R>();
         return async::connect(
-            s, receiver<Name, HandleTag, CompleteTag, S, std::remove_cvref_t<R>,
-                        Fs...>{std::forward<R>(r), fs});
+            s, receiver<Name, HandleTags, CompleteTag, S,
+                        std::remove_cvref_t<R>, Fs...>{std::forward<R>(r), fs});
     }
 
     template <typename... Ts>
@@ -107,31 +107,23 @@ struct sender {
                                     std::declval<stdx::tuple<Ts...>>()))>;
 
     template <typename Env>
-        requires std::same_as<HandleTag, set_value_t>
     [[nodiscard]] constexpr static auto get_completion_signatures(Env const &) {
-        return transform_completion_signatures_of<
-            S, Env, completion_signatures<>, signatures>{};
-    }
-
-    template <typename Env>
-        requires std::same_as<HandleTag, set_error_t>
-    [[nodiscard]] constexpr static auto get_completion_signatures(Env const &) {
-        return transform_completion_signatures_of<
-            S, Env, completion_signatures<>, ::async::detail::default_set_value,
-            signatures>{};
-    }
-
-    template <typename Env>
-        requires std::same_as<HandleTag, set_stopped_t>
-    [[nodiscard]] constexpr static auto get_completion_signatures(Env const &) {
-        if constexpr (not sends_stopped<S, Env>) {
-            return completion_signatures_of_t<S, Env>{};
-        } else {
-            return transform_completion_signatures_of<
-                S, Env, completion_signatures<signatures<>>,
-                ::async::detail::default_set_value,
-                ::async::detail::default_set_error, completion_signatures<>>{};
-        }
+        using tag_pred =
+            boost::mp11::mp_apply<::async::detail::with_any_tag, HandleTags>;
+        using raw_completions =
+            boost::mp11::mp_partition_q<completion_signatures_of_t<S, Env>,
+                                        tag_pred>;
+
+        using upstream_completions = boost::mp11::mp_first<raw_completions>;
+        using dependent_completions =
+            boost::mp11::mp_flatten<::async::detail::gather_signatures<
+                tag_pred, upstream_completions, signatures,
+                completion_signatures>>;
+
+        using unchanged_completions = boost::mp11::mp_second<raw_completions>;
+
+        return boost::mp11::mp_unique<boost::mp11::mp_append<
+            dependent_completions, unchanged_completions>>{};
     }
 
     using is_sender = void;
@@ -144,15 +136,15 @@ struct sender {
     }
 };
 
-template <stdx::ct_string Name, typename HandleTag, typename CompleteTag,
+template <stdx::ct_string Name, typename HandleTags, typename CompleteTag,
           typename... Fs>
 struct pipeable {
     [[no_unique_address]] stdx::tuple<Fs...> fs;
 
   private:
     template <async::sender S, stdx::same_as_unqualified<pipeable> Self>
     friend constexpr auto operator|(S &&s, Self &&self) -> async::sender auto {
-        return sender<Name, HandleTag, CompleteTag, std::remove_cvref_t<S>,
+        return sender<Name, HandleTags, CompleteTag, std::remove_cvref_t<S>,
                       Fs...>{std::forward<S>(s), std::forward<Self>(self).fs};
     }
 };
@@ -161,7 +153,7 @@ struct pipeable {
 template <stdx::ct_string Name = "then", stdx::callable... Fs>
 [[nodiscard]] constexpr auto then(Fs &&...fs) {
     return compose(
-        _then::pipeable<Name, set_value_t, set_value_t,
+        _then::pipeable<Name, stdx::type_list<set_value_t>, set_value_t,
                         std::remove_cvref_t<Fs>...>{std::forward<Fs>(fs)...});
 }
 
@@ -173,8 +165,8 @@ template <stdx::ct_string Name = "then", sender S, stdx::callable... Fs>
 template <stdx::ct_string Name = "upon_error", stdx::callable F>
 [[nodiscard]] constexpr auto upon_error(F &&f) {
     return compose(
-        _then::pipeable<Name, set_error_t, set_value_t, std::remove_cvref_t<F>>{
-            std::forward<F>(f)});
+        _then::pipeable<Name, stdx::type_list<set_error_t>, set_value_t,
+                        std::remove_cvref_t<F>>{std::forward<F>(f)});
 }
 
 template <stdx::ct_string Name = "upon_error", sender S, stdx::callable F>
@@ -184,8 +176,9 @@ template <stdx::ct_string Name = "upon_error", sender S, stdx::callable F>
 
 template <stdx::ct_string Name = "upon_stopped", stdx::callable F>
 [[nodiscard]] constexpr auto upon_stopped(F &&f) {
-    return compose(_then::pipeable<Name, set_stopped_t, set_value_t,
-                                   std::remove_cvref_t<F>>{std::forward<F>(f)});
+    return compose(
+        _then::pipeable<Name, stdx::type_list<set_stopped_t>, set_value_t,
+                        std::remove_cvref_t<F>>{std::forward<F>(f)});
 }
 
 template <stdx::ct_string Name = "upon_stopped", sender S, stdx::callable F>
@@ -196,7 +189,7 @@ template <stdx::ct_string Name = "upon_stopped", sender S, stdx::callable F>
 template <stdx::ct_string Name = "then_error", stdx::callable... Fs>
 [[nodiscard]] constexpr auto then_error(Fs &&...fs) {
     return compose(
-        _then::pipeable<Name, set_value_t, set_error_t,
+        _then::pipeable<Name, stdx::type_list<set_value_t>, set_error_t,
                         std::remove_cvref_t<Fs>...>{std::forward<Fs>(fs)...});
 }
 
@@ -208,7 +201,7 @@ template <stdx::ct_string Name = "then_error", sender S, stdx::callable... Fs>
 template <stdx::ct_string Name = "transform_error", stdx::callable... Fs>
 [[nodiscard]] constexpr auto transform_error(Fs &&...fs) {
     return compose(
-        _then::pipeable<Name, set_error_t, set_error_t,
+        _then::pipeable<Name, stdx::type_list<set_error_t>, set_error_t,
                         std::remove_cvref_t<Fs>...>{std::forward<Fs>(fs)...});
 }
 
@@ -218,46 +211,93 @@ template <stdx::ct_string Name = "transform_error", sender S,
     return std::forward<S>(s) | transform_error<Name>(std::forward<Fs>(fs)...);
 }
 
+template <
+    auto Channels = stdx::type_list<set_value_t, set_error_t, set_stopped_t>{},
+    stdx::ct_string Name = "multithen", stdx::callable... Fs>
+[[nodiscard]] constexpr auto multithen(Fs &&...fs) {
+    return compose(
+        _then::pipeable<Name, std::remove_cvref_t<decltype(Channels)>,
+                        set_value_t, std::remove_cvref_t<Fs>...>{
+            std::forward<Fs>(fs)...});
+}
+
+template <
+    auto Channels = stdx::type_list<set_value_t, set_error_t, set_stopped_t>{},
+    stdx::ct_string Name = "multithen", sender S, stdx::callable... Fs>
+[[nodiscard]] constexpr auto multithen(S &&s, Fs &&...fs) -> sender auto {
+    return std::forward<S>(s) |
+           multithen<Channels, Name>(std::forward<Fs>(fs)...);
+}
+
+template <channel_tag T, channel_tag U>
+[[nodiscard]] consteval auto operator|(T, U)
+    -> boost::mp11::mp_unique<stdx::type_list<T, U>> {
+    return {};
+}
+
+template <channel_tag T, channel_tag... Us>
+[[nodiscard]] consteval auto operator|(T, stdx::type_list<Us...>)
+    -> boost::mp11::mp_unique<stdx::type_list<T, Us...>> {
+    return {};
+}
+
+template <channel_tag T, channel_tag... Us>
+[[nodiscard]] consteval auto operator|(stdx::type_list<Us...>, T)
+    -> boost::mp11::mp_unique<stdx::type_list<T, Us...>> {
+    return {};
+}
+
+template <channel_tag... Ts, channel_tag... Us>
+[[nodiscard]] consteval auto operator|(stdx::type_list<Ts...>,
+                                       stdx::type_list<Us...>)
+    -> boost::mp11::mp_unique<stdx::type_list<Ts..., Us...>> {
+    return {};
+}
+
 struct then_t;
 struct upon_error_t;
 struct upon_stopped_t;
 
 struct then_error_t;
 struct transform_error_t;
 
+struct multithen_t;
+
 namespace _then {
 namespace detail {
-template <typename HandleTag, typename CompleteTag> struct debug_tag;
+template <typename HandleTags, typename CompleteTag> struct debug_tag {
+    using type = multithen_t;
+};
 
-template <> struct debug_tag<set_value_t, set_value_t> {
+template <> struct debug_tag<stdx::type_list<set_value_t>, set_value_t> {
     using type = then_t;
 };
-template <> struct debug_tag<set_error_t, set_value_t> {
+template <> struct debug_tag<stdx::type_list<set_error_t>, set_value_t> {
     using type = upon_error_t;
 };
-template <> struct debug_tag<set_stopped_t, set_value_t> {
+template <> struct debug_tag<stdx::type_list<set_stopped_t>, set_value_t> {
     using type = upon_stopped_t;
 };
 
-template <> struct debug_tag<set_value_t, set_error_t> {
+template <> struct debug_tag<stdx::type_list<set_value_t>, set_error_t> {
     using type = then_error_t;
 };
-template <> struct debug_tag<set_error_t, set_error_t> {
+template <> struct debug_tag<stdx::type_list<set_error_t>, set_error_t> {
     using type = transform_error_t;
 };
 
-template <typename HandleTag, typename CompleteTag>
-using debug_tag_t = typename debug_tag<HandleTag, CompleteTag>::type;
+template <typename HandleTags, typename CompleteTag>
+using debug_tag_t = typename debug_tag<HandleTags, CompleteTag>::type;
 } // namespace detail
 } // namespace _then
 
-template <stdx::ct_string Name, typename HandleTag, typename CompleteTag,
+template <stdx::ct_string Name, typename HandleTags, typename CompleteTag,
           typename... Ts>
 struct debug::context_for<
-    _then::receiver<Name, HandleTag, CompleteTag, Ts...>> {
-    using tag = _then::detail::debug_tag_t<HandleTag, CompleteTag>;
+    _then::receiver<Name, HandleTags, CompleteTag, Ts...>> {
+    using tag = _then::detail::debug_tag_t<HandleTags, CompleteTag>;
     constexpr static auto name = Name;
-    using type = _then::receiver<Name, HandleTag, CompleteTag, Ts...>;
+    using type = _then::receiver<Name, HandleTags, CompleteTag, Ts...>;
     using children = stdx::type_list<debug::erased_context_for<
         connect_result_t<typename type::sender_t &&, type &&>>>;
 };

include/async/type_traits.hpp

@@ -28,8 +28,17 @@ template <typename... Tags> struct with_any_tag {
         std::bool_constant<(... or std::is_same_v<Tags, stdx::return_t<Sig>>)>;
 };
 
+template <typename Sigs, typename Tag> struct signatures_by_tag_t {
+    using type = boost::mp11::mp_copy_if_q<Sigs, with_any_tag<Tag>>;
+};
+
+template <typename Sigs, typename... Tags>
+struct signatures_by_tag_t<Sigs, with_any_tag<Tags...>> {
+    using type = boost::mp11::mp_copy_if_q<Sigs, with_any_tag<Tags...>>;
+};
+
 template <typename Sigs, typename Tag>
-using signatures_by_tag = boost::mp11::mp_copy_if_q<Sigs, with_any_tag<Tag>>;
+using signatures_by_tag = typename signatures_by_tag_t<Sigs, Tag>::type;
 
 template <bool> struct indirect_meta_apply {
     template <template <typename...> typename T, typename... As>

test/then.cpp

@@ -8,6 +8,7 @@
 #include <async/just_result_of.hpp>
 #include <async/schedulers/inline_scheduler.hpp>
 #include <async/then.hpp>
+#include <async/variant_sender.hpp>
 
 #include <catch2/catch_test_macros.hpp>
 
@@ -297,3 +298,45 @@ TEST_CASE("then calls functions by need", "[then]") {
     async::start(op);
     CHECK(value == 5 + 7);
 }
+
+TEST_CASE("multithen advertises what it sends", "[then]") {
+    auto const i = 0;
+    auto const s = async::make_variant_sender(
+        i == 0, [] { return async::just(42.0); },
+        [] { return async::just_error(17.0f); });
+
+    [[maybe_unused]] auto n = s | async::multithen([](auto) { return 42; });
+
+    STATIC_CHECK(
+        std::is_same_v<async::completion_signatures_of_t<decltype(n)>,
+                       async::completion_signatures<async::set_value_t(int)>>);
+}
+
+TEST_CASE("multithen handles multiple channels (value)", "[then]") {
+    int value{};
+    constexpr auto channels =
+        async::set_value | async::set_error | async::set_stopped;
+    auto n = async::multithen<channels>([](auto...) { return 42; });
+
+    {
+        value = 0;
+        auto op = async::connect(async::just(17) | n,
+                                 receiver{[&](auto i) { value = i; }});
+        async::start(op);
+        CHECK(value == 42);
+    }
+    {
+        value = 0;
+        auto op = async::connect(async::just_error(17) | n,
+                                 receiver{[&](auto i) { value = i; }});
+        async::start(op);
+        CHECK(value == 42);
+    }
+    {
+        value = 0;
+        auto op = async::connect(async::just_stopped() | n,
+                                 receiver{[&](auto i) { value = i; }});
+        async::start(op);
+        CHECK(value == 42);
+    }
+}