Ranvier
KR

#Bus Access Patterns Cookbook

Version: 0.43.0 | Updated: 2026-03-25 | Applies to: ranvier-core 0.36+ | Category: Cookbook

#Overview

Bus is Ranvier's type-safe resource container. Every Transition receives a &mut Bus and uses it to share state across pipeline steps. This cookbook provides practical patterns for common Bus usage scenarios.

For the foundational decision guide, see Bus Access Patterns.

#1. Access Method Decision Matrix

Method Return On Missing When to Use
insert::<T>(value) -- overwrites Write a value for downstream steps
read::<T>() Option<&T> None Optional data -- can proceed without it
read_mut::<T>() Option<&mut T> None Modify an existing value in-place
get::<T>() Result<&T, BusAccessError> Err Production code needing error messages
get_mut::<T>() Result<&mut T, BusAccessError> Err Mutable access with error reporting
require::<T>() &T panic Framework guarantees existence
try_require::<T>() Option<&T> None Semantic alias for read()
has::<T>() bool false Check existence without borrowing
get_cloned::<T>() Result<T, BusAccessError> Err v0.43 Owned clone — replaces read().cloned()
remove::<T>() Option<T> None Take ownership, removing from Bus
provide::<T>(value) -- inserts Alias for insert, semantic emphasis on "providing"

#2. Newtype Wrappers for Collision Safety

Plain types like String or u64 collide when you store multiple values of the same type on the Bus. Always use newtype wrappers to avoid this.

#Problem: Type Collision

// BAD: Both insert String, the second overwrites the first
bus.insert::<String>("user-123".into());     // UserId?
bus.insert::<String>("order-456".into());    // OrderId? Overwrites UserId!

#Solution: Newtype Wrappers

#[derive(Debug, Clone)]
struct UserId(pub String);

#[derive(Debug, Clone)]
struct OrderId(pub String);

#[derive(Debug, Clone)]
struct TenantId(pub String);

// Each type occupies its own slot in the Bus
bus.insert(UserId("user-123".into()));
bus.insert(OrderId("order-456".into()));
bus.insert(TenantId("tenant-789".into()));

// Read each independently
let user = bus.read::<UserId>();    // Some(UserId("user-123"))
let order = bus.read::<OrderId>();  // Some(OrderId("order-456"))

#Convention

Define Bus types near the module that produces or consumes them. Guard-specific Bus types (such as RequestOrigin, ClientIdentity, CorsHeaders) are defined in ranvier-guard.

#3. Database Pool Sharing

Share a connection pool across all Transitions by inserting it into the Bus before the circuit executes.

use sqlx::PgPool;

#[derive(Debug, Clone)]
struct DbPool(pub PgPool);

// Insert pool via bus_injector (runs once per request)
Ranvier::http()
    .bus_injector({
        let pool = PgPool::connect(&database_url).await?;
        move |_parts: &http::request::Parts, bus: &mut Bus| {
            bus.insert(DbPool(pool.clone()));
        }
    })
    .get("/api/users", list_users_circuit)
    .run(())
    .await?;

Inside any Transition, read the pool:

#[async_trait]
impl Transition<(), Vec<User>, > for ListUsers {
    type Error = String;
    type Resources = ();

    async fn run(
        &self,
        _input: (),
        _resources: &Self::Resources,
        bus: &mut Bus,
    ) -> Outcome<Vec<User>, Self::Error> {
        let pool = &bus.require::<DbPool>().0;
        let users = sqlx::query_as::<_, User>("SELECT * FROM users")
            .fetch_all(pool)
            .await
            .map_err(|e| e.to_string())?;
        Outcome::next(users)
    }
}

Alternative: If the pool is the same for all requests, use Resources instead of the Bus. Resources are shared across the entire Axon lifetime, while the Bus is per-request.

#4. Cross-Transition Data Passing

The Bus is the standard way to pass data between Transitions in a pipeline. Earlier steps call insert; later steps call read.

use ranvier_core::prelude::*;
use ranvier_runtime::Axon;

#[derive(Debug, Clone)]
struct ValidationResult {
    is_valid: bool,
    warnings: Vec<String>,
}

#[derive(Debug, Clone)]
struct EnrichedOrder {
    original: CreateOrder,
    tax_amount: f64,
    shipping_cost: f64,
}

let order_pipeline = Axon::typed::<CreateOrder, String>("order-pipeline")
    // Step 1: Validate and write result to Bus
    .then_fn("validate", |order, _res, bus| async move {
        let result = ValidationResult {
            is_valid: order.quantity > 0,
            warnings: vec![],
        };
        bus.insert(result.clone());
        if result.is_valid {
            Outcome::next(order)
        } else {
            Outcome::fault("Validation failed".to_string())
        }
    })
    // Step 2: Enrich and write enriched data to Bus
    .then_fn("enrich", |order, _res, bus| async move {
        let enriched = EnrichedOrder {
            tax_amount: order.quantity as f64 * 0.08,
            shipping_cost: 5.99,
            original: order,
        };
        bus.insert(enriched.clone());
        Outcome::next(enriched)
    })
    // Step 3: Read both previous results
    .then_fn("finalize", |enriched, _res, bus| async move {
        let validation = bus.require::<ValidationResult>();
        let total = enriched.tax_amount + enriched.shipping_cost;
        Outcome::next(format!(
            "Order confirmed: ${:.2} total, {} warnings",
            total,
            validation.warnings.len()
        ))
    });

