# `Macaw.Data.Source.Push`
[🔗](https://github.com/dcoai/macaw_data/blob/v0.5.3/lib/macaw/data/source/push.ex#L1)

One push subscription, as a value (spec/01 §4).

A source that supports `:push` sends its subscriber two messages, both tagged with the
reference `subscribe/2` answered:

    {:macaw_source, ref, {:schema, schema}}      # first, and again when the schema changes
    {:macaw_source, ref, {:delivery, delivery}}  # an increment: the rows new since the last

`schema_message/2` and `delivery_message/2` build them. The subscriber holds a `Push`
in its state: `receive/2` takes each message (a schema is cached, a delivery buffered),
and `drain/2` merges everything buffered since the last pulse at once, so a burst of
pushes between two pulses costs one merge. `held/1` is the delivery as it stands.

- **The handshake sends the schema first.** A delivery names its schema's version; one
  naming a version not yet cached falls back to `describe` (`drain/2`'s function, the
  `Macaw.Data.Schema.Cache.resolve/4` describe function), asked once and cached.
- **An increment is merged by the window field:** its rows after what is held are
  appended; its late rows, at or before that, are kept unless the very row is held
  already; the rows stay ordered by the window field, the served window grown to cover
  them. A schema with no window field keeps the newest `limit` rows, marking the
  delivery `capped` when it drops any.

It is not a process: the process that subscribes receives the messages in its mailbox
and passes them here. Ported from the logic of visualize's `Visualize.Feed.Monitor`,
whose message tag was `:visualize_feed`.

# `t`

```elixir
@type t() :: %Macaw.Data.Source.Push{
  cache: Macaw.Data.Schema.Cache.t(),
  errors: [term()],
  held: Macaw.Data.Delivery.t() | nil,
  limit: pos_integer(),
  name: term(),
  pending: [Macaw.Data.Delivery.t()],
  ref: reference()
}
```

# `delivery_message`

```elixir
@spec delivery_message(reference(), Macaw.Data.Delivery.t()) :: tuple()
```

The message a source sends with an increment.

# `drain`

```elixir
@spec drain(t(), Macaw.Data.Schema.Cache.describe()) :: t()
```

Every buffered delivery merged into what is held, in the order received: one merge per
pulse, whatever the push rate. A delivery naming a version the cache lacks is resolved
through `describe` (`Schema.Cache.resolve/4`); one that cannot be is dropped and its
reason kept in `errors`. A delivery naming no version is read under the held schema's.

# `held`

```elixir
@spec held(t()) :: Macaw.Data.Delivery.t() | nil
```

The delivery as it stands, or `nil` before the first.

# `new`

```elixir
@spec new(term(), reference(), keyword()) :: t()
```

A subscription to the source named `name` under `ref`, holding nothing yet. Options:
`cache:` a `Schema.Cache` to start from, `limit:` the most rows held for a schema with
no window field (`100_000`).

# `receive`

```elixir
@spec receive(t(), term()) :: {:ok, t()} | :ignore
```

A message the subscriber received: `{:ok, push}` with a schema cached or a delivery
buffered, `:ignore` for a message that is not this subscription's.

# `schema_message`

```elixir
@spec schema_message(reference(), Macaw.Data.Schema.t()) :: tuple()
```

The message a source sends with its schema, first and on every change.

---

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