coro_utils.hpp

 
#pragma once
#include <ranges>
#include <type_traits>
#include <utility>
#include <vector>
 
#include <coro/coro.hpp>
 
namespace rapidsmpf::streaming {
 
template <std::ranges::range Range>
auto coro_results(Range&& task_results) {
    using first_ref_t = std::ranges::range_value_t<Range>;
    using raw_ret_t = decltype(std::declval<first_ref_t>().return_value());
    using val_t = std::remove_cvref_t<raw_ret_t>;
 
    if constexpr (std::is_void_v<val_t>) {
        // Just surface exceptions; return void.
        for (auto&& r : task_results) {
            r.return_value();
        }
    } else {
        std::vector<val_t> ret;
        if constexpr (std::ranges::sized_range<Range>) {
            ret.reserve(task_results.size());
        }
        for (auto&& r : task_results) {
            ret.emplace_back(r.return_value());
        }
        return ret;
    }
}
 
template <typename... Args>
auto coro_results(std::tuple<Args...>&& results) {
    return std::apply(
        [](auto&&... result) {
            if constexpr ((std::is_void_v<std::remove_cvref_t<
                               decltype(std::declval<
                                            std::remove_reference_t<decltype(result)>>()
                                            .return_value())>>
                           && ...))
            {
                (result.return_value(), ...);
            } else {
                return std::make_tuple(
                    std::forward<decltype(result)>(result).return_value()...
                );
            }
        },
        std::move(results)
    );
}
 
}  // namespace rapidsmpf::streaming