Struct FanInTracker 

pub struct FanInTracker { /* private fields */ }

Tracks which upstream paths have completed for a deferred node.

The tracker maintains a set of expected source nodes and records their outputs as they arrive. Once all expected sources have reported, the tracker is ready and outputs can be merged.

Example

use adk_graph::deferred::{FanInTracker, MergeStrategy};
use serde_json::json;

let mut tracker = FanInTracker::new(vec!["node_a", "node_b"]);

assert!(!tracker.is_ready());
assert_eq!(tracker.received_count(), 0);
assert_eq!(tracker.expected_count(), 2);

tracker.record("node_a", json!("output_a"));
assert!(!tracker.is_ready());

tracker.record("node_b", json!("output_b"));
assert!(tracker.is_ready());

let merged = tracker.merge(&MergeStrategy::Collect);
assert_eq!(merged, json!(["output_a", "output_b"]));

Implementations

impl FanInTracker

pub fn new(expected_sources: Vec<&str>) -> Self

Create a new tracker expecting outputs from the given source nodes.

Arguments

• expected_sources - Names of upstream nodes that must complete before this deferred node can execute.

Example

use adk_graph::deferred::FanInTracker;

let tracker = FanInTracker::new(vec!["branch_1", "branch_2", "branch_3"]);
assert_eq!(tracker.expected_count(), 3);
assert!(!tracker.is_ready());

pub fn is_ready(&self) -> bool

Returns true when all expected sources have reported their output.

Example

use adk_graph::deferred::FanInTracker;
use serde_json::json;

let mut tracker = FanInTracker::new(vec!["a"]);
assert!(!tracker.is_ready());

tracker.record("a", json!(42));
assert!(tracker.is_ready());

pub fn record(&mut self, source_node: &str, output: Value)

Record the output from a source node.

If the source has already been recorded, the previous value is overwritten (last-write-wins). Recording a source that is not in the expected set is a no-op for readiness but the value is still stored.

Arguments

• source_node - The name of the upstream node that produced the output.
• output - The output value from the source node.

Example

use adk_graph::deferred::FanInTracker;
use serde_json::json;

let mut tracker = FanInTracker::new(vec!["worker_1", "worker_2"]);
tracker.record("worker_1", json!({"status": "done"}));
assert_eq!(tracker.received_count(), 1);

pub fn merge(&self, strategy: &MergeStrategy) -> Value

Merge all received outputs according to the given strategy.

The merge operation combines all recorded outputs into a single Value based on the MergeStrategy:

• MergeStrategy::Collect: Returns a JSON array of all outputs in insertion order.
• MergeStrategy::MergeMap: Merges all JSON object outputs into a single object (last-write-wins). Non-object outputs are skipped.
• MergeStrategy::First: Returns the first recorded output.
• MergeStrategy::Custom: Invokes the custom function with all outputs.

Arguments

• strategy - The merge strategy to apply.

Example

use adk_graph::deferred::{FanInTracker, MergeStrategy};
use serde_json::json;

let mut tracker = FanInTracker::new(vec!["a", "b"]);
tracker.record("a", json!({"x": 1}));
tracker.record("b", json!({"y": 2}));

// Collect strategy
let result = tracker.merge(&MergeStrategy::Collect);
assert_eq!(result, json!([{"x": 1}, {"y": 2}]));

// MergeMap strategy
let result = tracker.merge(&MergeStrategy::MergeMap);
assert_eq!(result, json!({"x": 1, "y": 2}));

pub fn received_count(&self) -> usize

Returns the number of outputs received so far.

pub fn expected_count(&self) -> usize

Returns the number of expected source nodes.

pub fn pending_sources(&self) -> Vec<&str>

Returns the names of sources that have not yet reported.

pub fn completed_sources(&self) -> Vec<&str>

Returns the names of sources that have reported.

Trait Implementations

impl Debug for FanInTracker

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.