Cancellable sender in C++26 S/R library

cancellable-sender.cpp

#include <cassert>
#include <stop_token>
#include <utility>
#include <print>
// https://github.com/NVIDIA/stdexec/
#include <stdexec/execution.hpp>
namespace exec = stdexec;
namespace std {
  using exec::get_stop_token, exec::sync_wait;
  using exec::inplace_stop_source, exec::stoppable_token;
}

template<typename Rcvr, typename F>
struct OpState {
  using operation_state_concept = exec::operation_state_t;

  void start() & noexcept {
    auto st = std::get_stop_token(exec::get_env(rcvr_));
    if (st. stop_requested()) {
      exec::set_stopped(std::move(rcvr_));
      std::println("set_stopped");
      return;
    }
    try {
      exec::set_value(std::move(rcvr_), f_());
      std::println("set_value");
    } catch (...) {
      exec::set_error(std::move(rcvr_), std::current_exception());
      std::println("set_error");
    }
  }

  Rcvr rcvr_;
  F f_;
};

template<typename F>
struct MySender {
  using sender_concept = exec::sender_t;

  using result_type = std::invoke_result_t<F>;

  using completion_signatures = exec::completion_signatures<
    exec::set_value_t(result_type),
    exec::set_error_t(std::exception_ptr),
    exec::set_stopped_t()
  >;
  template <class Self>  // P3557R3
  static consteval auto get_completion_signatures() {
    return completion_signatures{};
  }

  template<typename Rcvr>
  auto connect(Rcvr rcvr) {
    return OpState{std::move(rcvr), std::move(f_)};
  }

  F f_;
};

template<exec::sender Sndr>
void invoke(Sndr&& sndr)
{
  try {
    auto result = std::sync_wait(std::forward<Sndr>(sndr));
    if (result) {
      std::println("val={}", *result);
    } else {
      std::println("<cancelled>");
    }
  } catch (int e) {
    std::println("catch {}", e);
  }
  std::println();
}

int main()
{
  invoke(MySender{[] -> int { return 1; }});

  invoke(MySender{[] -> int { throw 42; }});

  {
  exec::sender auto snd1 = MySender{[] -> int { return 1; }};
  exec::sender auto snd2 = MySender{[] -> int { throw 42; }};
#if 1
  exec::sender auto snd3 = MySender{[] -> int { return 3; }};
#else
  exec::sender auto snd3 = MySender{[] -> int { return 3; }}
    | exec::unstoppable;  // P3284R4
#endif
  exec::sender auto sndr = exec::when_all(snd1, snd2, snd3);
  invoke(std::move(sndr));
  }

  {
  std::inplace_stop_source ssource;
  std::stoppable_token auto stoken = ssource.get_token();
  // C++26 stoppable_token requires T::callback_type type,
  // but C++20/23 Spec. doesn't define such nested type.

  exec::sender auto sndr = MySender{[] -> int { return 1; }}
    | exec::write_env(exec::prop{std::get_stop_token, stoken}); // P3284R4

  ssource.request_stop();
  invoke(std::move(sndr));
  }
}

Result:

set_value
val=(1)

set_error
catch 42

set_value
set_error
set_stopped
catch 42

set_stopped
<cancelled>