glimmer

-module(glimmer@stream).
-compile([no_auto_import, nowarn_unused_vars]).

-export([new/0, next_with_timeout/2, next/1, write/2, close/1, from_list/1, to_iterator/1, collect/1, with/2, each/2, try_each/2, map/2, filter/2, reduce/3, generator/1]).
-export_type([pipe_message/1, stream/1]).

-opaque pipe_message(GXM) :: {another, GXM} | done.

-type stream(GXN) :: {stream, gleam@erlang@process:subject(pipe_message(GXN))}.

-spec new() -> stream(any()).
new() ->
    {stream, gleam@erlang@process:new_subject()}.

-spec next_with_timeout(stream(GXX), integer()) -> {ok, GXX} | {error, nil}.
next_with_timeout(S, Timeout) ->
    {stream, Subject} = S,
    case gleam@erlang@process:'receive'(Subject, Timeout) of
        {ok, {another, A}} ->
            {ok, A};

        {ok, done} ->
            {error, nil};

        {error, nil} ->
            {error, nil}
    end.

-spec next(stream(GXT)) -> {ok, GXT} | {error, nil}.
next(S) ->
    next_with_timeout(S, (15 * 60) * 1000).

-spec write(stream(GYB), GYB) -> nil.
write(S, Value) ->
    {stream, Subject} = S,
    gleam@erlang@process:send(Subject, {another, Value}).

-spec close(stream(any())) -> nil.
close(S) ->
    {stream, Subject} = S,
    gleam@erlang@process:send(Subject, done).

-spec from_list(list(GXQ)) -> stream(GXQ).
from_list(L) ->
    S = new(),
    gleam@otp@task:async(
        fun() ->
            gleam@list:map(L, fun(A) -> write(S, A) end),
            gleam@io:println(<<"from_list sent everything"/utf8>>),
            close(S),
            gleam@io:println(<<"from_list closed"/utf8>>)
        end
    ),
    S.

-spec to_iterator(stream(GYF)) -> gleam@iterator:iterator(GYF).
to_iterator(S) ->
    gleam@iterator:unfold(nil, fun(_) -> case next(S) of
                {ok, A} ->
                    {next, A, nil};

                {error, nil} ->
                    done
            end end).

-spec collect(stream(GYI)) -> list(GYI).
collect(S) ->
    _pipe = to_iterator(S),
    gleam@iterator:to_list(_pipe).

-spec with(stream(any()), fun(() -> GYN)) -> GYN.
with(Stream, F) ->
    Out = F(),
    close(Stream),
    Out.

-spec each(stream(GYO), fun((GYO) -> nil)) -> nil.
each(Stream, F) ->
    case next(Stream) of
        {ok, A} ->
            F(A),
            each(Stream, F);

        {error, nil} ->
            nil
    end.

-spec try_each(stream(GYQ), fun((GYQ) -> {ok, nil} | {error, GYS})) -> {ok, nil} |
    {error, GYS}.
try_each(Stream, F) ->
    case next(Stream) of
        {ok, A} ->
            case F(A) of
                {ok, nil} ->
                    try_each(Stream, F);

                {error, Err} ->
                    {error, Err}
            end;

        {error, nil} ->
            {ok, nil}
    end.

-spec map(stream(GYX), fun((GYX) -> GYZ)) -> stream(GYZ).
map(Input, F) ->
    Output = new(),
    gleam@io:println(<<"starting map"/utf8>>),
    glimmer@dream:spawn(
        fun() ->
            each(
                Input,
                fun(A) ->
                    gleam@io:debug(A),
                    write(Output, F(A))
                end
            ),
            close(Output)
        end
    ),
    Output.

-spec filter(stream(GZB), fun((GZB) -> boolean())) -> stream(GZB).
filter(Input, P) ->
    Output = new(),
    gleam@otp@task:async(
        fun() -> with(Output, fun() -> each(Input, fun(A) -> case P(A) of
                                true ->
                                    write(Output, A);

                                false ->
                                    nil
                            end end) end) end
    ),
    Output.

-spec reduce(stream(GZE), GZG, fun((GZE, GZG) -> GZG)) -> GZG.
reduce(Input, Start, F) ->
    case next(Input) of
        {ok, A} ->
            reduce(Input, F(A, Start), F);

        {error, nil} ->
            Start
    end.

-spec generator(fun((fun((GZH) -> nil), fun(() -> nil)) -> nil)) -> stream(GZH).
generator(G) ->
    Output = new(),
    G(fun(B) -> write(Output, B) end, fun() -> close(Output) end),
    Output.