Trait Handler 

pub trait Handler<E> {
    // Required method
    fn run(&mut self, world: &mut World, event: E);

    // Provided methods
    fn inputs_changed(&self, world: &World) -> bool { ... }
    fn name(&self) -> &'static str { ... }
}

Object-safe dispatch trait for event handlers.

Enables Box<dyn Handler<E>> for type-erased heterogeneous dispatch. Storage and scheduling are the driver’s responsibility — this trait only defines the dispatch interface.

Takes &mut World — drivers call this directly in their poll loop.

Required Methods

fn run(&mut self, world: &mut World, event: E)

Run this handler with the given event.

Provided Methods

fn inputs_changed(&self, world: &World) -> bool

Returns true if any input resource was modified this sequence.

Default returns true (always run). Callback overrides by checking its state via SystemParam::any_changed.

Examples found in repository

examples/mock_runtime.rs (line 243)

fn main() {
    // -- Build ----------------------------------------------------------------

    let mut wb = WorldBuilder::new();
    wb.install_plugin(TradingPlugin {
        initial_prices: vec![("BTC", 50_000.0), ("ETH", 3_000.0)],
        risk_cap: 100,
    });
    let mut md = wb.install_driver(MarketDataInstaller);
    let mut world = wb.build();

    // Standalone handler — demonstrates change detection.
    fn on_signal(signals: Res<SignalBuffer>, _event: ()) {
        black_box(signals.signals.len());
    }
    let mut signal_handler = on_signal.into_handler(world.registry_mut());

    // -- Correctness check ----------------------------------------------------

    let ticks = [
        MarketTick {
            symbol: "BTC",
            price: 50_100.0,
        }, // delta=100 > 50 → signal
        MarketTick {
            symbol: "ETH",
            price: 3_010.0,
        }, // delta=10 < 50 → no signal
        MarketTick {
            symbol: "BTC",
            price: 49_900.0,
        }, // delta=200 > 50 → signal
    ];

    md.poll(&mut world, &ticks);

    // 2 signals accepted, risk cap=100 so both go through.
    assert_eq!(world.resource::<OrderCount>().0, 2);

    // Change detection: SignalBuffer was modified → handler should run.
    assert!(signal_handler.inputs_changed(&world));
    signal_handler.run(&mut world, ());

    // Advance sequence, no new writes → handler should skip.
    world.next_sequence();
    assert!(!signal_handler.inputs_changed(&world));

    println!("Correctness checks passed.\n");

    // -- Latency measurement --------------------------------------------------

    const WARMUP: usize = 1_000;
    const ITERATIONS: usize = 1_000;

    // Ticks that exercise the full pipeline path (signal detection + cache write).
    let bench_ticks = [
        MarketTick {
            symbol: "BTC",
            price: 50_100.0,
        },
        MarketTick {
            symbol: "ETH",
            price: 3_100.0,
        },
        MarketTick {
            symbol: "BTC",
            price: 50_200.0,
        },
        MarketTick {
            symbol: "ETH",
            price: 3_200.0,
        },
    ];

    println!(
        "=== nexus-rt Dispatch Latency (cycles, {} iterations) ===\n",
        ITERATIONS
    );
    println!(
        "{:<44} {:>8} {:>8} {:>8}",
        "Operation", "p50", "p99", "p999"
    );
    println!("{}", "-".repeat(72));

    // Single tick through dyn pipeline
    {
        let tick = bench_ticks[0];
        for _ in 0..WARMUP {
            world.next_sequence();
            md.pipeline.run(&mut world, black_box(tick));
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            world.next_sequence();
            let start = rdtsc_start();
            md.pipeline.run(&mut world, black_box(tick));
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        report("single tick (dyn pipeline, 3 stages)", &mut samples);
    }

    // Standalone handler (1 param, Res<T>)
    {
        for _ in 0..WARMUP {
            black_box(());
            signal_handler.run(&mut world, ());
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            let start = rdtsc_start();
            black_box(());
            signal_handler.run(&mut world, ());
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        report("handler dispatch (1 param, Res<T>)", &mut samples);
    }

    // 4-tick batch through driver poll
    {
        for _ in 0..WARMUP {
            md.poll(&mut world, &bench_ticks);
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            let start = rdtsc_start();
            md.poll(&mut world, black_box(&bench_ticks));
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        let mut per_tick: Vec<u64> = samples
            .iter()
            .map(|&s| s / bench_ticks.len() as u64)
            .collect();
        report("4-tick poll (total)", &mut samples);
        report("4-tick poll (per tick)", &mut per_tick);
    }

    // Change detection
    {
        world.next_sequence(); // ensure stale
        for _ in 0..WARMUP {
            black_box(signal_handler.inputs_changed(&world));
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            let start = rdtsc_start();
            black_box(signal_handler.inputs_changed(&world));
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        report("inputs_changed (1 param, stale)", &mut samples);
    }

    println!();
}

examples/perf_pipeline.rs (line 544)

fn main() {
    let mut wb = WorldBuilder::new();
    wb.register::<u64>(42);
    wb.register::<u32>(7);
    let mut world = wb.build();
    let r = world.registry_mut();

    // --- Bare 3-stage pipeline (no Option, no World access) ---

    let mut bare = PipelineStart::<u64>::new()
        .stage(|x: u64| x.wrapping_mul(3), r)
        .stage(|x: u64| x.wrapping_add(7), r)
        .stage(|x: u64| x >> 1, r);

    // --- Option 3-stage pipeline ---

    let mut option = PipelineStart::<u64>::new()
        .stage(
            |x: u64| -> Option<u64> { if x > 0 { Some(x) } else { None } },
            r,
        )
        .map(|x: u64| x.wrapping_mul(3), r)
        .filter(|_w, x| *x < 1_000_000);

    // --- World-accessing pipeline (pre-resolved via Res<T>) ---

    let mut world_resolved = PipelineStart::<u64>::new()
        .stage(add_resource, r)
        .stage(mul_resource, r);

    // --- World-accessing 3-stage pipeline ---

    let mut stage_3 = PipelineStart::<u64>::new()
        .stage(add_resource, r)
        .stage(mul_resource, r)
        .stage(sub_resource, r);

    // --- Built (boxed) pipeline ---

    let mut boxed = PipelineStart::<u64>::new()
        .stage(|x: u64| x.wrapping_mul(3), r)
        .stage(|x: u64| x.wrapping_add(7), r)
        .stage(|_x: u64| {}, r)
        .build();

    // --- Batch pipelines (same chains as their linear counterparts) ---

    fn sink(mut acc: ResMut<u64>, x: u64) {
        *acc = acc.wrapping_add(x);
    }

    // Bare: 3 compute stages + sink (same chain for both batch and linear)
    let mut batch_bare = PipelineStart::<u64>::new()
        .stage(|x: u64| x.wrapping_mul(3), r)
        .stage(|x: u64| x.wrapping_add(7), r)
        .stage(sink, r)
        .build_batch(1024);

    let mut linear_bare = PipelineStart::<u64>::new()
        .stage(|x: u64| x.wrapping_mul(3), r)
        .stage(|x: u64| x.wrapping_add(7), r)
        .stage(sink, r);

    // Res<T>: 3 world-access stages + sink (same chain for both)
    let mut batch_res = PipelineStart::<u64>::new()
        .stage(add_resource, r)
        .stage(mul_resource, r)
        .stage(sub_resource, r)
        .stage(sink, r)
        .build_batch(1024);

    let mut linear_res = PipelineStart::<u64>::new()
        .stage(add_resource, r)
        .stage(mul_resource, r)
        .stage(sub_resource, r)
        .stage(sink, r);

    // --- Result→catch→map→unwrap_or ---

    let mut catch_pipeline = PipelineStart::<u64>::new()
        .stage(
            |x: u64| -> Result<u64, &'static str> { if x > 0 { Ok(x) } else { Err("zero") } },
            r,
        )
        .catch(|_err: &'static str| {}, r)
        .map(|x: u64| x.wrapping_mul(2), r)
        .unwrap_or(0);

    // --- Handler dispatch setup ---

    let mut sys_res = handler_res_read.into_handler(world.registry_mut());
    let mut sys_res_mut = handler_res_mut_write.into_handler(world.registry_mut());
    let mut sys_two = handler_two_res.into_handler(world.registry_mut());
    let mut sys_dyn: Box<dyn Handler<u64>> =
        Box::new(handler_res_read.into_handler(world.registry_mut()));

    // --- Pipeline benchmarks ---

    print_header("Pipeline Dispatch Latency (cycles)");

    let mut input = 1u64;

    bench_batched("baseline (hand-written fn)", || {
        input = input.wrapping_add(1);
        baseline_handwritten(&mut world, black_box(input))
    });

    bench_batched("bare 3-stage pipe", || {
        input = input.wrapping_add(1);
        bare_3stage_run(&mut bare, &mut world, black_box(input))
    });

    bench_batched("option 3-stage (Some path)", || {
        input = input.wrapping_add(1);
        option_3stage_run(&mut option, &mut world, black_box(input + 1)).unwrap_or(0)
    });

    bench_batched("option 3-stage (None path)", || {
        option_3stage_run(&mut option, &mut world, black_box(0)).unwrap_or(0)
    });

    bench_batched("world-access 2-stage (Res<T>)", || {
        input = input.wrapping_add(1);
        world_access_run(&mut world_resolved, &mut world, black_box(input))
    });

    bench_batched("boxed Pipeline (dyn dispatch)", || {
        input = input.wrapping_add(1);
        boxed_pipeline_run(&mut boxed, &mut world, black_box(input));
        0
    });

    bench_batched("result→catch→map→unwrap_or", || {
        input = input.wrapping_add(1);
        catch_pipeline.run(&mut world, black_box(input))
    });

    // --- Handler dispatch benchmarks ---

    println!();
    print_header("Handler Dispatch Latency (cycles)");

    bench_batched("Handler + Res<u64> (read)", || {
        input = input.wrapping_add(1);
        probe_handler_res_read(&mut sys_res, &mut world, black_box(input));
        0
    });

    bench_batched("Handler + ResMut<u64> (write+stamp)", || {
        input = input.wrapping_add(1);
        probe_handler_res_mut(&mut sys_res_mut, &mut world, black_box(input));
        0
    });

    bench_batched("Handler + 2x Res (tuple fetch)", || {
        input = input.wrapping_add(1);
        probe_handler_two_res(&mut sys_two, &mut world, black_box(input));
        0
    });

    bench_batched("Box<dyn Handler> + Res<u64>", || {
        input = input.wrapping_add(1);
        probe_dyn_handler(&mut *sys_dyn, &mut world, black_box(input));
        0
    });

    // --- Stage pipeline with Res<T> (3-stage) ---

    println!();
    print_header("Stage Pipeline with Res<T> (cycles)");

    bench_batched("3-stage pipeline (Res<T>)", || {
        input = input.wrapping_add(1);
        stage_3.run(&mut world, black_box(input))
    });

    // --- Batch vs Linear throughput (total cycles for 100 items) ---

    println!();
    print_header("Batch vs Linear Throughput (total cycles, 100 items)");

    let items_100: Vec<u64> = (0..100).collect();

    // Batch bare: fill + run
    {
        for _ in 0..WARMUP {
            batch_bare.input_mut().extend_from_slice(&items_100);
            batch_bare.run(&mut world);
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            batch_bare.input_mut().extend_from_slice(&items_100);
            let start = rdtsc_start();
            batch_bare.run(&mut world);
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        samples.sort_unstable();
        println!(
            "{:<44} {:>8} {:>8} {:>8}",
            "batch bare (100 items)",
            percentile(&samples, 50.0),
            percentile(&samples, 99.0),
            percentile(&samples, 99.9),
        );
    }

    // Linear bare: 100 individual calls (same chain)
    {
        for _ in 0..WARMUP {
            for i in 0..100u64 {
                linear_bare.run(&mut world, black_box(i));
            }
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            let start = rdtsc_start();
            for i in 0..100u64 {
                linear_bare.run(&mut world, black_box(i));
            }
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        samples.sort_unstable();
        println!(
            "{:<44} {:>8} {:>8} {:>8}",
            "linear bare (100 calls)",
            percentile(&samples, 50.0),
            percentile(&samples, 99.0),
            percentile(&samples, 99.9),
        );
    }

    // Batch Res<T>: fill + run
    {
        for _ in 0..WARMUP {
            batch_res.input_mut().extend_from_slice(&items_100);
            batch_res.run(&mut world);
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            batch_res.input_mut().extend_from_slice(&items_100);
            let start = rdtsc_start();
            batch_res.run(&mut world);
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        samples.sort_unstable();
        println!(
            "{:<44} {:>8} {:>8} {:>8}",
            "batch Res<T> (100 items)",
            percentile(&samples, 50.0),
            percentile(&samples, 99.0),
            percentile(&samples, 99.9),
        );
    }

    // Linear Res<T>: 100 individual calls (same chain)
    {
        for _ in 0..WARMUP {
            for i in 0..100u64 {
                linear_res.run(&mut world, black_box(i));
            }
        }
        let mut samples = Vec::with_capacity(ITERATIONS);
        for _ in 0..ITERATIONS {
            let start = rdtsc_start();
            for i in 0..100u64 {
                linear_res.run(&mut world, black_box(i));
            }
            let end = rdtsc_end();
            samples.push(end.wrapping_sub(start));
        }
        samples.sort_unstable();
        println!(
            "{:<44} {:>8} {:>8} {:>8}",
            "linear Res<T> (100 calls)",
            percentile(&samples, 50.0),
            percentile(&samples, 99.0),
            percentile(&samples, 99.9),
        );
    }

    // --- inputs_changed cost ---

    println!();
    print_header("inputs_changed Latency (cycles)");

    // Build a world with enough resources for 8-param handlers.
    let mut ic_wb = WorldBuilder::new();
    ic_wb.register::<u64>(0);
    ic_wb.register::<u32>(0);
    ic_wb.register::<bool>(false);
    ic_wb.register::<f64>(0.0);
    ic_wb.register::<i64>(0);
    ic_wb.register::<i32>(0);
    ic_wb.register::<u8>(0);
    ic_wb.register::<u16>(0);
    let mut ic_world = ic_wb.build();
    let ic_r = ic_world.registry_mut();

    let ic1 = ic_1p.into_handler(ic_r);
    let ic2 = ic_2p.into_handler(ic_r);
    let ic4 = ic_4p.into_handler(ic_r);
    let ic8 = ic_8p.into_handler(ic_r);

    // Tick 0: all changed (changed_at == current_sequence).
    bench_batched("inputs_changed 1-param (changed)", || {
        if ic1.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    bench_batched("inputs_changed 2-param (changed)", || {
        if ic2.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    bench_batched("inputs_changed 4-param (changed)", || {
        if ic4.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    bench_batched("inputs_changed 8-param (changed)", || {
        if ic8.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    // Advance tick so inputs are stale.
    ic_world.next_sequence();

    bench_batched("inputs_changed 1-param (stale)", || {
        if ic1.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    bench_batched("inputs_changed 2-param (stale)", || {
        if ic2.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    bench_batched("inputs_changed 4-param (stale)", || {
        if ic4.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    bench_batched("inputs_changed 8-param (stale)", || {
        if ic8.inputs_changed(&ic_world) { 1 } else { 0 }
    });

    println!();
}

fn name(&self) -> &'static str

Returns the handler’s name.

Default returns "<unnamed>". Callback captures the function’s type_name at construction time.

Implementors

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static> Handler<E> for Callback<C, F, (P0,)>

where for<'a> &'a mut F: FnMut(&mut C, P0, E) + FnMut(&mut C, P0::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static, P2: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1, P2)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, P2, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, P2::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static, P2: SystemParam + 'static, P3: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1, P2, P3)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, P2, P3, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, P2::Item<'a>, P3::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static, P2: SystemParam + 'static, P3: SystemParam + 'static, P4: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1, P2, P3, P4)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, P2, P3, P4, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, P2::Item<'a>, P3::Item<'a>, P4::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static, P2: SystemParam + 'static, P3: SystemParam + 'static, P4: SystemParam + 'static, P5: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1, P2, P3, P4, P5)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, P2, P3, P4, P5, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, P2::Item<'a>, P3::Item<'a>, P4::Item<'a>, P5::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static, P2: SystemParam + 'static, P3: SystemParam + 'static, P4: SystemParam + 'static, P5: SystemParam + 'static, P6: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1, P2, P3, P4, P5, P6)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, P2, P3, P4, P5, P6, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, P2::Item<'a>, P3::Item<'a>, P4::Item<'a>, P5::Item<'a>, P6::Item<'a>, E),

impl<C: 'static, E, F: 'static, P0: SystemParam + 'static, P1: SystemParam + 'static, P2: SystemParam + 'static, P3: SystemParam + 'static, P4: SystemParam + 'static, P5: SystemParam + 'static, P6: SystemParam + 'static, P7: SystemParam + 'static> Handler<E> for Callback<C, F, (P0, P1, P2, P3, P4, P5, P6, P7)>

where for<'a> &'a mut F: FnMut(&mut C, P0, P1, P2, P3, P4, P5, P6, P7, E) + FnMut(&mut C, P0::Item<'a>, P1::Item<'a>, P2::Item<'a>, P3::Item<'a>, P4::Item<'a>, P5::Item<'a>, P6::Item<'a>, P7::Item<'a>, E),

impl<C: 'static, E, F: FnMut(&mut C, E) + 'static> Handler<E> for Callback<C, F, ()>

impl<In: 'static, F: FnMut(&mut World, In) + 'static> Handler<In> for Pipeline<In, F>