barrel_mcp_subscriptions (barrel_mcp v3.0.0)

View Source

Long-lived notification streams (MCP 2026-07-28).

subscriptions/listen replaced the standalone GET SSE stream and the resources/subscribe RPC. A client opens one by POSTing a request whose response stream stays open, naming the notification types it wants; the server sends those and nothing else.

This module is the registry behind that. Each entry belongs to the request process holding the stream, and is keyed by that process together with the JSON-RPC id of the subscriptions/listen request, which is the subscription id on the wire. That id is only unique per connection, hence the pid in the key.

Entries are monitored, so a stream whose process dies without cleaning up does not leak.

Fan-out reads the table directly rather than going through the gen_server: a notification broadcast must not queue behind a registration.

Summary

Types

filter/0
handler_type/0

Functions

close_all()

Ask every open stream to end gracefully, so a client can tell a clean shutdown from a dropped connection.

count()
handle_call(Request, From, State)
handle_cast(Msg, State)
handle_info(Info, State)
init(_)
list_changed(Kind)

Fan a list-changed notification out to every subscriber that asked for that kind.

normalize_filter(Params)

Read a client's notification filter off the wire.

resource_updated(Uri, Extra)

Fan a resource update out to the subscribers watching that URI.

start_link()
subscribe(SubId, Filter)

Register the calling process as the holder of a subscription.

task_changed(TaskId, Owner, Task)

Fan a task status change out to the streams that asked for that task by id.

terminate(Reason, State)
unsubscribe(SubId)

Types

filter/0

-type filter() ::
          #{tools_list_changed => boolean(),
            prompts_list_changed => boolean(),
            resources_list_changed => boolean(),
            resource_subscriptions => [binary()],
            task_ids => [binary()],
            principal => term()}.

handler_type/0

-type handler_type() :: tool | resource | prompt | resource_template | completion.

Functions

close_all()

-spec close_all() -> ok.

Ask every open stream to end gracefully, so a client can tell a clean shutdown from a dropped connection.

count()

-spec count() -> non_neg_integer().

handle_call(Request, From, State)

handle_cast(Msg, State)

handle_info(Info, State)

init(_)

list_changed(Kind)

-spec list_changed(handler_type()) -> ok.

Fan a list-changed notification out to every subscriber that asked for that kind.

normalize_filter(Params)

-spec normalize_filter(map()) -> filter().

Read a client's notification filter off the wire.

Unknown keys are dropped rather than rejected: the acknowledgment tells the client what was actually honoured, which is the mechanism the spec gives for a server that does not support a type.

resource_updated(Uri, Extra)

-spec resource_updated(binary(), map()) -> ok.

Fan a resource update out to the subscribers watching that URI.

start_link()

subscribe(SubId, Filter)

-spec subscribe(term(), filter()) -> ok.

Register the calling process as the holder of a subscription.

Called from the request process that owns the response stream, so the monitor established here is what cleans up on disconnect.

task_changed(TaskId, Owner, Task)

-spec task_changed(binary(), term(), map()) -> ok.

Fan a task status change out to the streams that asked for that task by id.

terminate(Reason, State)

unsubscribe(SubId)

-spec unsubscribe(term()) -> ok.