senders_and_receivers_error.cpp

#include <thread>

auto async_work() {
	return [](auto p) {
		std::thread { [p=std::move(p)]() mutable {
			std::exception_ptr ep;
			try {
				throw std::runtime_error("Test Error");
			} catch (...) {
				ep = std::current_exception();
				p.set_error(ep);
			}
			return 42;
		}}.detach();
	};
}

template <typename P, typename N>
struct then_sender {
	P p;
	N next;
	
	void set_value(auto ...args) {
		p.set_value(next(args...));
	}
	
	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 state {
	std::mutex mtx;
	std::condition_variable cv;
	std::variant<std::monostate, std::exception_ptr, T> data;
};

template <typename T>
struct promise {
	state<T> *ps;
	
	// set
	template <int I>
	void _set(auto ...args) {
		std::unique_lock lck{ps->mtx};
		ps->data.template emplace<I>(args...);
		ps->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) {
	state<T> s;
	
	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_work();
	auto next = then(start, [](int i) {
		return i / 2;
	});
	
	try {
		printf("%d\n", sync_wait<int>(next));
	} catch (const std::exception& e) {
		printf("%s\n", e.what());
	}
	
	return 0;
}