gtransducer

-module(gtransducer).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([mapping/1, filtering/1, compose/2, reduce/4, parallel_reduce/7]).

-spec mapping(fun((GIP) -> GIQ)) -> fun((fun((GIR, GIQ) -> GIR)) -> fun((GIR, GIP) -> GIR)).
mapping(F) ->
    fun(Reducer) -> fun(Acc, X) -> Reducer(Acc, F(X)) end end.

-spec filtering(fun((GIV) -> boolean())) -> fun((fun((GIW, GIV) -> GIW)) -> fun((GIW, GIV) -> GIW)).
filtering(Pred) ->
    fun(Reducer) -> fun(Acc, X) -> case Pred(X) of
                true ->
                    Reducer(Acc, X);

                false ->
                    Acc
            end end end.

-spec compose(
    fun((fun((GJC, GJB) -> GJC)) -> fun((GJC, GJA) -> GJC)),
    fun((fun((GJC, GJG) -> GJC)) -> fun((GJC, GJB) -> GJC))
) -> fun((fun((GJC, GJG) -> GJC)) -> fun((GJC, GJA) -> GJC)).
compose(T1, T2) ->
    fun(Reducer) -> _pipe = Reducer,
        _pipe@1 = T2(_pipe),
        T1(_pipe@1) end.

-spec reduce(
    list(GJN),
    GJP,
    fun((fun((GJP, GJQ) -> GJP)) -> fun((GJP, GJN) -> GJP)),
    fun((GJP, GJQ) -> GJP)
) -> GJP.
reduce(Data, Initial, Transducer, Reducer) ->
    Transformed_reducer = Transducer(Reducer),
    gleam@list:fold(Data, Initial, Transformed_reducer).

-spec parallel_reduce(
    list(GJU),
    GJW,
    fun((fun((GJW, GJX) -> GJW)) -> fun((GJW, GJU) -> GJW)),
    fun((GJW, GJX) -> GJW),
    fun((GJW, GJW) -> GJW),
    fun(() -> GJW),
    integer()
) -> GJW.
parallel_reduce(
    Data,
    Initial,
    Transducer,
    Reducer,
    Combiner,
    Neutral_element,
    Num_workers
) ->
    Chunks = gleam@list:chunk(Data, fun(_) -> Num_workers end),
    Parent_subject = gleam@erlang@process:new_subject(),
    _pipe = Chunks,
    gleam@list:map(
        _pipe,
        fun(Chunk) ->
            gleam@erlang@process:start(
                fun() ->
                    Child_subject = gleam@erlang@process:new_subject(),
                    gleam@erlang@process:send(Parent_subject, Child_subject),
                    _assert_subject = gleam@erlang@process:'receive'(
                        Child_subject,
                        5000
                    ),
                    {ok, Reply} = case _assert_subject of
                        {ok, _} -> _assert_subject;
                        _assert_fail ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Assertion pattern match failed"/utf8>>,
                                        value => _assert_fail,
                                        module => <<"gtransducer"/utf8>>,
                                        function => <<"parallel_reduce"/utf8>>,
                                        line => 56})
                    end,
                    Result = reduce(
                        Chunk,
                        Neutral_element(),
                        Transducer,
                        Reducer
                    ),
                    gleam@erlang@process:send(Reply, Result)
                end,
                true
            )
        end
    ),
    _pipe@1 = Chunks,
    _pipe@2 = gleam@list:map(
        _pipe@1,
        fun(_) ->
            _assert_subject@1 = gleam@erlang@process:'receive'(
                Parent_subject,
                5000
            ),
            {ok, Child_subject@1} = case _assert_subject@1 of
                {ok, _} -> _assert_subject@1;
                _assert_fail@1 ->
                    erlang:error(#{gleam_error => let_assert,
                                message => <<"Assertion pattern match failed"/utf8>>,
                                value => _assert_fail@1,
                                module => <<"gtransducer"/utf8>>,
                                function => <<"parallel_reduce"/utf8>>,
                                line => 64})
            end,
            gleam@erlang@process:call(
                Child_subject@1,
                fun(Subject) -> Subject end,
                5000
            )
        end
    ),
    gleam@list:fold(_pipe@2, Initial, Combiner).