Snodo.Client.Subscription (snodo v0.3.2)

A subscriptions/listen stream opened with Snodo.Client.listen/3.

The struct is the handle: accepted is the filter the server acknowledged, id the request ID (the subscription ID on the wire), ref the tag of the messages the stream sends, owner the process that opened it, and pid the transport process that receives the stream.

Messages

The owner receives {:snodo_subscription, ref, payload} with one of:

{:notification, method, params}: one event, as the server sent it. method is the notification's method, for example "notifications/resources/updated", "notifications/tools/list_changed", or an extension's such as "notifications/tasks"; params is its params map, "_meta" included.
{:dropped, count}: count events were discarded because the buffer was full. It precedes the next delivered event.
{:closed, reason}: the stream ended. :complete is the server's terminal result; {:error, %Snodo.Error{}} is its terminal error response, or a failure of the connection. Nothing follows it.

Demand and the buffer

Events are sent to the owner only while it has asked for them. demand/2 asks for n more, next/2 asks for one and waits for it, and stream/1 wraps next/2 as an Enumerable. Events that arrive without demand wait in the transport process, at most :max_buffer of them (100 by default). A full buffer follows the :overflow policy given to Snodo.Client.listen/3: :drop_oldest (the default) discards the oldest queued event, :drop_newest discards the arriving one. Either way the owner is told with {:dropped, n}. {:closed, reason} is delivered after the queued events.

Ending the stream

close/1 ends the stream from the client side; the server sees a cancellation. The owner's exit does the same. No message follows close/1; events delivered before it stay in the owner's mailbox.

Summary

Types

close_reason()

Why a stream ended: the server's terminal result, or an error.

payload()

What the owner receives inside {:snodo_subscription, ref, payload}.

t()

Functions

close(subscription)

Ends the stream.

demand(subscription, n)

Asks for n more events, which arrive as messages to the owner.

next(subscription, timeout \\ :infinity)

Asks for one event and waits for the next message.

stream(subscription)

The stream's payloads as an Enumerable, ending with {:closed, reason}.

Types

close_reason()

@type close_reason() :: :complete | {:error, Snodo.Error.t()}

Why a stream ended: the server's terminal result, or an error.

payload()

@type payload() ::
  {:notification, String.t(), map()}
  | {:dropped, pos_integer()}
  | {:closed, close_reason()}

What the owner receives inside {:snodo_subscription, ref, payload}.

t()

@type t() :: %Snodo.Client.Subscription{
  accepted: map(),
  id: integer(),
  owner: pid(),
  pid: pid(),
  ref: reference()
}

Functions

close(subscription)

@spec close(t()) :: :ok

Ends the stream.

The server is sent a cancellation: stdio writes notifications/cancelled, HTTP closes the connection, and the direct client closes the source. Returns :ok, also for a stream that has already ended.

demand(subscription, n)

@spec demand(t(), pos_integer()) :: :ok

Asks for n more events, which arrive as messages to the owner.

Demand accumulates: queued events are sent at once, up to n, and later events are sent as they arrive until the demand is used up.

next(subscription, timeout \\ :infinity)

@spec next(t(), timeout()) :: payload() | {:error, :timeout}

Asks for one event and waits for the next message.

Returns the payload: {:notification, method, params}, {:dropped, n}, or {:closed, reason}. Returns {:error, :timeout} when nothing arrives in timeout milliseconds; the demand stays, so that event arrives as a message later and the next call returns it. If the transport process exits, returns {:closed, {:error, %Snodo.Error{}}}. Must be called by the owner.

stream(subscription)

@spec stream(t()) :: Enumerable.t()

The stream's payloads as an Enumerable, ending with {:closed, reason}.

Each element is fetched with next/2, so the stream runs in the owner. Stopping early leaves the subscription open; call close/1.