-module(lightspeed@pubsub).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/pubsub.gleam").
-export([new_single_node/0, new_cluster_ready/2, adapter/1, adapter_label/1, subscribe/3, unsubscribe/3, subscribers/2, publish/3, subscriber_count/2, topic_labels/1, delivery_subscriber/1, delivery_topic/1, delivery_message/1]).
-export_type([adapter/0, delivery/1, topic_subscribers/0, broker/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(" Deterministic PubSub abstraction for channel fanout.\n").
-type adapter() :: single_node | {cluster_ready, binary(), binary()}.
-type delivery(GYA) :: {delivery, binary(), binary(), GYA}.
-type topic_subscribers() :: {topic_subscribers, binary(), list(binary())}.
-opaque broker(GYB) :: {broker, adapter(), list(topic_subscribers())} |
{gleam_phantom, GYB}.
-file("src/lightspeed/pubsub.gleam", 27).
?DOC(" New single-node broker.\n").
-spec new_single_node() -> broker(any()).
new_single_node() ->
{broker, single_node, []}.
-file("src/lightspeed/pubsub.gleam", 32).
?DOC(" New cluster-ready broker.\n").
-spec new_cluster_ready(binary(), binary()) -> broker(any()).
new_cluster_ready(Node_id, Cluster) ->
{broker, {cluster_ready, Node_id, Cluster}, []}.
-file("src/lightspeed/pubsub.gleam", 40).
?DOC(" Broker adapter.\n").
-spec adapter(broker(any())) -> adapter().
adapter(Broker) ->
erlang:element(2, Broker).
-file("src/lightspeed/pubsub.gleam", 45).
?DOC(" Stable adapter label.\n").
-spec adapter_label(adapter()) -> binary().
adapter_label(Adapter) ->
case Adapter of
single_node ->
<<"single_node"/utf8>>;
{cluster_ready, Node_id, Cluster} ->
<<<<<<"cluster_ready:"/utf8, Cluster/binary>>/binary, ":"/utf8>>/binary,
Node_id/binary>>
end.
-file("src/lightspeed/pubsub.gleam", 189).
-spec contains(list(binary()), binary()) -> boolean().
contains(Values, Target) ->
case Values of
[] ->
false;
[Value | Rest] ->
case Value =:= Target of
true ->
true;
false ->
contains(Rest, Target)
end
end.
-file("src/lightspeed/pubsub.gleam", 122).
-spec upsert_subscriber(list(topic_subscribers()), binary(), binary()) -> list(topic_subscribers()).
upsert_subscriber(Topics_rev, Topic, Subscriber_id) ->
case Topics_rev of
[] ->
[{topic_subscribers, Topic, [Subscriber_id]}];
[Entry | Rest] ->
case erlang:element(2, Entry) =:= Topic of
true ->
case contains(erlang:element(3, Entry), Subscriber_id) of
true ->
[Entry | Rest];
false ->
[{topic_subscribers,
erlang:element(2, Entry),
[Subscriber_id | erlang:element(3, Entry)]} |
Rest]
end;
false ->
[Entry | upsert_subscriber(Rest, Topic, Subscriber_id)]
end
end.
-file("src/lightspeed/pubsub.gleam", 54).
?DOC(" Subscribe one subscriber to one topic.\n").
-spec subscribe(broker(GYI), binary(), binary()) -> broker(GYI).
subscribe(Broker, Topic, Subscriber_id) ->
{broker,
erlang:element(2, Broker),
upsert_subscriber(erlang:element(3, Broker), Topic, Subscriber_id)}.
-file("src/lightspeed/pubsub.gleam", 200).
-spec remove_item(list(binary()), binary()) -> list(binary()).
remove_item(Values, Target) ->
case Values of
[] ->
[];
[Value | Rest] ->
case Value =:= Target of
true ->
remove_item(Rest, Target);
false ->
[Value | remove_item(Rest, Target)]
end
end.
-file("src/lightspeed/pubsub.gleam", 149).
-spec remove_subscriber(list(topic_subscribers()), binary(), binary()) -> list(topic_subscribers()).
remove_subscriber(Topics_rev, Topic, Subscriber_id) ->
case Topics_rev of
[] ->
[];
[Entry | Rest] ->
case erlang:element(2, Entry) =:= Topic of
true ->
Remaining = remove_item(
erlang:element(3, Entry),
Subscriber_id
),
case Remaining of
[] ->
Rest;
_ ->
[{topic_subscribers,
erlang:element(2, Entry),
Remaining} |
Rest]
end;
false ->
[Entry | remove_subscriber(Rest, Topic, Subscriber_id)]
end
end.
-file("src/lightspeed/pubsub.gleam", 66).
?DOC(" Unsubscribe one subscriber from one topic.\n").
-spec unsubscribe(broker(GYL), binary(), binary()) -> broker(GYL).
unsubscribe(Broker, Topic, Subscriber_id) ->
{broker,
erlang:element(2, Broker),
remove_subscriber(erlang:element(3, Broker), Topic, Subscriber_id)}.
-file("src/lightspeed/pubsub.gleam", 211).
-spec deliver_loop(binary(), GZQ, list(binary()), list(delivery(GZQ))) -> list(delivery(GZQ)).
deliver_loop(Topic, Message, Subscribers, Deliveries_rev) ->
case Subscribers of
[] ->
lists:reverse(Deliveries_rev);
[Subscriber_id | Rest] ->
deliver_loop(
Topic,
Message,
Rest,
[{delivery, Topic, Subscriber_id, Message} | Deliveries_rev]
)
end.
-file("src/lightspeed/pubsub.gleam", 175).
-spec find_topic(list(topic_subscribers()), binary()) -> list(binary()).
find_topic(Topics_rev, Topic) ->
case Topics_rev of
[] ->
[];
[Entry | Rest] ->
case erlang:element(2, Entry) =:= Topic of
true ->
erlang:element(3, Entry);
false ->
find_topic(Rest, Topic)
end
end.
-file("src/lightspeed/pubsub.gleam", 88).
?DOC(" Subscribers for one topic in stable join order.\n").
-spec subscribers(broker(any()), binary()) -> list(binary()).
subscribers(Broker, Topic) ->
_pipe = erlang:element(3, Broker),
_pipe@1 = find_topic(_pipe, Topic),
lists:reverse(_pipe@1).
-file("src/lightspeed/pubsub.gleam", 78).
?DOC(" Publish one message to one topic.\n").
-spec publish(broker(GYO), binary(), GYO) -> {broker(GYO), list(delivery(GYO))}.
publish(Broker, Topic, Message) ->
Subscribers = subscribers(Broker, Topic),
{Broker, deliver_loop(Topic, Message, Subscribers, [])}.
-file("src/lightspeed/pubsub.gleam", 95).
?DOC(" Number of subscribers for one topic.\n").
-spec subscriber_count(broker(any()), binary()) -> integer().
subscriber_count(Broker, Topic) ->
erlang:length(subscribers(Broker, Topic)).
-file("src/lightspeed/pubsub.gleam", 100).
?DOC(" Stable topic labels for tests and logs.\n").
-spec topic_labels(broker(any())) -> list(binary()).
topic_labels(Broker) ->
_pipe = erlang:element(3, Broker),
gleam@list:map(
_pipe,
fun(Topic) ->
<<<<(erlang:element(2, Topic))/binary, ":"/utf8>>/binary,
(erlang:integer_to_binary(
erlang:length(erlang:element(3, Topic))
))/binary>>
end
).
-file("src/lightspeed/pubsub.gleam", 108).
?DOC(" Subscriber id for one delivery.\n").
-spec delivery_subscriber(delivery(any())) -> binary().
delivery_subscriber(Delivery) ->
erlang:element(3, Delivery).
-file("src/lightspeed/pubsub.gleam", 113).
?DOC(" Topic for one delivery.\n").
-spec delivery_topic(delivery(any())) -> binary().
delivery_topic(Delivery) ->
erlang:element(2, Delivery).
-file("src/lightspeed/pubsub.gleam", 118).
?DOC(" Message payload for one delivery.\n").
-spec delivery_message(delivery(GZF)) -> GZF.
delivery_message(Delivery) ->
erlang:element(4, Delivery).
