from.hpp

//                  ReactivePlusPlus library
//
//          Copyright Aleksey Loginov 2022 - present.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE_1_0.txt or copy at
//          https://www.boost.org/LICENSE_1_0.txt)
//
// Project home: https://github.com/victimsnino/ReactivePlusPlus
//
 
#pragma once
 
#include <rpp/memory_model.hpp>
#include <rpp/schedulers/trampoline_scheduler.hpp>
#include <rpp/sources/create.hpp>
#include <rpp/sources/fwd.hpp>
#include <rpp/utils/utilities.hpp>
#include <rpp/operators/map.hpp>
#include <rpp/utils/function_traits.hpp>
#include <rpp/defs.hpp>
 
 
#include <array>
#include <type_traits>
 
IMPLEMENTATION_FILE(just_tag);
IMPLEMENTATION_FILE(from_tag);
 
namespace rpp::observable::details
{
template<typename T>
auto extract_iterable_from_packed(const T & v) -> const auto&
{
    return v;
}
 
template<typename T>
auto extract_iterable_from_packed(const std::shared_ptr<T> & v) -> const auto&
{
    return *v;
}
 
void iterate(const auto&                                   iterable,
             const schedulers::constraint::scheduler auto& scheduler,
             constraint::subscriber auto&&                 subscriber)
{
    if constexpr (constraint::decayed_same_as<decltype(scheduler), schedulers::immediate>)
    {
        for (const auto& v : extract_iterable_from_packed(iterable))
        {
            if (subscriber.is_subscribed())
                subscriber.on_next(utils::as_const(v));
            else
                return;
        }
        subscriber.on_completed();
    }
    else
    {
        auto worker = scheduler.create_worker(subscriber.get_subscription());
        worker.schedule([iterable=iterable, 
                         subscriber=std::forward<decltype(subscriber)>(subscriber), 
                         index = size_t{0}]() mutable-> schedulers::optional_duration
        {            
            try
            {
                const auto& extracted_iterable = extract_iterable_from_packed(iterable);
                const auto  end = std::cend(extracted_iterable);
                auto        itr = std::cbegin(extracted_iterable);
 
                std::advance(itr, static_cast<int64_t>(index));
 
                if (itr != end)
                {
                    subscriber.on_next(utils::as_const(*itr));
                    if (std::next(itr) != end) // it was not last
                    {
                        ++index;
                        return schedulers::duration{}; // re-schedule this
                    }
                }
 
                subscriber.on_completed();
            }
            catch(...)
            {
                subscriber.on_error(std::current_exception());
            }
            return std::nullopt;
 
        });
    }
}
 
 
 
template<memory_model memory_model, constraint::iterable Container, typename ...Ts>
auto pack_to_container(Ts&& ...items)
{
    if constexpr (memory_model == memory_model::use_stack)
        return Container{std::forward<Ts>(items)...};
    else
        // raw new call to avoid extra copy/moves for items
        return std::shared_ptr<Container>(new Container{std::forward<Ts>(items)...});
}
 
template<memory_model memory_model, constraint::decayed_type T, typename ...Ts>
auto pack_variadic(Ts&& ...items)
{
    return pack_to_container<memory_model, std::array<T, sizeof...(Ts)>>(std::forward<Ts>(items)...);
}
 
template<typename PackedIterable, schedulers::constraint::scheduler TScheduler>
class iterate_impl
{
public:
    iterate_impl(const PackedIterable& iterable, const TScheduler& scheduler)
        : m_iterable{iterable}
        , m_scheduler{scheduler} {}
 
    iterate_impl(PackedIterable&& iterable, const TScheduler& scheduler)
        : m_iterable{std::move(iterable)}
        , m_scheduler{scheduler} {}
 
    template<constraint::subscriber TSub>
    void operator()(TSub&& subscriber) const
    {
        details::iterate(m_iterable, m_scheduler, std::forward<TSub>(subscriber));
    }
 
private:
    PackedIterable                   m_iterable;
    RPP_NO_UNIQUE_ADDRESS TScheduler m_scheduler;
};
} // namespace rpp::observable::details
 
namespace rpp::observable
{
template<memory_model memory_model /* = memory_model::use_stack */, typename T, typename ...Ts>
auto just(const schedulers::constraint::scheduler auto& scheduler, T&& item, Ts&& ...items) requires (rpp::details::is_header_included<rpp::details::just_tag, T, Ts...> && (constraint::decayed_same_as<T, Ts> && ...))
{
    return create<std::decay_t<T>>(details::iterate_impl{details::pack_variadic<memory_model, std::decay_t<T>>(std::forward<T>(item), std::forward<Ts>(items)...), scheduler });
}
 
template<memory_model memory_model /* = memory_model::use_stack */, typename T, typename ...Ts>
auto just(T&& item, Ts&& ...items) requires (rpp::details::is_header_included<rpp::details::just_tag, T, Ts...> && (constraint::decayed_same_as<T, Ts> && ...))
{
    return just<memory_model>(schedulers::trampoline{}, std::forward<T>(item), std::forward<Ts>(items)...);
}
 
template<memory_model memory_model /* = memory_model::use_stack */, schedulers::constraint::scheduler TScheduler /* = schedulers::trampoline */>
auto from_iterable(constraint::iterable auto&& iterable, const TScheduler& scheduler /* = TScheduler{} */) requires rpp::details::is_header_included<rpp::details::from_tag, TScheduler >
{
    using Container = std::decay_t<decltype(iterable)>;
    return create<utils::iterable_value_t<Container>>(details::iterate_impl{ details::pack_to_container<memory_model, Container>(std::forward<decltype(iterable)>(iterable)), scheduler });
}
 
template<memory_model memory_model /* = memory_model::use_stack */>
auto from_callable(std::invocable<> auto&& callable) requires rpp::details::is_header_included<rpp::details::from_tag, decltype(callable)>
{
    auto obs = just<memory_model>(std::forward<decltype(callable)>(callable));
 
    if constexpr (std::same_as<utils::decayed_invoke_result_t<decltype(callable)>, void>)
        return std::move(obs).map([](const auto& fn) { fn(); return utils::none{};});
    else
        return std::move(obs).map([](const auto& fn) { return fn(); });
}
} // namespace rpp::observable