Method __complete_

include/exec/async_scope.hpp:310–353

        stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>>, __forward_stopped>;

      constexpr void __complete_() noexcept
      {
        STDEXEC_TRY
        {
          __forward_consumer_.reset();
          auto __state = std::move(__state_);
          STDEXEC_ASSERT(__state != nullptr);
          std::unique_lock __guard{__state->__mutex_};
          // either the future is still in use or it has passed ownership to __state->__no_future_
          if (__state->__no_future_.get() != nullptr || __state->__step_ != __future_step::__future)
          {
            // invalid state - there is a code bug in the state machine
            std::terminate();
          }
          else if (get_stop_token(get_env(__rcvr_)).stop_requested())
          {
            __guard.unlock();
            STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_));
            __guard.lock();
          }
          else
          {
            std::visit(
              [this, &__guard]<class _Tup>(_Tup& __tup)
              {
                if constexpr (__std::same_as<_Tup, std::monostate>)
                {
                  std::terminate();
                }
                else
                {
                  std::apply(
                    [this, &__guard]<class... _As>(auto tag, _As&... __as)
                    {
                      __guard.unlock();
                      tag(static_cast<_Receiver&&>(__rcvr_), static_cast<_As&&>(__as)...);
                      __guard.lock();
                    },
                    __tup);
                }
              },
              __state->__data_);
          }
        }
        STDEXEC_CATCH_ALL
        {
          STDEXEC::set_error(static_cast<_Receiver&&>(__rcvr_), std::current_exception());