StatifierRouter.Broadway (StatifierRouter v0.1.0)

The Broadway front: a pipeline that hands every message its producer emits to StatifierRouter.route/3.

The host starts it in its own supervision tree, with any Broadway producer it already operates; this package ships no producer and starts no process of its own.

children = [
  MyApp.Repo,
  {StatifierRouter.Broadway,
   name: MyApp.AdEventsRouter,
   producer: {BroadwayKafka.Producer, kafka_opts},
   router: router_config,
   processors: [default: [concurrency: 8]]}
]

Options

start_link/1 takes a keyword list:

Option	Value	Default
`:name`	the pipeline's name, an atom	required
`:producer`	`{module, opts}`: any Broadway producer and its options	required
`:router`	a `%StatifierRouter.Config{}`	required
`:normalize`	a fun of one `%Broadway.Message{}` returning the event `StatifierRouter.route/3` takes	`normalize/1`
`:processors`	Broadway's `:processors` option, passed through	`[default: []]`

Any other key, or a value the table does not allow, raises ArgumentError, as Broadway.start_link/2 does for its own options. There are no batchers in this release.

Each message

handle_message/3 normalizes the message and calls StatifierRouter.route/3, which returns once every delivery's transaction has ended. {:ok, outcomes} returns the message unchanged, so Broadway acknowledges it after the transactions committed. {:error, reason} returns it failed with reason, so it is not acknowledged and the source hands it over again. A raise inside route/3 is not rescued here either: Broadway fails the message it raised on, and it is not acknowledged (ADR-0003, section 1).

Partitioning

partition/3 is the pipeline's partition_by. It keeps every message for one address on one processor, so deliveries to one execution reach it one after another instead of queueing on its lock, each holding a pooled connection while it waits. That is an optimisation and never the guarantee: two deliveries to one execution are stepped one at a time by statifier_persistence's per-execution lock, whichever processor they came through (ADR-0003, section 5). A binding whose order is :none is not partitioned by its key (ADR-0003, section 10).

Summary

Functions

normalize(message)

The default :normalize: the event StatifierRouter.route/3 takes, read from the message. :scope, :message_id and :source are read from the message's metadata, and the message's data is the event's data.

partition(message, router, normalize)

The partition of one message: the hash, by :erlang.phash2/1, of the address {scope, document, key} of the first binding in router's order that is enabled, is for the event's source, has order: :by_key, and whose match holds and key produces a key for the event.

start_link(opts)

Starts the pipeline, linked to the calling process. The module documentation lists the options.

Functions

normalize(message)

@spec normalize(Broadway.Message.t()) :: map()

A message that does not carry all four is not refused here: StatifierRouter.route/3 refuses the event it produces with {:error, {:invalid_event, event}}, and the message fails. A producer whose messages carry them elsewhere is paired with a :normalize of the host's own.

iex> message = %Broadway.Message{
...>   data: %{"kind" => "click", "impression_id" => "imp_7f3a"},
...>   metadata: %{scope: "7c1e", message_id: "ad_events/3/1107", source: "ad_events"},
...>   acknowledger: Broadway.NoopAcknowledger.init()
...> }
iex> StatifierRouter.Broadway.normalize(message)
%{
  scope: "7c1e",
  message_id: "ad_events/3/1107",
  source: "ad_events",
  data: %{"kind" => "click", "impression_id" => "imp_7f3a"}
}

partition(message, router, normalize)

@spec partition(
  Broadway.Message.t(),
  StatifierRouter.Config.t(),
  (Broadway.Message.t() -> map())
) ::
  non_neg_integer()

A message no such binding addresses is partitioned by the hash of its message id. So is a message normalize builds no routable event from, one StatifierRouter.route/3 would refuse: a partitioner answers for every message, and handle_message/3 is where such a message fails.

normalize is called here as well as in handle_message/3, and here it runs in the producer's dispatcher, where a raise takes the producer down with the messages it holds. It is not rescued: a :normalize must answer for every message, and the default one does.

start_link(opts)

@spec start_link(keyword()) :: Broadway.on_start()

Starts the pipeline, linked to the calling process. The module documentation lists the options.