MatterEx.IM.SubscriptionManager (matter_ex v0.3.1)

Tracks active Matter subscriptions for a session.

Pure functional module — caller threads state through.

Each subscription monitors a set of attribute paths with min/max reporting intervals. The due_reports/2 function checks which subscriptions need a periodic report based on elapsed time.

Example

mgr = SubscriptionManager.new()

{sub_id, mgr} = SubscriptionManager.subscribe(mgr,
  [%{endpoint: 1, cluster: 6, attribute: 0}],
  0,   # min_interval (seconds)
  60   # max_interval (seconds)
)

# Later, check for due reports
due = SubscriptionManager.due_reports(mgr, System.monotonic_time(:second))

Summary

Functions

active?(state)

Check if any subscriptions are active.

due_reports(state, now)

Check which subscriptions are due for a report check.

get(state, sub_id)

Get a subscription by ID.

new()

Create a new empty SubscriptionManager.

record_report(state, sub_id, values, now)

Record that a report was checked for a subscription.

record_sent(state, sub_id, values, now)

Record that a report was actually sent for a subscription.

subscribe(state, paths, min_interval, max_interval)

subscriptions(state)

List all active subscriptions.

throttled?(state, sub_id, now)

Check if a subscription is throttled by min_interval.

unsubscribe(state, sub_id)

Remove a subscription by ID.

unsubscribe_all(state)

Remove all subscriptions. Used for session cleanup.

Types

subscription()

@type subscription() :: %{
  id: non_neg_integer(),
  paths: [map()],
  min_interval: non_neg_integer(),
  max_interval: non_neg_integer(),
  last_report_at: integer(),
  last_sent_at: integer() | nil,
  last_values: map()
}

t()

@type t() :: %MatterEx.IM.SubscriptionManager{
  next_id: non_neg_integer(),
  subscriptions: %{required(non_neg_integer()) => subscription()}
}

Functions

active?(state)

@spec active?(t()) :: boolean()

Check if any subscriptions are active.

due_reports(state, now)

@spec due_reports(t(), integer()) :: [{non_neg_integer(), [map()]}]

Check which subscriptions are due for a report check.

Returns a list of {sub_id, paths} tuples when either the minimum reporting interval allows change detection or the maximum interval has elapsed. The caller decides whether to send a report after comparing values.

get(state, sub_id)

@spec get(t(), non_neg_integer()) :: subscription() | nil

Get a subscription by ID.

new()

@spec new() :: t()

Create a new empty SubscriptionManager.

record_report(state, sub_id, values, now)

@spec record_report(t(), non_neg_integer(), map(), integer()) :: t()

Record that a report was checked for a subscription.

Updates last_report_at and last_values for change detection. Does NOT update last_sent_at — use record_sent/3 for that.

record_sent(state, sub_id, values, now)

@spec record_sent(t(), non_neg_integer(), map(), integer()) :: t()

Record that a report was actually sent for a subscription.

Updates last_sent_at, last_report_at, and last_values.

subscriptions(state)

@spec subscriptions(t()) :: [subscription()]

List all active subscriptions.

throttled?(state, sub_id, now)

@spec throttled?(t(), non_neg_integer(), integer()) :: boolean()

Check if a subscription is throttled by min_interval.

Returns true when the time since the last sent report is less than min_interval, meaning a change-triggered report should be suppressed.

unsubscribe(state, sub_id)

@spec unsubscribe(t(), non_neg_integer()) :: t()

Remove a subscription by ID.

unsubscribe_all(state)

@spec unsubscribe_all(t()) :: t()

Remove all subscriptions. Used for session cleanup.