Jido.
View Source
A test utility for observing signals crossing process boundaries via telemetry events.
The BusSpy allows test processes to capture signals that travel across process boundaries through the Signal Bus without interfering with normal signal delivery. It integrates cleanly with existing cross-process test infrastructure.
Usage
test "cross-process signal observation" do
# Start the spy to capture bus events
spy = BusSpy.start_spy()
# Set up your cross-process test scenario
%{producer: producer, consumer: consumer} = setup_cross_process_agents()
# Send a signal that will cross process boundaries
send_signal_sync(producer, :root, %{test_data: "cross-process"})
# Wait for completion
wait_for_cross_process_completion([consumer])
# Verify the signal was observed crossing the bus
dispatched_signals = BusSpy.get_dispatched_signals(spy)
assert length(dispatched_signals) == 1
[signal_event] = dispatched_signals
assert signal_event.signal.type == "child.event"
assert signal_event.signal.data.test_data == "cross-process"
# Verify trace context was preserved
assert signal_event.signal.trace_context != nil
BusSpy.stop_spy(spy)
endEvents Captured
The spy captures these telemetry events:
[:jido, :signal, :bus, :before_dispatch]- Before signal dispatch[:jido, :signal, :bus, :after_dispatch]- After successful dispatch[:jido, :signal, :bus, :dispatch_skipped]- When middleware skips dispatch[:jido, :signal, :bus, :dispatch_error]- When dispatch fails
The spy prefers full signal and subscription metadata from telemetry when available and falls back to bounded metadata if needed.
Summary
Functions
Returns a specification to start this module under a supervisor.
Clears all captured events from the spy.
Gets all signals that have been dispatched through the bus since the spy started.
Gets the most recent signal event for a specific bus.
Gets signals that match a specific signal type pattern.
Callback implementation for GenServer.init/1.
Starts a new bus spy process to collect telemetry events.
Stops a bus spy process and cleans up telemetry handlers.
Waits for a signal matching the given type pattern to be dispatched.
Types
@type signal_event() :: %{ event: atom(), timestamp: integer(), bus_name: atom(), signal_id: String.t(), signal_type: String.t(), subscription_id: String.t(), subscription_path: String.t(), signal: Jido.Signal.t(), subscription: map(), dispatch_result: term() | nil, error: term() | nil, reason: atom() | nil }
@type spy_ref() :: pid()
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec clear_events(spy_ref()) :: :ok
Clears all captured events from the spy.
@spec get_dispatched_signals(spy_ref()) :: [signal_event()]
Gets all signals that have been dispatched through the bus since the spy started.
Returns a list of signal events in chronological order.
@spec get_latest_signal(spy_ref(), atom()) :: signal_event() | nil
Gets the most recent signal event for a specific bus.
@spec get_signals_by_type(spy_ref(), String.t()) :: [signal_event()]
Gets signals that match a specific signal type pattern.
Examples
# Get all "user.*" events
user_events = BusSpy.get_signals_by_type(spy, "user.*")
# Get exact matches
child_events = BusSpy.get_signals_by_type(spy, "child.event")Callback implementation for GenServer.init/1.
@spec start_spy() :: spy_ref()
Starts a new bus spy process to collect telemetry events.
Returns a spy reference that can be used to query captured events.
@spec stop_spy(spy_ref()) :: :ok
Stops a bus spy process and cleans up telemetry handlers.
@spec wait_for_signal(spy_ref(), String.t(), non_neg_integer()) :: {:ok, signal_event()} | :timeout
Waits for a signal matching the given type pattern to be dispatched.
Returns the matching signal event or times out.