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

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

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

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

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

-spec next_with_timeout(stream(GVS), integer()) -> {ok, GVS} | {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(GVO)) -> {ok, GVO} | {error, nil}.
next(S) ->
    next_with_timeout(S, (15 * 60) * 1000).

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

-spec from_list(list(GVL)) -> stream(GVL).
from_list(L) ->
    S = new(),
    gleam@otp@task:async(
        fun() -> gleam@list:map(L, fun(A) -> write(S, A) end) end
    ),
    S.

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

-spec to_iterator(stream(GWA)) -> gleam@iterator:iterator(GWA).
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(GWD)) -> list(GWD).
collect(S) ->
    _pipe = to_iterator(S),
    gleam@iterator:to_list(_pipe).

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

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

        {error, nil} ->
            nil
    end.

-spec foreach_unless_error(stream(GWL), fun((GWL) -> {ok, nil} | {error, GWN})) -> {ok,
        nil} |
    {error, GWN}.
foreach_unless_error(Stream, F) ->
    case next(Stream) of
        {ok, A} ->
            case F(A) of
                {ok, nil} ->
                    foreach_unless_error(Stream, F);

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

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

-spec map(stream(GWS), fun((GWS) -> GWU)) -> stream(GWU).
map(Input, F) ->
    Output = new(),
    gleam@otp@task:async(
        fun() ->
            with(
                Output,
                fun() ->
                    foreach(
                        Input,
                        fun(A) ->
                            write(Output, F(A)),
                            nil
                        end
                    )
                end
            )
        end
    ),
    Output.

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

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

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

        {error, nil} ->
            Start
    end.