# `Snodo.Client.Subscription`
[🔗](https://github.com/joshrotenberg/snodo/blob/v0.3.2/lib/snodo/client/subscription.ex#L1)

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.

# `close_reason`

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

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

# `payload`

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

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

# `t`

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

# `close`

```elixir
@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`

```elixir
@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`

```elixir
@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`

```elixir
@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`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*