glats

-module(glats@internal@subscription).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([decode_raw_msg/1, start_subscriber/3]).
-export_type([raw_message/0, state/2, message/0]).

-type raw_message() :: {raw_message,
        integer(),
        gleam@option:option(integer()),
        binary(),
        gleam@dict:dict(binary(), binary()),
        gleam@option:option(binary()),
        binary()}.

-type state(HNB, HNC) :: {state,
        HNB,
        gleam@erlang@process:subject(HNC),
        fun((HNB, raw_message()) -> HNC)}.

-opaque message() :: {received_message, raw_message()} |
    {decode_error, gleam@dynamic:dynamic_()} |
    subscriber_exited.

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 73).
-spec loop(message(), state(HNL, HNM)) -> gleam@otp@actor:next(any(), state(HNL, HNM)).
loop(Message, State) ->
    case Message of
        {received_message, Raw_msg} ->
            gleam@otp@actor:send(
                erlang:element(3, State),
                (erlang:element(4, State))(erlang:element(2, State), Raw_msg)
            ),
            {continue, State, none};

        {decode_error, Data} ->
            gleam_stdlib:println(
                <<"failed to decode: "/utf8,
                    (gleam@string:inspect(Data))/binary>>
            ),
            {continue, State, none};

        subscriber_exited ->
            gleam_stdlib:println(<<"subscriber exited"/utf8>>),
            {stop, normal}
    end.

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 146).
-spec atom_field(
    binary(),
    fun((gleam@dynamic:dynamic_()) -> {ok, HPD} |
        {error, list(gleam@dynamic:decode_error())})
) -> fun((gleam@dynamic:dynamic_()) -> {ok, HPD} |
    {error, list(gleam@dynamic:decode_error())}).
atom_field(Key, Value) ->
    gleam@dynamic:field(erlang:binary_to_atom(Key), Value).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 107).
-spec sid(gleam@dynamic:dynamic_()) -> {ok, integer()} |
    {error, list(gleam@dynamic:decode_error())}.
sid(Data) ->
    _pipe = Data,
    _pipe@1 = (atom_field(<<"sid"/utf8>>, fun gleam@dynamic:int/1))(_pipe),
    gleam@result:'or'(_pipe@1, {ok, -1}).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 114).
-spec status(gleam@dynamic:dynamic_()) -> {ok, gleam@option:option(integer())} |
    {error, list(gleam@dynamic:decode_error())}.
status(Data) ->
    _pipe = Data,
    _pipe@1 = (atom_field(<<"status"/utf8>>, fun gleam@dynamic:string/1))(_pipe),
    _pipe@4 = gleam@result:map(
        _pipe@1,
        fun(S) -> _pipe@2 = gleam_stdlib:parse_int(S),
            _pipe@3 = gleam@result:map(
                _pipe@2,
                fun(Field@0) -> {some, Field@0} end
            ),
            gleam@result:unwrap(_pipe@3, none) end
    ),
    gleam@result:'or'(_pipe@4, {ok, none}).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 128).
-spec headers(gleam@dynamic:dynamic_()) -> {ok,
        gleam@dict:dict(binary(), binary())} |
    {error, list(gleam@dynamic:decode_error())}.
headers(Data) ->
    _pipe = Data,
    _pipe@1 = (atom_field(
        <<"headers"/utf8>>,
        gleam@dynamic:list(
            gleam@dynamic:tuple2(
                fun gleam@dynamic:string/1,
                fun gleam@dynamic:string/1
            )
        )
    ))(_pipe),
    _pipe@2 = gleam@result:map(_pipe@1, fun maps:from_list/1),
    gleam@result:'or'(_pipe@2, {ok, maps:new()}).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 140).
-spec reply_to(gleam@dynamic:dynamic_()) -> {ok, gleam@option:option(binary())} |
    {error, list(gleam@dynamic:decode_error())}.
reply_to(Data) ->
    _pipe = Data,
    _pipe@1 = (gleam@dynamic:optional(
        atom_field(<<"reply_to"/utf8>>, fun gleam@dynamic:string/1)
    ))(_pipe),
    gleam@result:'or'(_pipe@1, {ok, none}).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 93).
-spec decode_raw_msg(gleam@dynamic:dynamic_()) -> {ok, raw_message()} |
    {error, list(gleam@dynamic:decode_error())}.
decode_raw_msg(Data) ->
    _pipe = Data,
    (gleam@dynamic:decode6(
        fun(Field@0, Field@1, Field@2, Field@3, Field@4, Field@5) -> {raw_message, Field@0, Field@1, Field@2, Field@3, Field@4, Field@5} end,
        fun sid/1,
        fun status/1,
        atom_field(<<"topic"/utf8>>, fun gleam@dynamic:string/1),
        fun headers/1,
        fun reply_to/1,
        atom_field(<<"body"/utf8>>, fun gleam@dynamic:string/1)
    ))(_pipe).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 66).
-spec map_gnat_message(gleam@dynamic:dynamic_()) -> message().
map_gnat_message(Data) ->
    _pipe = Data,
    _pipe@1 = decode_raw_msg(_pipe),
    _pipe@2 = gleam@result:map(
        _pipe@1,
        fun(Field@0) -> {received_message, Field@0} end
    ),
    gleam@result:unwrap(_pipe@2, {decode_error, Data}).

-file("/home/arnar/Code/glats/src/glats/internal/subscription.gleam", 37).
-spec start_subscriber(
    HNF,
    gleam@erlang@process:subject(HNG),
    fun((HNF, raw_message()) -> HNG)
) -> {ok, gleam@erlang@process:subject(message())} |
    {error, gleam@otp@actor:start_error()}.
start_subscriber(Conn, Receiver, Mapper) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Monitor = gleam@erlang@process:monitor_process(
                    begin
                        _pipe = Receiver,
                        gleam@erlang@process:subject_owner(_pipe)
                    end
                ),
                Selector = begin
                    _pipe@1 = gleam_erlang_ffi:new_selector(),
                    _pipe@2 = gleam@erlang@process:selecting_process_down(
                        _pipe@1,
                        Monitor,
                        fun(_) -> subscriber_exited end
                    ),
                    gleam@erlang@process:selecting_record2(
                        _pipe@2,
                        erlang:binary_to_atom(<<"msg"/utf8>>),
                        fun map_gnat_message/1
                    )
                end,
                {ready, {state, Conn, Receiver, Mapper}, Selector}
            end,
            1000,
            fun loop/2}
    ).