async-nodes 0.1.0

//! # Quickstart: your first graph (beginner)
//!
//! The smallest useful thing `async-nodes` does: take a few async functions, wire them
//! into a graph, and let the engine schedule them — in parallel where the topology
//! allows — with zero `tokio::spawn`, zero channels, zero join handles.
//!
//! What you'll see:
//! - `#[node]` turns an ordinary async fn into a graph node with **typed, named ports**.
//! - `add_link` wires one node's output to another's input — checked against the
//!   declared ports at validation time, not by convention.
//! - The engine runs the two counters **in the same superstep** (in parallel), and the
//!   report node fires only when *both* arrive (a BSP barrier join).
//!
//! Run it: `cargo run --example quickstart`

use async_nodes::{BspEngine, BspGraph, BspOutputMapExt, EngineResult, node};

/// A node with no inputs is a *source* — it becomes an entry point automatically.
/// The macro generates a `SourceTextNode` struct plus typed pin accessors.
#[node("Source Text")]
async fn source_text() -> String {
    "the quick brown fox jumps over the lazy dog".to_string()
}

/// One input port per function parameter. Types are checked across the wire:
/// wiring a `String` output into an `f64` input fails at validation, not mid-run.
#[node("Word Count")]
async fn word_count(text: String) -> i64 {
    text.split_whitespace().count() as i64
}

#[node("Char Count")]
async fn char_count(text: String) -> i64 {
    text.chars().count() as i64
}

/// Two inputs = a join. This node is only *ready* once both upstream nodes completed —
/// that's the barrier: no polling, no waiting code, the topology is the synchronization.
#[node("Report")]
async fn report(words: i64, chars: i64) -> String {
    format!("{words} words, {chars} characters")
}

#[tokio::main]
async fn main() {
    let mut graph = BspGraph::new();

    // Register the macro-generated node structs. You get back typed handles whose
    // accessor methods (`.text()`, `.words()`, …) are named after the fn parameters —
    // no magic port numbers to miswire.
    let source = graph.add_node(SourceTextNode::new());
    let words = graph.add_node(WordCountNode::new());
    let chars = graph.add_node(CharCountNode::new());
    let report = graph.add_node(ReportNode::new());

    // Wire the pipeline. Both counters hang off the same source output: that's a
    // fan-out, and the engine runs them concurrently in one superstep.
    graph.add_link(source.output(), words.text());
    graph.add_link(source.output(), chars.text());
    graph.add_link(words.output(), report.words());
    graph.add_link(chars.output(), report.chars());

    // Every graph runs over a caller-owned state value; a pure dataflow graph like
    // this one doesn't need any, so we pass `()`.
    let engine = BspEngine::new(graph);

    match engine.run(()).await {
        Ok(EngineResult::Complete(state)) => {
            // Read typed values straight back out of the output pins.
            let text: String = state.get_typed(report.output()).unwrap();
            println!("✅ {text}");

            // The engine counted supersteps for you: source, then the two counters
            // together, then the report — three supersteps, barrier-synchronized.
            println!("completed in {} supersteps", state.step);
        }
        Ok(EngineResult::Suspended(_)) => println!("graph asked for external input"),
        Ok(EngineResult::StepLimitReached(_)) => println!("hit the step budget"),
        Err(e) => eprintln!("graph failed: {e}"),
    }
}