diff --git a/include/exec/sequence_senders.hpp b/include/exec/sequence_senders.hpp index 77b904f07..2c1919c86 100644 --- a/include/exec/sequence_senders.hpp +++ b/include/exec/sequence_senders.hpp @@ -35,6 +35,7 @@ import stdexec; # include "../stdexec/__detail/__receivers.hpp" # include "../stdexec/__detail/__senders.hpp" # include "../stdexec/__detail/__stop_token.hpp" +# include "../stdexec/__detail/__subscribe.hpp" # include "../stdexec/__detail/__transform_sender.hpp" # include "../stdexec/__detail/__type_traits.hpp" # include "../stdexec/__detail/__utility.hpp" @@ -880,18 +881,6 @@ namespace experimental::execution && sequence_receiver_of<_Receiver, item_types>> && sender_to, __stopped_means_break<_Receiver>>; - template - using __subscribe_member_result_t = decltype(__declval<_Sequence>().subscribe( - __declval<_Receiver>())); - - template - concept __subscribable_with_member = - __minvocable_q<__subscribe_member_result_t, _Sequence, _Receiver>; - - template - using __subscribe_static_member_result_t = decltype(STDEXEC_REMOVE_REFERENCE( - _Sequence)::__static_subscribe(__declval<_Sequence>(), __declval<_Receiver>())); - // Receiver wrapper used to subscribe to the child of a transparent // sequence adaptor (see `__sequence_adaptor_traits` above): it forwards // all completions and `set_next` calls to the wrapped receiver, but @@ -939,10 +928,6 @@ namespace experimental::execution _Env __env_; }; - template - concept __subscribable_with_static_member = - __minvocable_q<__subscribe_static_member_result_t, _Sequence, _Receiver>; - struct subscribe_t { private: @@ -1030,21 +1015,12 @@ namespace experimental::execution return __declfn<__result_t, __nothrow_subscribe && __nothrow_child_env && __nothrow_tfx_seq>(); } - else if constexpr (__subscribable_with_static_member<__tfx_seq_t, _Receiver>) + else if constexpr (STDEXEC::__subscribable<__tfx_seq_t, _Receiver>) { - using __result_t = decltype(STDEXEC_REMOVE_REFERENCE( - __tfx_seq_t)::__static_subscribe(__declval<__tfx_seq_t>(), __declval<_Receiver>())); + using __result_t = STDEXEC::__subscribe_result_t<__tfx_seq_t, _Receiver>; __check_operation_state<__result_t>(); - constexpr bool __nothrow_subscribe = noexcept(STDEXEC_REMOVE_REFERENCE( - __tfx_seq_t)::__static_subscribe(__declval<__tfx_seq_t>(), __declval<_Receiver>())); - return __declfn<__result_t, __nothrow_subscribe && __nothrow_tfx_seq>(); - } - else if constexpr (__subscribable_with_member<__tfx_seq_t, _Receiver>) - { - using __result_t = decltype(__declval<__tfx_seq_t>().subscribe(__declval<_Receiver>())); - __check_operation_state<__result_t>(); - constexpr bool __nothrow_subscribe = noexcept( - __declval<__tfx_seq_t>().subscribe(__declval<_Receiver>())); + constexpr bool __nothrow_subscribe = + __nothrow_callable; return __declfn<__result_t, __nothrow_subscribe && __nothrow_tfx_seq>(); } else if constexpr (__is_debug_env>) @@ -1103,14 +1079,10 @@ namespace experimental::execution __xform_t{}(static_cast(__env), STDEXEC::__forward_like<__tfx_seq_t>(__data))}); } - else if constexpr (__subscribable_with_static_member<__tfx_seq_t, _Receiver>) - { // NOLINT(bugprone-branch-clone) - return __tfx_seq.__static_subscribe(static_cast<__tfx_seq_t&&>(__tfx_seq), - static_cast<_Receiver&&>(__rcvr)); - } - else if constexpr (__subscribable_with_member<__tfx_seq_t, _Receiver>) + else if constexpr (STDEXEC::__subscribable<__tfx_seq_t, _Receiver>) { - return static_cast<__tfx_seq_t&&>(__tfx_seq).subscribe(static_cast<_Receiver&&>(__rcvr)); + return STDEXEC::__subscribe(static_cast<__tfx_seq_t&&>(__tfx_seq), + static_cast<_Receiver&&>(__rcvr)); } else if constexpr (enable_sequence_sender>) { diff --git a/include/stdexec/__detail/__subscribe.hpp b/include/stdexec/__detail/__subscribe.hpp new file mode 100644 index 000000000..6097dbfd4 --- /dev/null +++ b/include/stdexec/__detail/__subscribe.hpp @@ -0,0 +1,91 @@ +/* + * Copyright (c) 2026 NVIDIA Corporation + * + * Licensed under the Apache License Version 2.0 with LLVM Exceptions + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://llvm.org/LICENSE.txt + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include "__config.hpp" + +#if STDEXEC_USE_MODULES() && !defined(STDEXEC_IN_MODULE_PURVIEW) + +import stdexec; + +#else + +# include "__execution_fwd.hpp" + +# include "__concepts.hpp" +# include "__type_traits.hpp" +# include "__utility.hpp" + +# include "__prologue.hpp" + +// NOLINTNEXTLINE(bugprone-std-namespace-modification, modernize-concat-nested-namespaces) +namespace STDEXEC +{ + namespace __subscribe_ + { + template + concept __has_static_memfn = requires(_Sender && (*__sndr)(), _Receiver && (*__rcvr)()) { + STDEXEC_REMOVE_REFERENCE(_Sender)::__static_subscribe(__sndr(), __rcvr()); + }; + + template + concept __has_memfn = requires(_Sender && (*__sndr)(), _Receiver && (*__rcvr)()) { + __sndr().subscribe(__rcvr()); + }; + + // The customization point behind exec::subscribe. A sender hooks it with a + // static __static_subscribe(sndr, rcvr) member or a subscribe(rcvr) member + // function, the static one taking precedence. Senders defined in stdexec can + // use this to behave as sequence senders without depending on exec. + struct __subscribe_t + { + template + requires __has_static_memfn<_Sender, _Receiver> + STDEXEC_ATTRIBUTE(always_inline) + constexpr auto operator()(_Sender&& __sndr, _Receiver __rcvr) const + noexcept(noexcept(__sndr.__static_subscribe(__declval<_Sender>(), __declval<_Receiver>()))) + -> decltype(__sndr.__static_subscribe(__declval<_Sender>(), __declval<_Receiver>())) + { + return __sndr.__static_subscribe(static_cast<_Sender&&>(__sndr), + static_cast<_Receiver&&>(__rcvr)); + } + + template + requires(!__has_static_memfn<_Sender, _Receiver>) && __has_memfn<_Sender, _Receiver> + STDEXEC_ATTRIBUTE(always_inline) + constexpr auto operator()(_Sender&& __sndr, _Receiver __rcvr) const + noexcept(noexcept(__declval<_Sender>().subscribe(__declval<_Receiver>()))) + -> decltype(__declval<_Sender>().subscribe(__declval<_Receiver>())) + { + return static_cast<_Sender&&>(__sndr).subscribe(static_cast<_Receiver&&>(__rcvr)); + } + }; + } // namespace __subscribe_ + + STDEXEC_MODULE_EXPORT_AUTHORING + inline constexpr __subscribe_::__subscribe_t __subscribe{}; + + STDEXEC_MODULE_EXPORT_AUTHORING + template + using __subscribe_result_t = __call_result_t<__subscribe_::__subscribe_t, _Sender, _Receiver>; + + STDEXEC_MODULE_EXPORT_AUTHORING + template + concept __subscribable = __callable<__subscribe_::__subscribe_t, _Sender, _Receiver>; +} // namespace STDEXEC + +# include "__epilogue.hpp" +#endif // !STDEXEC_USE_MODULES() || defined(STDEXEC_IN_MODULE_PURVIEW) diff --git a/include/stdexec/execution.hpp b/include/stdexec/execution.hpp index 8db3a1f51..4438a6042 100644 --- a/include/stdexec/execution.hpp +++ b/include/stdexec/execution.hpp @@ -63,6 +63,7 @@ import stdexec; # include "__detail/__stopped_as_error.hpp" # include "__detail/__stopped_as_optional.hpp" # include "__detail/__submit.hpp" +# include "__detail/__subscribe.hpp" # include "__detail/__sync_wait.hpp" # include "__detail/__task.hpp" # include "__detail/__task_scheduler.hpp" diff --git a/test/exec/test_sequence_senders.cpp b/test/exec/test_sequence_senders.cpp index 0ab6ad859..624337916 100644 --- a/test/exec/test_sequence_senders.cpp +++ b/test/exec/test_sequence_senders.cpp @@ -151,4 +151,44 @@ namespace STATIC_REQUIRE(sequence_sender_to); STATIC_REQUIRE(sequence_sender_to, next_receiver_t>); } + + struct tagged_operation + { + int tag; + + void start() & noexcept {} + }; + + template <__completion_signature... _Sigs> + struct static_subscribe_sequence_of : some_sequence_sender_of<_Sigs...> + { + template + static auto __static_subscribe(Self&&, R&&) noexcept -> tagged_operation + { + return {1}; + } + + template + auto subscribe(R&&) const -> tagged_operation + { + return {2}; + } + }; + + TEST_CASE("sequence_senders - __subscribe hooks", "[sequence_senders]") + { + using next_receiver_t = next_receiver; + using member_seq_t = some_sequence_sender_of; + using static_seq_t = static_subscribe_sequence_of; + + STATIC_REQUIRE(__subscribable); + STATIC_REQUIRE(__same_as<__subscribe_result_t, nop_operation>); + STATIC_REQUIRE_FALSE(__subscribable); + + // the static hook wins over the member function + STATIC_REQUIRE(__subscribable); + STATIC_REQUIRE(noexcept(STDEXEC::__subscribe(static_seq_t{}, next_receiver_t{}))); + CHECK(STDEXEC::__subscribe(static_seq_t{}, next_receiver_t{}).tag == 1); + CHECK(exec::subscribe(static_seq_t{}, next_receiver_t{}).tag == 1); + } } // namespace