Orkestra.MessageBus behaviour (orkestra v0.2.0)

Copy Markdown View Source

Behaviour for dispatching commands and publishing events.

Two implementations are provided:

  • MessageBus.PubSub — in-process via Phoenix.PubSub (dev, test, single-node)
  • MessageBus.RabbitMQ — distributed via RabbitMQ (production, multi-node)

Topic strategy

Topics are derived automatically from the command/event module name:

Ultimus.Tasks.Commands.StartAssessment → "tasks.commands.start_assessment"
Ultimus.Tasks.Events.AssessmentCompleted → "tasks.events.assessment_completed"

The convention is: drop the app prefix, downcase, underscore, dot-separated.

Usage

alias Orkestra.MessageBus

# Get the configured implementation
bus = MessageBus.impl()

# Dispatch a command (point-to-point, one handler)
:ok = bus.dispatch(command_envelope)

# Publish an event (broadcast, many handlers)
:ok = bus.publish(event_envelope)

Handler behaviour

Handlers implement handle/1:

defmodule MyHandler do
  @behaviour Orkestra.MessageBus.Handler

  @impl true
  def handle(envelope) do
    # process...
    :ok
  end
end

Return :ok to ack, {:error, reason} to nack.

Summary

Types

topic()

Callbacks

dispatch(t)
publish(t)
subscribe_command(topic, module)
subscribe_event(topic, module, keyword)

Functions

impl()

Returns the configured MessageBus implementation.

topic_for(module)

Derives a topic from a command or event struct module. Strips the configured app prefix (default: none) from the module name.

topic_for_envelope(arg1)

Derives a topic from an envelope (command or event).

Types

topic()

@type topic() :: String.t()

Callbacks

dispatch(t)

@callback dispatch(Orkestra.CommandEnvelope.t()) :: :ok | {:error, term()}

publish(t)

@callback publish(Orkestra.EventEnvelope.t()) :: :ok | {:error, term()}

subscribe_command(topic, module)

@callback subscribe_command(topic(), module()) :: :ok | {:error, term()}

subscribe_event(topic, module, keyword)

@callback subscribe_event(topic(), module(), keyword()) :: :ok | {:error, term()}

Functions

impl()

@spec impl() :: module()

Returns the configured MessageBus implementation.

topic_for(module)

@spec topic_for(module()) :: topic()

Derives a topic from a command or event struct module. Strips the configured app prefix (default: none) from the module name.

# config :orkestra, Orkestra.MessageBus, app_prefix: MyApp
iex> MessageBus.topic_for(MyApp.Tasks.Commands.StartAssessment)
"tasks.commands.start_assessment"

topic_for_envelope(arg1)

@spec topic_for_envelope(Orkestra.CommandEnvelope.t() | Orkestra.EventEnvelope.t()) ::
  topic()

Derives a topic from an envelope (command or event).