-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).