🎨 Add terse forms of *_until

Problem:
- A common use case for the predicate passed to `repeat_until` is simply to
  repeat until the argument passed is equal to some value. e.g.

```cpp
auto s = some_sender | async::repeat_until([] (auto const &x) { return x == 42; });
```

or, even more simply,

```cpp
auto s = some_sender |
         async::repeat_until([] (std::convertible_to<bool> auto const &x) { return x; });
```

and this is quite verbose.

Solution:
- Allow `repeat_until` to take a value as a terse way to express this.

```cpp
auto s = some_sender | async::repeat_until(42);
```

or

```cpp
auto s = some_sender | async::repeat_until(true);
```

Notes:
- As for `repeat_until`, so for `retry_until` and `periodic_until`.

Author: elbeno

docs/sender_adaptors.adoc

@@ -296,6 +296,18 @@ NOTE: `periodic` never completes other than by error or cancellation, but
 `periodic_n` and `periodic_until` both complete successfully with the same
 completion as the adapted sender.
 
+`periodic_until` can take a plain value rather than a predicate, and this is a
+terse way to repeat until the argument is equal to that value.
+
+[source,cpp]
+----
+// these senders are equivalent
+auto s1 = some_sender | async::periodic_until(1s, [] (auto const&x) { return x == 42; });
+auto s2 = some_sender | async::periodic_until(1s, 42);
+----
+
+NOTE: A "plain value" given to `periodic_until` must be movable.
+
 === `repeat`
 
 Found in the header: `async/repeat.hpp`
@@ -375,6 +387,18 @@ NOTE: `repeat` never completes other than by error or cancellation, but
 `repeat_n` and `repeat_until` both complete successfully with the same
 completion as the adapted sender.
 
+`repeat_until` can take a plain value rather than a predicate, and this is a
+terse way to repeat until the argument is equal to that value.
+
+[source,cpp]
+----
+// these senders are equivalent
+auto s1 = some_sender | async::repeat_until([] (auto const&x) { return x == 42; });
+auto s2 = some_sender | async::repeat_until(42);
+----
+
+NOTE: A "plain value" given to `repeat_until` must be movable.
+
 === `retry`
 
 Found in the header: `async/retry.hpp`
@@ -409,6 +433,19 @@ auto s = some_sender | async::retry_until([] (auto&&) { return true; });
 NOTE: The arguments passed to the predicate are those in the error completion(s)
 of the sender.
 
+Like `repeat_until`, `retry_until` can take a plain value rather than a
+predicate, and this is a terse way to retry until the (error channel) argument
+is equal to that value.
+
+[source,cpp]
+----
+// these senders are equivalent
+auto s1 = some_sender | async::retry_until([] (auto const&x) { return x == 42; });
+auto s2 = some_sender | async::retry_until(42);
+----
+
+NOTE: A "plain value" given to `retry_until` must be movable.
+
 === `sequence`
 
 Found in the header: `async/sequence.hpp`

include/async/periodic.hpp

@@ -238,18 +238,43 @@ struct sender {
     }
 };
 
+template <stdx::ct_string, typename, typename, typename> struct pipeable;
+
+template <stdx::ct_string Name, typename Expiry, typename... Args>
+constexpr auto make_sender(Args &&...args)
+    -> sender<Name, Expiry, std::remove_cvref_t<Args>...> {
+    return {std::forward<Args>(args)...};
+}
+
 template <stdx::ct_string Name, typename Expiry, typename Duration,
-          stdx::callable Pred>
+          typename MatchValue>
 struct pipeable {
     Duration d;
-    Pred p;
+    MatchValue v;
 
   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, Expiry, std::remove_cvref_t<S>, Duration, Pred>{
+        return make_sender<Name, Expiry>(
             std::forward<S>(s), std::forward<Self>(self).d,
-            std::forward<Self>(self).p};
+            [value = std::forward<Self>(self).v](auto const &x) {
+                return x == value;
+            });
+    }
+};
+
+template <stdx::ct_string Name, typename Expiry, typename Duration,
+          stdx::callable Pred>
+struct pipeable<Name, Expiry, Duration, Pred> {
+    Duration d;
+    Pred p;
+
+  private:
+    template <async::sender S, stdx::same_as_unqualified<pipeable> Self>
+    friend constexpr auto operator|(S &&s, Self &&self) -> async::sender auto {
+        return make_sender<Name, Expiry>(std::forward<S>(s),
+                                         std::forward<Self>(self).d,
+                                         std::forward<Self>(self).p);
     }
 };
 } // namespace _periodic

include/async/repeat.hpp

@@ -190,23 +190,48 @@ struct sender {
     }
 };
 
