macula_pubsub_subscription (macula v0.20.5)

View Source

Subscription management for pub/sub.

Responsibilities: - Store and retrieve subscriptions - Pattern matching with wildcards (*, **) - Find matching subscriptions for a topic - Invoke subscriber callbacks

Extracted from macula_pubsub_handler.erl (Phase 4)

Summary

Types

callback/0
node_id/0
payload/0
subscription_ref/0
subscriptions/0
topic/0

Functions

add_subscription(Topic, Callback, Subscriptions, SubRef)

Add a subscription. Returns {ok, UpdatedSubscriptions, SubRef}.

find_matches(Topic, Subscriptions, Config)

Find matching subscriptions for a topic. Returns list of {SubRef, {Pattern, Callback}} tuples.

invoke_callbacks(Matches, Topic, Payload, NodeId)

Invoke callbacks for matching subscriptions. Spawns async tasks to invoke each callback.

remove_subscription(SubRef, Subscriptions)

Remove a subscription. Returns {ok, UpdatedSubscriptions, Topic} | {error, not_found}.

Types

callback/0

-type callback() :: fun((map()) -> ok).

node_id/0

-type node_id() :: binary().

payload/0

-type payload() :: binary().

subscription_ref/0

-type subscription_ref() :: reference().

subscriptions/0

-type subscriptions() :: #{subscription_ref() => {topic(), callback()}}.

topic/0

-type topic() :: binary().

Functions

add_subscription(Topic, Callback, Subscriptions, SubRef)

-spec add_subscription(topic(), callback(), subscriptions(), subscription_ref()) ->
                          {ok, subscriptions(), subscription_ref()}.

Add a subscription. Returns {ok, UpdatedSubscriptions, SubRef}.

find_matches(Topic, Subscriptions, Config)

-spec find_matches(topic(), subscriptions(), #{atom() => binary()}) ->
                      [{subscription_ref(), {topic(), callback()}}].

Find matching subscriptions for a topic. Returns list of {SubRef, {Pattern, Callback}} tuples.

invoke_callbacks(Matches, Topic, Payload, NodeId)

-spec invoke_callbacks([{subscription_ref(), {topic(), callback()}}], topic(), payload(), node_id()) ->
                          ok.

Invoke callbacks for matching subscriptions. Spawns async tasks to invoke each callback.

remove_subscription(SubRef, Subscriptions)

-spec remove_subscription(subscription_ref(), subscriptions()) ->
                             {ok, subscriptions(), topic()} | {error, not_found}.

Remove a subscription. Returns {ok, UpdatedSubscriptions, Topic} | {error, not_found}.