senders_and_receivers.cpp

/*
 * This is all Eric Nieblers work, see: https://youtu.be/h-ExnuD6jms
 */
 
#include <thread>
#include <vector>

auto new_thread() {
    return [](auto p) {
        std::thread { [p=std::move(p)]() mutable {
            p.set_value();
        }}.detach();
    };
}

auto async_algo(auto executor) {
    return then(executor, []() {
        return 42;
    });
}

template <typename P, typename N>
struct then_sender {
    P p;
    N next;

    void set_value(auto ...args) {
        try {
            p.set_value(next(args...));
        } catch (...) {
            p.set_error(std::current_exception());
        }
    }

    void set_error(auto e) {
        p.set_error(e);
    }
};

auto then(auto task, auto next) {
    return [=](auto p){
        task(then_sender<decltype(p), decltype(next)>{p, next});
    };
}

template <typename T>
struct shared {
    std::mutex mtx;
    std::condition_variable cv;
    std::variant<std::monostate, std::exception_ptr, T> data;
};

template <typename T>
struct promise {
    shared<T>* s;

    // set
    template <int I>
    void _set(auto ...args) {
        std::unique_lock lck{s->mtx};
        s->data.template emplace<I>(args...);
        s->cv.notify_one();
    }

    void set_value(auto ...args) {
        _set<2>(args...);
    }

    void set_error(auto e) {
        _set<1>(e);
    }
};

template <typename T, typename Task>
T sync_wait(Task task) {
    shared<T> s;

    // start
    task(promise<T>{&s});

    // get
    {
        std::unique_lock lck{s.mtx};
        s.cv.wait(lck, [&s]() { return s.data.index() != 0; });
    }

    if (s.data.index() == 1) std::rethrow_exception( std::get<1>(s.data) );

    return std::move(std::get<2>(s.data));
}

int main() {
    auto start = async_algo(new_thread());
    auto next = then(start, [](int i) {
        return i / 2;
    });

    printf("%d\n", sync_wait<int>(next));

    return 0;
}