radish

-module(radish@pub_sub).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function]).

-export([publish/4, subscribe/5, subscribe_to_patterns/5]).
-export_type([next/0]).

-type next() :: continue |
    unsubscribe_from_all |
    {unsubscribe_from, list(binary())}.

-spec publish(
    gleam@erlang@process:subject(radish@client:message()),
    binary(),
    binary(),
    integer()
) -> {ok, integer()} | {error, radish@error:error()}.
publish(Client, Channel, Message, Timeout) ->
    _pipe = radish@command@pub_sub:publish(Channel, Message),
    _pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
    _pipe@2 = gleam@result:map(_pipe@1, fun(Value) -> case Value of
                [{integer, N}] ->
                    {ok, N};

                _ ->
                    {error, resp_error}
            end end),
    gleam@result:flatten(_pipe@2).

-spec unsubscribe(
    gleam@erlang@process:subject(radish@client:message()),
    list(binary()),
    integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe(Client, Channels, Timeout) ->
    _pipe = radish@command@pub_sub:unsubscribe(Channels),
    _pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
    gleam@result:map(
        _pipe@1,
        fun(Value) ->
            gleam@list:all(
                Value,
                fun(Item) ->
                    {push,
                        [{bulk_string, <<"unsubscribe"/utf8>>},
                            {bulk_string, _},
                            {integer, N}]} = case Item of
                        {push,
                            [{bulk_string, <<"unsubscribe"/utf8>>},
                                {bulk_string, _},
                                {integer, _}]} -> Item;
                        _assert_fail ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Assertion pattern match failed"/utf8>>,
                                        value => _assert_fail,
                                        module => <<"radish/pub_sub"/utf8>>,
                                        function => <<"unsubscribe"/utf8>>,
                                        line => 144})
                    end,
                    N > 0
                end
            )
        end
    ).

-spec unsubscribe_from_all(
    gleam@erlang@process:subject(radish@client:message()),
    integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe_from_all(Client, Timeout) ->
    _pipe = radish@command@pub_sub:unsubscribe_from_all(),
    _pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
    gleam@result:map(
        _pipe@1,
        fun(Value) ->
            Value,
            gleam@list:all(
                Value,
                fun(Item) ->
                    {push,
                        [{bulk_string, <<"unsubscribe"/utf8>>},
                            {bulk_string, _},
                            {integer, N}]} = case Item of
                        {push,
                            [{bulk_string, <<"unsubscribe"/utf8>>},
                                {bulk_string, _},
                                {integer, _}]} -> Item;
                        _assert_fail ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Assertion pattern match failed"/utf8>>,
                                        value => _assert_fail,
                                        module => <<"radish/pub_sub"/utf8>>,
                                        function => <<"unsubscribe_from_all"/utf8>>,
                                        line => 163})
                    end,
                    N > 0
                end
            )
        end
    ).

-spec subscribe(
    gleam@erlang@process:subject(radish@client:message()),
    list(binary()),
    fun((binary(), integer()) -> nil),
    fun((binary(), binary()) -> next()),
    integer()
) -> nil.
subscribe(Client, Channels, Init_handler, Message_handler, Timeout) ->
    _ = begin
        _pipe = radish@command@pub_sub:subscribe(Channels),
        _pipe@1 = radish@utils:execute_blocking(Client, _pipe, Timeout),
        gleam@result:map(
            _pipe@1,
            fun(Value) -> gleam@list:each(Value, fun(Item) -> case Item of
                            {push,
                                [{bulk_string, <<"subscribe"/utf8>>},
                                    {bulk_string, Channel},
                                    {integer, N}]} ->
                                {ok, Init_handler(Channel, N)};

                            _ ->
                                {error, resp_error}
                        end end) end
        )
    end,
    radish@utils:receive_forever(
        Client,
        Timeout,
        fun(Value@1) -> case Value@1 of
                {ok,
                    [{push,
                            [{bulk_string, <<"message"/utf8>>},
                                {bulk_string, Channel@1},
                                {bulk_string, Message}]}]} ->
                    case Message_handler(Channel@1, Message) of
                        continue ->
                            true;

                        unsubscribe_from_all ->
                            _ = unsubscribe_from_all(Client, Timeout),
                            false;

                        {unsubscribe_from, Channels@1} ->
                            case unsubscribe(Client, Channels@1, Timeout) of
                                {ok, Result} ->
                                    Result;

                                {error, _} ->
                                    false
                            end
                    end;

                _ ->
                    false
            end end
    ).