#5. Guard Bus Types

Guards use specific Bus types for HTTP-level communication. When writing custom Bus injectors or reading Guard output, reference these types directly.

#Reading Guard Output in Your Circuit

use ranvier_guard::prelude::*;

let circuit = Axon::simple::<String>("user-info")
    .then_fn("read-auth", |_input, _res, bus| async move {
        // AuthGuard inserts IamIdentity after successful authentication
        let identity = bus.get_cloned::<ranvier_core::iam::IamIdentity>();
        let subject = identity.map(|id| id.subject)
            .unwrap_or_else(|_| "anonymous".into());

        // RequestIdGuard inserts RequestId (UUID v4)
        let request_id = bus.get_cloned::<RequestId>()
            .map(|r| r.0)
            .unwrap_or_default();

        // CompressionGuard inserts CompressionConfig
        let encoding = bus.read::<CompressionConfig>()
            .map(|c| c.encoding.as_str())
            .unwrap_or("identity");

        Outcome::next(format!(
            "user={}, request={}, encoding={}",
            subject, request_id, encoding
        ))
    });

#6. Bus Access Policy

Restrict which types a Transition can read from or write to the Bus. This is useful for enforcing architectural boundaries.

use ranvier_core::bus::{Bus, BusAccessPolicy, BusTypeRef};

// Allow only specific types
let policy = BusAccessPolicy::allow_only(vec![
    BusTypeRef::of::<UserId>(),
    BusTypeRef::of::<DbPool>(),
]);
bus.set_access_policy(policy);

// Or deny specific sensitive types
let policy = BusAccessPolicy::deny_only(vec![
    BusTypeRef::of::<AuthorizationHeader>(),
]);
bus.set_access_policy(policy);

// Remove policy when done
bus.clear_access_policy();

When an access policy is active, attempts to read disallowed types return None and emit a tracing::warn! log entry.

#7. Testing with Bus

Use Bus::new() directly in unit tests to verify Transition behavior:

#[tokio::test]
async fn test_order_enrichment() {
    let mut bus = Bus::new();
    bus.insert(DbPool(test_pool().await));

    let transition = EnrichOrder;
    let input = CreateOrder {
        product_id: "prod-1".into(),
        quantity: 2,
        shipping_address: "123 Main St".into(),
    };

    let result = transition.run(input, &(), &mut bus).await;
    assert!(matches!(result, Outcome::Next(_)));

    // Verify Bus side effects
    let enriched = bus.get_cloned::<EnrichedOrder>().expect("should be in Bus");
    assert!(enriched.tax_amount > 0.0);
}

#9. `Bus::get_cloned()` — Owned Clone (v0.43)

get_cloned() returns an owned clone in a single call, replacing the common bus.read::<T>().cloned() pattern:

// BEFORE (v0.42):
let pool = bus.read::<PgPool>().cloned()
    .ok_or_else(|| "PgPool missing".to_string())?;

// AFTER (v0.43):
let pool = bus.get_cloned::<PgPool>().expect("PgPool");

// With try_outcome! for Outcome context:
use ranvier_core::try_outcome;
let pool = try_outcome!(bus.get_cloned::<PgPool>(), "PgPool not in Bus");

When to use: Any time you need an owned T from Bus and the type implements Clone. Replaces the 3-step read() → cloned() → unwrap/expect chain with a single method.

#10. `BusHttpExt` Trait — HTTP Parameter Extraction (v0.43)

The BusHttpExt trait (from ranvier_http) adds convenience methods for extracting path and query parameters from Bus in HTTP transitions:

use ranvier_http::BusHttpExt;  // or via prelude

// Path parameter (auto-parsed): /users/:id
let id: u64 = bus.path_param("id")?;  // Result<T: FromStr, String>

// Query parameter (optional): /items?page=2
let page: Option<i64> = bus.query_param("page");

// Query with default: /items?per_page=10
let per_page: i64 = bus.query_param_or("per_page", 10);

Before (manual PathParams extraction — 7 lines):

let id: u64 = match bus.read::<PathParams>().and_then(|p| p.get("id")) {
    Some(raw) => match raw.parse() {
        Ok(id) => id,
        Err(_) => return Outcome::Fault("Invalid ID".into()),
    },
    None => return Outcome::Fault("Missing ID".into()),
};

After (BusHttpExt — 4 lines):

let id: u64 = match bus.path_param("id") {
    Ok(id) => id,
    Err(e) => return Outcome::Fault(e),
};

#See Also

  • Bus Access Patterns -- Decision Guide -- foundational reference
  • Guard Patterns Cookbook -- Guard Bus wiring
  • Saga Compensation Cookbook -- Bus in saga contexts
  • Outcome Patterns Cookbook -- try_outcome!, combinators
  • JSON Outcomes Cookbook -- typed JSON at route boundary