# `Ankusa.Sink.Messaging`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/sink/messaging.ex#L1)

What a queue-style sink shares: `use` it, implement `c:publish/4`, and the
sink is an `Ankusa.Sink` whose `deliver/3` encodes the hook as
`Ankusa.Sink.Message` (inline up to `:inline_max_bytes`, a claim ticket
above it — `c:Ankusa.Sink.inline_max_bytes/1` is implemented for you) and
hands the payload to `publish/4`. Everything transport-specific — the
destination, headers (`Ankusa.Sink.Message.headers/2` /
`Ankusa.Sink.Message.attributes/2`), the size gate
(`Ankusa.Sink.Message.fits/2`), the connection (`Ankusa.Sink.Connections`)
and the error classes — is the sink's. `durable?/1` and `describe/2` are
still the sink's to add.

    defmodule MyApp.Sink.Queue do
      use Ankusa.Sink.Messaging

      @impl true
      def publish(payload, env, _ctx, opts) do
        MyApp.Queue.send(Keyword.fetch!(opts, :queue), payload, message_id: env.id)
      end
    end

An option that may differ per hook — a routing key, a subject, a record key
— is a `t:per_hook/1`: one string for every hook, or a function of the
`Ankusa.Envelope`. `resolve/2` is its value for one hook and `static/1` the
value a `describe/2` can name (`nil` when a function computes it).
`endpoints/1` reads a broker list in every shape an adapter accepts: a list
of `"host:port"` strings or `{host, port}` tuples, or one comma-separated
string from a YAML loader.

A sink that needs its own `deliver/3` implements `Ankusa.Sink` directly
instead, as `Ankusa.Sink.Http` does.

# `per_hook`

```elixir
@type per_hook(value) :: value | (Ankusa.Envelope.t() -> value)
```

A sink option that is one value for every hook, or a function of the hook.

# `max_message_bytes`

```elixir
@callback max_message_bytes(opts :: keyword()) :: pos_integer()
```

The largest message this sink publishes. Implemented for you: the
`:max_message_bytes` option, default 1 MiB (`Ankusa.Sink.Message.max_bytes/1`).
Override it for a transport with another limit or another option name.

`Ankusa.Sink.Message.encode/4` checks the message against it before
`publish/4` runs: a message over it goes to the claim check, and a body that
is over it by itself is `{:error, {:permanent, {:message_too_large, size,
max}}}` without a publish. A transport that counts more than the payload
(a key, attributes) still gates its own total with
`Ankusa.Sink.Message.fits/2`.

# `publish`

```elixir
@callback publish(
  payload :: binary(),
  Ankusa.Envelope.t(),
  Ankusa.Sink.ctx(),
  opts :: keyword()
) ::
  :ok | {:error, Ankusa.Sink.failure()}
```

Publish `payload`, the encoded `Ankusa.Sink.Message` for `env`. `:ok` only
once the transport confirmed it; otherwise `{:error, reason}` in the classes
`Ankusa.Sink` documents.

# `deliver`

```elixir
@spec deliver(module(), Ankusa.Envelope.t(), Ankusa.Sink.ctx(), keyword()) ::
  :ok | {:error, term()}
```

`c:Ankusa.Sink.deliver/3` for a sink that `use`s this module: encode, then
`mod.publish/4`. A failed claim check-in is `{:error, {:claim_check,
reason}}`, and a message over `c:max_message_bytes/1` that no claim check can
shrink is `{:error, {:permanent, {:message_too_large, size, max}}}`;
`publish/4` never runs for either.

# `endpoints`

```elixir
@spec endpoints(String.t() | [String.t() | {String.t() | charlist(), pos_integer()}]) ::
  [
    {String.t(), pos_integer()}
  ]
```

Broker endpoints as `{host, port}` pairs with an integer port, from a list
of `"host:port"` strings and/or `{host, port}` tuples, or one
comma-separated `"host:port,host:port"` string (what a YAML loader hands
over). Raises `ArgumentError` on an entry with no port.

# `resolve`

```elixir
@spec resolve(per_hook(String.t()) | nil, Ankusa.Envelope.t()) :: String.t() | nil
```

A `t:per_hook/1` option's value for `env`; `nil` stays `nil`.

# `static`

```elixir
@spec static(per_hook(String.t()) | nil) :: String.t() | nil
```

The one value a `t:per_hook/1` option has for every hook, or `nil` when a
function computes it, so a description cannot name it.

---

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