What a queue-style sink shares: use it, implement 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 — 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
endAn option that may differ per hook — a routing key, a subject, a record key
— is a 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.
Summary
Types
A sink option that is one value for every hook, or a function of the hook.
Callbacks
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.
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.
Functions
Ankusa.Sink.deliver/3 for a sink that uses this module: encode, then
mod.publish/4. A failed claim check-in is {:error, {:claim_check, reason}}, and a message over max_message_bytes/1 that no claim check can
shrink is {:error, {:permanent, {:message_too_large, size, max}}};
publish/4 never runs for either.
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.
A per_hook/1 option's value for env; nil stays nil.
The one value a per_hook/1 option has for every hook, or nil when a
function computes it, so a description cannot name it.
Types
@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.
Callbacks
@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.
@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.
Functions
@spec deliver(module(), Ankusa.Envelope.t(), Ankusa.Sink.ctx(), keyword()) :: :ok | {:error, term()}
Ankusa.Sink.deliver/3 for a sink that uses this module: encode, then
mod.publish/4. A failed claim check-in is {:error, {:claim_check, reason}}, and a message over max_message_bytes/1 that no claim check can
shrink is {:error, {:permanent, {:message_too_large, size, max}}};
publish/4 never runs for either.
@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.
@spec resolve(per_hook(String.t()) | nil, Ankusa.Envelope.t()) :: String.t() | nil
A per_hook/1 option's value for env; nil stays nil.
The one value a per_hook/1 option has for every hook, or nil when a
function computes it, so a description cannot name it.