glubsub

-module(glubsub).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([destroy_topic/1, broadcast/2, new_topic/0, subscribe/2, unsubscribe/2, get_subscribers/1]).
-export_type([topic/1, message/1, glubsub_error/0, state/1, subscriber/1]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

?MODULEDOC(
    " This module implements a simple pubsub system using gleam actors.\n"
    "\n"
).

-type topic(HHF) :: {topic, gleam@erlang@process:subject(message(HHF))}.

-opaque message(HHG) :: {subscribe,
        gleam@erlang@process:subject({ok, nil} | {error, glubsub_error()}),
        gleam@erlang@process:subject(HHG)} |
    {unsubscribe,
        gleam@erlang@process:subject({ok, nil} | {error, glubsub_error()}),
        gleam@erlang@process:subject(HHG)} |
    {broadcast, HHG} |
    {get_subscribers, gleam@erlang@process:subject(list(subscriber(HHG)))} |
    {subscriber_down, gleam@erlang@process:process_down()} |
    shutdown.

-opaque glubsub_error() :: already_subscribed |
    not_subscribed |
    {start_error, gleam@otp@actor:start_error()}.

-type state(HHH) :: {state,
        list(subscriber(HHH)),
        gleam@erlang@process:selector(message(HHH))}.

-type subscriber(HHI) :: {subscriber,
        gleam@erlang@process:subject(HHI),
        gleam@erlang@process:process_monitor()}.

-file("src/glubsub.gleam", 176).
-spec topic_to_subject(topic(HIO)) -> gleam@erlang@process:subject(message(HIO)).
topic_to_subject(Topic) ->
    {topic, Subject} = Topic,
    Subject.

-file("src/glubsub.gleam", 56).
?DOC(" Destroys the given topic, unsubscribing all clients.\n").
-spec destroy_topic(topic(any())) -> nil.
destroy_topic(Topic) ->
    gleam@otp@actor:send(topic_to_subject(Topic), shutdown).

-file("src/glubsub.gleam", 85).
?DOC(" Broadcasts a message to all subscribers of the given topic.\n").
-spec broadcast(topic(HHZ), HHZ) -> {ok, nil} | {error, nil}.
broadcast(Topic, Message) ->
    _pipe = gleam@otp@actor:send(topic_to_subject(Topic), {broadcast, Message}),
    {ok, _pipe}.

-file("src/glubsub.gleam", 181).
-spec remove_subscriber(list(subscriber(HIS)), subscriber(HIS)) -> list(subscriber(HIS)).
remove_subscriber(Subscribers, Unsubscriber) ->
    gleam@list:filter(Subscribers, fun(Sub) -> Sub /= Unsubscriber end).

-file("src/glubsub.gleam", 188).
-spec selector_down_from_subscribers(
    gleam@erlang@process:selector(message(HIY)),
    list(subscriber(HIY))
) -> gleam@erlang@process:selector(message(HIY)).
selector_down_from_subscribers(Selector, Subscribers) ->
    gleam@list:fold(
        Subscribers,
        Selector,
        fun(Selector@1, Sub) ->
            gleam@erlang@process:selecting_process_down(
                Selector@1,
                erlang:element(3, Sub),
                fun(Field@0) -> {subscriber_down, Field@0} end
            )
        end
    ).

-file("src/glubsub.gleam", 95).
-spec handle_message(message(HIH), state(HIH)) -> gleam@otp@actor:next(message(HIH), state(HIH)).
handle_message(Message, State) ->
    case Message of
        {subscribe, Reply, Client} ->
            case gleam@list:find(
                erlang:element(2, State),
                fun(Sub) -> erlang:element(2, Sub) =:= Client end
            ) of
                {ok, _} ->
                    gleam@otp@actor:send(Reply, {error, already_subscribed}),
                    gleam@otp@actor:continue(State);

                {error, nil} ->
                    Monitor = gleam@erlang@process:monitor_process(
                        gleam@erlang@process:subject_owner(Client)
                    ),
                    New_selector = begin
                        _pipe = erlang:element(3, State),
                        gleam@erlang@process:selecting_process_down(
                            _pipe,
                            Monitor,
                            fun(Field@0) -> {subscriber_down, Field@0} end
                        )
                    end,
                    New_subs = [{subscriber, Client, Monitor} |
                        erlang:element(2, State)],
                    gleam@otp@actor:send(Reply, {ok, nil}),
                    _pipe@1 = gleam@otp@actor:continue(
                        {state, New_subs, New_selector}
                    ),
                    gleam@otp@actor:with_selector(_pipe@1, New_selector)
            end;

        {unsubscribe, Reply@1, Client@1} ->
            case gleam@list:find(
                erlang:element(2, State),
                fun(Sub@1) -> erlang:element(2, Sub@1) =:= Client@1 end
            ) of
                {error, nil} ->
                    gleam@otp@actor:send(Reply@1, {error, not_subscribed}),
                    gleam@otp@actor:continue(State);

                {ok, Unsub} ->
                    New_subs@1 = remove_subscriber(
                        erlang:element(2, State),
                        Unsub
                    ),
                    New_selector@1 = begin
                        _pipe@2 = gleam_erlang_ffi:new_selector(),
                        selector_down_from_subscribers(_pipe@2, New_subs@1)
                    end,
                    gleam@otp@actor:send(Reply@1, {ok, nil}),
                    gleam@otp@actor:continue(
                        {state, New_subs@1, New_selector@1}
                    )
            end;

        {broadcast, Message@1} ->
            _pipe@3 = erlang:element(2, State),
            gleam@list:each(
                _pipe@3,
                fun(Sub@2) ->
                    gleam@otp@actor:send(erlang:element(2, Sub@2), Message@1)
                end
            ),
            gleam@otp@actor:continue(State);

        {get_subscribers, Reply@2} ->
            gleam@otp@actor:send(Reply@2, erlang:element(2, State)),
            gleam@otp@actor:continue(State);

        {subscriber_down, Client@2} ->
            case begin
                _pipe@4 = erlang:element(2, State),
                gleam@list:find(
                    _pipe@4,
                    fun(Sub@3) ->
                        gleam@erlang@process:subject_owner(
                            erlang:element(2, Sub@3)
                        )
                        =:= erlang:element(2, Client@2)
                    end
                )
            end of
                {error, nil} ->
                    gleam@otp@actor:continue(State);

                {ok, Unsub@1} ->
                    gleam@erlang@process:demonitor_process(
                        erlang:element(3, Unsub@1)
                    ),
                    New_subs@2 = remove_subscriber(
                        erlang:element(2, State),
                        Unsub@1
                    ),
                    New_selector@2 = begin
                        _pipe@5 = gleam_erlang_ffi:new_selector(),
                        selector_down_from_subscribers(_pipe@5, New_subs@2)
                    end,
                    gleam@otp@actor:continue(
                        {state, New_subs@2, New_selector@2}
                    )
            end;

        shutdown ->
            {stop, normal}
    end.

-file("src/glubsub.gleam", 42).
?DOC(" Creates a new topic. Which is a pubsub channel that clients can subscribe to.\n").
-spec new_topic() -> {ok, topic(any())} | {error, glubsub_error()}.
new_topic() ->
    _pipe = gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Selector = gleam_erlang_ffi:new_selector(),
                {ready, {state, [], Selector}, Selector}
            end,
            1000,
            fun handle_message/2}
    ),
    _pipe@1 = gleam@result:map(_pipe, fun(Subject) -> {topic, Subject} end),
    gleam@result:map_error(_pipe@1, fun(Err) -> {start_error, Err} end).

-file("src/glubsub.gleam", 61).
?DOC(" Subscribes the given client to the given topic.\n").
-spec subscribe(topic(HHP), gleam@erlang@process:subject(HHP)) -> {ok, nil} |
    {error, glubsub_error()}.
subscribe(Topic, Client) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        fun(Self) -> {subscribe, Self, Client} end,
        1000
    ).

-file("src/glubsub.gleam", 73).
?DOC(" Unsubscribes the given client from the given topic.\n").
-spec unsubscribe(topic(HHU), gleam@erlang@process:subject(HHU)) -> {ok, nil} |
    {error, glubsub_error()}.
unsubscribe(Topic, Client) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        fun(Self) -> {unsubscribe, Self, Client} end,
        1000
    ).

-file("src/glubsub.gleam", 91).
?DOC(" Returns a set of all subscribers to the given topic.\n").
-spec get_subscribers(topic(HID)) -> list(subscriber(HID)).
get_subscribers(Topic) ->
    gleam@otp@actor:call(
        topic_to_subject(Topic),
        fun(Field@0) -> {get_subscribers, Field@0} end,
        1000
    ).