Jido.Bus (Jido v1.1.0-rc)

View Source

Use the signal store configured for a Commanded application.

Telemetry Signals

Adds telemetry signals for the following functions. Signals are emitted in the form

[:jido, :bus, signal] with their spannable postfixes (start, stop, exception)

  • ack/3
  • adapter/2
  • publish/4
  • delete_snapshot/2
  • unsubscribe/3
  • read_snapshot/2
  • record_snapshot/2
  • replay/2
  • replay/3
  • replay/4
  • subscribe/2
  • subscribe_persistent/5
  • subscribe_persistent/6
  • unsubscribe/2

Summary

Types

start_option()
t()

Functions

ack(name, subscription, signal)

Acknowledge receipt and successful processing of a signal.

child_spec(init_arg)

Start a bus as part of a supervision tree.

delete_snapshot(name, source_id)

Delete a previously recorded snapshot for a given source

publish(name, stream_id, expected_version, signals, opts \\ [])

Append one or more signals to a stream atomically.

read_snapshot(name, source_id)

Read a snapshot, if available, for a given source.

record_snapshot(name, snapshot)

Record a snapshot of the data and metadata for a given source

replay(name, stream_id, start_version \\ 0, read_batch_size \\ 1000)

Streams signals from the given stream, in the order in which they were originally written.

start_link(opts)

Start a bus process.

subscribe(name, stream_id)

Create a transient subscription to a single signal stream.

subscribe_persistent(name, stream_id, subscription_name, subscriber, start_from, opts \\ [])

Create a persistent subscription to an signal stream.

unsubscribe(name, subscription)

Unsubscribe an existing subscriber from signal notifications.

unsubscribe(name, subscribe_persistent, handler_name)

Delete an existing subscription.

via_tuple(name, opts \\ [])

Returns a via tuple for addressing the bus process.

whereis(name, opts \\ [])

Gets the PID of a running bus by name.

Types

start_option()

@type start_option() ::
  {:name, atom()}
  | {:adapter, :pubsub | :in_memory | module()}
  | {:pubsub_name, atom()}
  | {atom(), term()}

t()

@type t() :: %Jido.Bus{
  adapter: module() | nil,
  adapter_meta: map() | nil,
  config: Keyword.t(),
  id: String.t(),
  name: atom() | nil
}

Functions

ack(name, subscription, signal)

Acknowledge receipt and successful processing of a signal.

child_spec(init_arg)

Start a bus as part of a supervision tree.

Options

  • :name - Required. The name to register the bus process under
  • :adapter - The adapter type. Either :pubsub, :in_memory, or a custom module
  • Other options are passed to the chosen adapter

Examples 

children = [
  {Jido.Bus, name: :jido_agent_123, adapter: :pubsub, pubsub_name: MyApp.PubSub}
]

delete_snapshot(name, source_id)

Delete a previously recorded snapshot for a given source

publish(name, stream_id, expected_version, signals, opts \\ [])

Append one or more signals to a stream atomically.

read_snapshot(name, source_id)

Read a snapshot, if available, for a given source.

record_snapshot(name, snapshot)

Record a snapshot of the data and metadata for a given source

replay(name, stream_id, start_version \\ 0, read_batch_size \\ 1000)

Streams signals from the given stream, in the order in which they were originally written.

start_link(opts)

Start a bus process.

See child_spec/1 for options.

subscribe(name, stream_id)

Create a transient subscription to a single signal stream.

subscribe_persistent(name, stream_id, subscription_name, subscriber, start_from, opts \\ [])

Create a persistent subscription to an signal stream.

unsubscribe(name, subscription)

Unsubscribe an existing subscriber from signal notifications.

unsubscribe(name, subscribe_persistent, handler_name)

Delete an existing subscription.

via_tuple(name, opts \\ [])

Returns a via tuple for addressing the bus process.

Options

  • :registry - The registry to use (defaults to Jido.BusRegistry)

Examples 

iex> Jido.Bus.via_tuple(:my_bus)
{:via, Registry, {Jido.BusRegistry, :my_bus}}

iex> Jido.Bus.via_tuple(:my_bus, registry: MyApp.Registry)
{:via, Registry, {MyApp.Registry, :my_bus}}

whereis(name, opts \\ [])

Gets the PID of a running bus by name.

Returns {:ok, pid} if found, {:error, :not_found} otherwise.