FixAlchemy.Dispatch (FIXAlchemy v0.2.0)

View Source

Keyed pub/sub subscription state for one session.

Subscribers (processes) register interest in a message type as one of:

  • all — every message of that type (subscribe(d, pid, "8"))
  • exact — messages where a tag equals a value (subscribe(d, pid, "W", "EUR/USD"))
  • predicate — messages where a tag satisfies a function (subscribe(d, pid, "8", tag: 271, where: &(String.to_integer(&1) > 10_000)))

Pure state: every function returns a new struct, and delivery is the caller's job.

Routing tag resolution

For a keyed or predicate subscription the routing tag resolves, most specific first: the subscription's :tag option, then the connection's routing_keys override for that message type, then the shipped default for that type, then the fallback (Symbol, tag 55).

Summary

Types

predicate()
t()
tag()
type_subs()

Functions

default_routing_keys()

The protocol-level default routing tags, overridable per connection.

new(overrides \\ %{})

Build dispatch state; overrides merge over the shipped defaults.

recipients(dispatch, type, values)

Pids to deliver a message of type to, given values (%{tag => wire_value} extracted for the subscribed tags).

remove(dispatch, pid)

Remove pid from every subscription (e.g. on its :DOWN).

subscribe(dispatch, pid, type, value \\ :all, opts \\ [])

Subscribe pid to type.

subscribed?(dispatch, type)

Whether any subscriber wants this message type.

tags_for(dispatch, type)

The distinct routing tags subscribed for a type (the tags to scan).

Types

predicate()

@type predicate() :: (binary() -> boolean())

t()

@type t() :: %FixAlchemy.Dispatch{
  routing_keys: %{required(binary()) => tag()},
  subs: %{required(binary()) => type_subs()}
}

tag()

@type tag() :: pos_integer()

type_subs()

@type type_subs() :: %{
  all: MapSet.t(pid()),
  exact: %{required({tag(), binary()}) => MapSet.t(pid())},
  preds: [{tag(), predicate(), pid()}]
}

Functions

default_routing_keys()

@spec default_routing_keys() :: %{required(binary()) => tag()}

The protocol-level default routing tags, overridable per connection.

new(overrides \\ %{})

@spec new(%{required(binary()) => tag()}) :: t()

Build dispatch state; overrides merge over the shipped defaults.

recipients(dispatch, type, values)

@spec recipients(t(), binary(), %{required(tag()) => binary()}) :: MapSet.t(pid())

Pids to deliver a message of type to, given values (%{tag => wire_value} extracted for the subscribed tags).

remove(dispatch, pid)

@spec remove(t(), pid()) :: t()

Remove pid from every subscription (e.g. on its :DOWN).

subscribe(dispatch, pid, type, value \\ :all, opts \\ [])

@spec subscribe(t(), pid(), binary(), binary() | :all, keyword()) :: t()

Subscribe pid to type.

  • subscribe(d, pid, type) — all messages of that type
  • subscribe(d, pid, type, value) — where the routing tag equals value
  • subscribe(d, pid, type, :all, tag: t, where: fun) — where fun.(tag_value)

subscribed?(dispatch, type)

@spec subscribed?(t(), binary()) :: boolean()

Whether any subscriber wants this message type.

tags_for(dispatch, type)

@spec tags_for(t(), binary()) :: [tag()]

The distinct routing tags subscribed for a type (the tags to scan).