reseiver.cpp

// Prototype partial implementation of the proposed <execution> header,
// from P0443R12 + P2006R0
#include <https://gist.githubusercontent.com/ericniebler/8cc25656a0a496bd682edc8314d9576b/raw/2c20ea65a5f2ffe625e05a77c7ba30e57a0f0ffd/execution.h>
#include <compare>
#include <cstdio>
#include <optional>

namespace ex = std::execution;

/////////////////////////////////////////////////////////////////////////
// retry algorithm implementation begins here

// _conv needed so we can emplace construct non-movable types into
// a std::optional.
template<std::invocable F>
    requires std::is_nothrow_move_constructible_v<F>
struct _conv {
    F f_;
    explicit _conv(F f) noexcept : f_((F&&) f) {}
    operator std::invoke_result_t<F>() && {
        return ((F&&) f_)();
    }
};

// pass through set_value and set_error, but retry the operation
// from set_error.
template<class O, class R>
struct _retry_receiver {
    O* o_;
    template<class... As>
        requires ex::receiver_of<R, As...>
    void set_value(As&&... as) &&
        noexcept(ex::is_nothrow_receiver_of_v<R, As...>) {
        ex::set_value(std::move(o_->r_), (As&&) as...);
    }
    void set_error(auto&&) && noexcept {
        o_->_retry(); // This causes the op to be retried
    }
    void set_done() && noexcept {
        ex::set_done(std::move(o_->r_));
    }
};

template<ex::sender S>
struct _retry_sender : ex::sender_base {
    S s_;
    // Hold the nested operation state in an optional so we can
    // re-construct and re-start it when the operation fails.
    template<ex::receiver R>
    struct _op {
        S s_;
        R r_;
        std::optional<ex::connect_result_t<S&, _retry_receiver<_op, R>>> o_;

        _op(S s, R r): s_((S&&)s), r_((R&&)r), o_{_connect()} {}
        _op(_op&&) = delete;

        auto _connect() noexcept {
            return _conv{[this] {
                return ex::connect(s_, _retry_receiver<_op, R>{this});
            }};
        }
        void _retry() noexcept try {
            o_.emplace(_connect()); // potentially throwing
            ex::start(*o_);
        } catch(...) {
            ex::set_error((R&&) r_, std::current_exception());
        }
        void start() noexcept {
            ex::start(*o_);
        }
    };

    template<ex::receiver R>
        requires ex::sender_to<S&, _retry_receiver<_op<R>, R>>
    auto connect(R r) && -> _op<R> {
        return _op<R>{(S&&) s_, (R&&) r};
    }
};

template<ex::sender S>
ex::sender auto retry(S s) {
    return _retry_sender<S>{{}, (S&&)s};
}

// retry algorithm implementation ends here
/////////////////////////////////////////////////////////////////////////

inline constexpr struct _sink {
    void set_value(auto&&...) const noexcept {}
    [[noreturn]] void set_error(auto&&) const noexcept {
        std::terminate();
    }
    [[noreturn]] void set_done() const noexcept {
        std::terminate();
    }
} sink{};

// Here is a test sender that fails the first three times it is
// started and then succeeds on the fourth try.
struct fail_3 : ex::sender_base {
    int count_ = 0;

    template<ex::receiver_of R>
    struct _op {
        int const count_;
        R r_;
        _op(int count, R r): count_(count), r_((R&&) r) {}
        _op(_op&&) = delete;
        void start() noexcept try {
            if(count_ > 3) {
                std::puts("success");
                ex::set_value((R&&) r_, count_);
            } else {
                std::puts("error");
                ex::set_error((R&&) r_, 42);
            }
        } catch(...) {
            ex::set_error((R&&) r_, std::current_exception());
        }
    };

    template<ex::receiver_of R>
        requires ex::receiver<R, int>
    auto connect(R r) & -> _op<R> {
        return _op<R>{++count_, (R&&) r};
    }
};

int main() {
    fail_3 s;
    auto op = ex::connect(retry(s), sink);
    ex::start(op);
}