Parrhesia.API.Stream (parrhesia v0.12.0)

Copy Markdown

In-process subscription API with relay-equivalent catch-up and live fanout semantics.

Subscriptions are process-local bridges. After subscribing, the caller receives messages in the same order a relay client would expect:

  • {:parrhesia, :event, ref, subscription_id, event} for catch-up and live events
  • {:parrhesia, :eose, ref, subscription_id} after the initial replay finishes

This API requires a Parrhesia.API.RequestContext so read policies are applied exactly as they would be for a transport-backed subscriber.

Summary

Functions

subscribe(subscriber, subscription_id, filters, opts \\ [])

Starts an in-process subscription for a subscriber pid.

unsubscribe(ref)

Stops a subscription previously created with subscribe/4.

Functions

subscribe(subscriber, subscription_id, filters, opts \\ [])

@spec subscribe(pid(), String.t(), [map()], keyword()) ::
  {:ok, reference()} | {:error, term()}

Starts an in-process subscription for a subscriber pid.

opts[:context] must be a Parrhesia.API.RequestContext.

On success the returned reference is both:

  • the subscription handle used by unsubscribe/1
  • the value embedded in emitted subscriber messages

unsubscribe(ref)

@spec unsubscribe(reference()) :: :ok

Stops a subscription previously created with subscribe/4.

This function is idempotent. Unknown or already-stopped references return :ok.