-spec unsubscribe_from_patterns(
    gleam@erlang@process:subject(radish@client:message()),
    list(binary()),
    integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe_from_patterns(Client, Patterns, Timeout) ->
    _pipe = radish@command@pub_sub:unsubscribe_from_patterns(Patterns),
    _pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
    gleam@result:map(
        _pipe@1,
        fun(Value) ->
            gleam@list:all(
                Value,
                fun(Item) ->
                    {push,
                        [{bulk_string, <<"punsubscribe"/utf8>>},
                            {bulk_string, _},
                            {integer, N}]} = case Item of
                        {push,
                            [{bulk_string, <<"punsubscribe"/utf8>>},
                                {bulk_string, _},
                                {integer, _}]} -> Item;
                        _assert_fail ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Assertion pattern match failed"/utf8>>,
                                        value => _assert_fail,
                                        module => <<"radish/pub_sub"/utf8>>,
                                        function => <<"unsubscribe_from_patterns"/utf8>>,
                                        line => 181})
                    end,
                    N > 0
                end
            )
        end
    ).

-spec unsubscribe_from_all_patterns(
    gleam@erlang@process:subject(radish@client:message()),
    integer()
) -> {ok, boolean()} | {error, radish@error:error()}.
unsubscribe_from_all_patterns(Client, Timeout) ->
    _pipe = radish@command@pub_sub:unsubscribe_from_all_patterns(),
    _pipe@1 = radish@utils:execute(Client, _pipe, Timeout),
    gleam@result:map(
        _pipe@1,
        fun(Value) ->
            gleam@list:all(
                Value,
                fun(Item) ->
                    {push,
                        [{bulk_string, <<"punsubscribe"/utf8>>},
                            {bulk_string, _},
                            {integer, N}]} = case Item of
                        {push,
                            [{bulk_string, <<"punsubscribe"/utf8>>},
                                {bulk_string, _},
                                {integer, _}]} -> Item;
                        _assert_fail ->
                            erlang:error(#{gleam_error => let_assert,
                                        message => <<"Assertion pattern match failed"/utf8>>,
                                        value => _assert_fail,
                                        module => <<"radish/pub_sub"/utf8>>,
                                        function => <<"unsubscribe_from_all_patterns"/utf8>>,
                                        line => 199})
                    end,
                    N > 0
                end
            )
        end
    ).

-spec subscribe_to_patterns(
    gleam@erlang@process:subject(radish@client:message()),
    list(binary()),
    fun((binary(), integer()) -> nil),
    fun((binary(), binary(), binary()) -> next()),
    integer()
) -> nil.
subscribe_to_patterns(Client, Patterns, Init_handler, Message_handler, Timeout) ->
    _ = begin
        _pipe = radish@command@pub_sub:subscribe_to_patterns(Patterns),
        _pipe@1 = radish@utils:execute_blocking(Client, _pipe, Timeout),
        gleam@result:map(
            _pipe@1,
            fun(Value) -> gleam@list:each(Value, fun(Item) -> case Item of
                            {push,
                                [{bulk_string, <<"psubscribe"/utf8>>},
                                    {bulk_string, Channel},
                                    {integer, N}]} ->
                                Init_handler(Channel, N);

                            _ ->
                                nil
                        end end) end
        )
    end,
    radish@utils:receive_forever(
        Client,
        Timeout,
        fun(Value@1) -> case Value@1 of
                {ok,
                    [{push,
                            [{bulk_string, <<"pmessage"/utf8>>},
                                {bulk_string, Pattern},
                                {bulk_string, Channel@1},
                                {bulk_string, Message}]}]} ->
                    case Message_handler(Pattern, Channel@1, Message) of
                        continue ->
                            true;

                        unsubscribe_from_all ->
                            _ = unsubscribe_from_all_patterns(Client, Timeout),
                            false;

                        {unsubscribe_from, Patterns@1} ->
                            case unsubscribe_from_patterns(
                                Client,
                                Patterns@1,
                                Timeout
                            ) of
                                {ok, Result} ->
                                    Result;

                                {error, _} ->
                                    false
                            end
                    end;

                _ ->
                    false
            end end
    ).