module Rooibos::Command

Creates an aggregating parallel command.

Applications load dashboards that combine user, settings, and stats. Fire-and-forget loses correlation. This command waits for all children and returns their results together in a single message.

commands

One or more commands to run in parallel. Pass multiple

arguments or a single array.

Example

# Variadic syntax
Command.all(
  Command.http(:get, "/users", :_),
  Command.http(:get, "/stats", :_),
)
# Produces: [:all, [user_result, stats_result]]

Creates a parallel batch command.

Applications fetch data from multiple sources. Dashboard panels load users, stats, and notifications. Waiting sequentially is slow. Managing threads and error handling manually is error-prone.

This command runs children in parallel. Each child sends its own messages independently. The batch completes when all children finish or when cancellation fires.

Use it for parallel fetches, concurrent refreshes, or any work that does not need coordinated results.

commands

One or more commands to run in parallel. Pass multiple

arguments or a single array.

Example

# Variadic syntax
Command.batch(
  Command.http(:get, "/users", :users),
  Command.http(:get, "/stats", :stats),
)

# Array syntax
Command.batch([cmd1, cmd2, cmd3])

Bubbles a message outward through the fragment hierarchy.

Nested fragments produce results. Sometimes those results belong to an outer fragment. Passing callbacks or references inward couples fragments tightly. The hierarchy becomes rigid.

This command wraps a message for outward propagation. Outer fragments intercept the bubble and decide how to handle it. With the Router DSL, use observe or intercept. Without the Router, check for Command::Bubble manually and extract the message.

Use it for notifications, validation results, or any signal that flows from nested fragments to outer containers.

Example (Router DSL)

# Nested fragment signals completion
class TaskComplete < Data.define(:envelope, :task_id)
  include Rooibos::Message::Predicates
end

# Return from nested Update
[model, Command.bubble(TaskComplete.new(envelope: :task, task_id: 42))]

# Outer Router observes the bubble
observe TaskComplete do |model, message|
  model.with(completed_tasks: model.completed_tasks + [message.task_id])
end

Example (Manual Bubbling)

# Outer Update handles bubbles without Router
def self.handle_nested_result(cmd, model)
  return [model, nil] unless cmd.is_a?(Command::Bubble)

  case cmd.message
  when TaskComplete
    [model.with(completed_tasks: model.completed_tasks + [cmd.message.task_id]), nil]
  else
    [model, cmd]  # Re-bubble outward
  end
end

Request cancellation of a running command.

The model stores the command handle (the command object itself). Returning Command.cancel(handle) signals the runtime to stop it.

handle

The command object to cancel.

Example

# Dispatch and store handle
cmd = FetchData.new(url)
[model.with(active_fetch: cmd), cmd]

# User clicks cancel
when :cancel_clicked
  [model.with(active_fetch: nil), Command.cancel(model.active_fetch)]

Gives a callable unique identity for cancellation.

Reusable procs and lambdas share identity. Dispatch them twice, and Command.cancel would cancel both. Wrap them to get distinct handles.

The callable must be Ractor-shareable (cannot capture mutable state). Create resources like database connections inside the callable, not in the closure. See the Custom Commands guide for details.

callable

Proc, lambda, or any object responding to +call(out, token)+. If omitted, the block is used.

grace_period

Cleanup time override. Default: 2.0 seconds.

Example

# With callable
cmd = Command.custom(->(out, token) { out.put(:fetched, data) })

# With block
cmd = Command.custom(grace_period: 5.0) do |out, token|
    until token.canceled?
    out.put(:tick, Time.now)
    sleep 1
  end
end

Delivers a message to Update.

Custom commands produce results. Those results feed back into your update function. This factory method wraps a message in a command that delivers it when executed.

Example

# Define a message type
class FetchComplete < Data.define(:envelope, :data)
  include Rooibos::Message::Predicates
end

# Send after a synchronous operation
result = fetch_data_sync()
[model, Command.deliver(FetchComplete.new(envelope: :items, data: result))]

# Receive in Update
in { type: :fetch_complete, envelope: :items, data: }
  model.with(items: data)

Creates a quit command.

Returns a sentinel the runtime detects to terminate the application.

Example

def update(message, model)
  case message
  in { type: :key, code: "q" }
    [model, Command.exit]
  else
    [model, nil]
  end
end

Creates an HTTP request command. Supports DWIM arity - see Http.new for patterns.

Creates a mapped command for Fractal Architecture composition.

Wraps an inner command. When the inner command completes, the mapper block transforms the result into a parent message. This prevents monolithic update functions (the “God Reducer” anti-pattern).

inner_command

The child command to wrap.

mapper

Block that transforms child message to parent message.

Example

# Child returns Command.execute that produces [:got_files, {...}]
child_command = Command.system("ls", :got_files)

# Parent wraps to route as [:sidebar, :got_files, {...}]
parent_command = Command.map(child_command) { |child_result| [:sidebar, *child_result] }

Opens a file or URL with the system’s default application. Cross-platform: uses open on macOS, xdg-open on Linux, start on Windows.

On success (exit 0), sends Message::Open. On failure (non-zero), sends Message::Error.

Example

case message
in { type: :open, envelope: path }
  model.with(status: "Opened #{path}")
in { type: :error, envelope: path }
  model.with(error: "Could not open #{path}")
end

Creates a shell execution command.

command

Shell command string to execute.

tag

Symbol or class to tag the result message.

stream

If true, the runtime sends incremental stdout/stderr

messages as they arrive. If <tt>false</tt> (default), waits for
completion and sends a single message with all output.

Example (Batch Mode)

# Return this from update:
[model.with(loading: true), Command.system("ls -la", :got_files)]

# Then handle it later:
def update(message, model)
  case message
  in { type: :system, envelope: :got_files, stdout:, status: 0 }
    [model.with(files: stdout.lines), nil]
  in { type: :system, envelope: :got_files, stderr:, status: }
    [model.with(error: stderr), nil]
  end
end

Example (Streaming Mode)

# Return this from update:
[model.with(loading: true), Command.system("tail -f log.txt", :log, stream: true)]

# Then handle incremental messages:
def update(message, model)
  case message
  in { type: :system, envelope: :log, stream: :stdout, content: line }
    [model.with(lines: [*model.lines, line]), nil]
  in { type: :system, envelope: :log, stream: :stderr, content: line }
    [model.with(errors: [*model.errors, line]), nil]
  in { type: :system, envelope: :log, stream: :complete, status: }
    [model.with(loading: false, exit_status: status), nil]
  in { type: :system, envelope: :log, stream: :error, content: msg }
    [model.with(loading: false, error: msg), nil]
  end
end

Creates a fresh cancellation that never fires.

Some I/O operations cannot be canceled mid-execution. Ruby’s Net::HTTP blocks until completion or timeout — there is no way to interrupt it.

A shared singleton would be unsafe. If any code path accidentally resolves the origin, all commands using it become canceled.

Use it for commands that wrap non-cancellable blocking I/O.

Example

token = Command.uncancellable
HttpCommand.new(url).call(outlet, token)

Creates a one-shot timer command.

Waits for seconds then sends TimerResponse to the update function. Use for delayed actions like notification dismissal or debounced search.

seconds

Duration to wait (Float or Integer).

envelope

Symbol to tag the result message.