-template <stdx::ct_string Name, stdx::callable Pred, stdx::callable LoopFn>
+template <stdx::ct_string, typename, stdx::callable> struct pipeable;
+
+template <stdx::ct_string Name, typename... Args>
+constexpr auto make_sender(Args &&...args)
+    -> sender<Name, std::remove_cvref_t<Args>...> {
+    return {std::forward<Args>(args)...};
+}
+
+template <stdx::ct_string Name, typename MatchValue, stdx::callable LoopFn>
 struct pipeable {
+    MatchValue v;
+    LoopFn f;
+
+  private:
+    template <async::sender S, stdx::same_as_unqualified<pipeable> Self>
+    friend constexpr auto operator|(S &&s, Self &&self) -> async::sender auto {
+        return make_sender<Name>(
+            std::forward<S>(s),
+            [value = std::forward<Self>(self).v](auto const &x) {
+                return x == value;
+            },
+            std::forward<Self>(self).f);
+    }
+};
+
+template <stdx::ct_string Name, stdx::callable Pred, stdx::callable LoopFn>
+struct pipeable<Name, Pred, LoopFn> {
     Pred p;
     LoopFn f;
 
   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, std::remove_cvref_t<S>, Pred, LoopFn>{
-            std::forward<S>(s), std::forward<Self>(self).p,
-            std::forward<Self>(self).f};
+        return make_sender<Name>(std::forward<S>(s), std::forward<Self>(self).p,
+                                 std::forward<Self>(self).f);
     }
 };
 } // namespace _repeat
 
 template <stdx::ct_string Name = "repeat_until", typename P,
           typename F = std::remove_cvref_t<decltype(_repeat::no_loop_fn)>>
+    requires(not sender<P>)
 [[nodiscard]] constexpr auto repeat_until(P &&p, F &&f = {})
     -> _repeat::pipeable<Name, std::remove_cvref_t<P>, std::remove_cvref_t<F>> {
     return {std::forward<P>(p), std::forward<F>(f)};

include/async/retry.hpp

@@ -180,14 +180,35 @@ template <stdx::ct_string Name, typename Sndr, typename Pred> struct sender {
     }
 };
 
-template <stdx::ct_string Name, typename Pred> struct pipeable {
+template <stdx::ct_string, typename> struct pipeable;
+
+template <stdx::ct_string Name, typename... Args>
+constexpr auto make_sender(Args &&...args)
+    -> sender<Name, std::remove_cvref_t<Args>...> {
+    return {std::forward<Args>(args)...};
+}
+
+template <stdx::ct_string Name, typename MatchValue> struct pipeable {
+    MatchValue v;
+
+  private:
+    template <async::sender S, stdx::same_as_unqualified<pipeable> Self>
+    friend constexpr auto operator|(S &&s, Self &&self) -> async::sender auto {
+        return make_sender<Name>(std::forward<S>(s),
+                                 [value = std::forward<Self>(self).v](
+                                     auto const &x) { return x == value; });
+    }
+};
+
+template <stdx::ct_string Name, stdx::callable Pred>
+struct pipeable<Name, Pred> {
     Pred p;
 
   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, std::remove_cvref_t<S>, Pred>{
-            std::forward<S>(s), std::forward<Self>(self).p};
+        return make_sender<Name>(std::forward<S>(s),
+                                 std::forward<Self>(self).p);
     }
 };
 } // namespace _retry

test/periodic.cpp

@@ -132,6 +132,23 @@ TEST_CASE("periodic repeats periodically", "[periodic]") {
     CHECK(var == 42);
 }
 
+TEST_CASE("periodic_until terse form", "[periodic]") {
+    int var{};
+    [[maybe_unused]] auto s = async::time_scheduler{}.schedule() |
+                              async::then([&] { return ++var; }) |
+                              async::periodic_until(1s, 2);
+    auto op = async::connect(s, receiver{[&](auto) { var = 42; }});
+    async::start(op);
+    CHECK(enabled<default_domain>);
+    CHECK(not async::timer_mgr::is_idle());
+    async::timer_mgr::service_task();
+    CHECK(var == 1);
+    CHECK(not async::timer_mgr::is_idle());
+    async::timer_mgr::service_task();
+    CHECK(async::timer_mgr::is_idle());
+    CHECK(var == 42);
+}
+
 TEST_CASE("periodic allows continue_on another scheduler", "[periodic]") {
     int var{};
     [[maybe_unused]] auto s =

test/repeat.cpp

@@ -130,6 +130,19 @@ TEST_CASE("repeat_until is pipeable", "[repeat]") {
     CHECK(var == 43);
 }
 
+TEST_CASE("repeat_until terse form", "[repeat]") {
+    int var{};
+
+    auto sub = async::just() | async::sequence([&] {
+                   ++var;
+                   return async::just(42);
+               });
+    auto s = sub | async::repeat_until(42);
+    auto op = async::connect(s, receiver{[&](auto i) { var += i; }});
+    async::start(op);
+    CHECK(var == 43);
+}
+
 TEST_CASE("repeat can be cancelled", "[repeat]") {
     int var{};
     stoppable_receiver r{[&] { var += 42; }};

test/retry.cpp

@@ -103,6 +103,19 @@ TEST_CASE("retry_until retries on error", "[retry]") {
     CHECK(var == 3);
 }
 
+TEST_CASE("retry_until terse form", "[retry]") {
+    int var{};
+
+    auto sub = async::just() | async::sequence([&] {
+                   ++var;
+                   return async::just_error(var);
+               });
+    auto s = sub | async::retry_until(3);
+    auto op = async::connect(s, receiver{[] {}});
+    async::start(op);
+    CHECK(var == 3);
+}
+
 TEST_CASE("retry can be cancelled", "[retry]") {
     int var{};
     stoppable_receiver r{[&] { var += 42; }};