Struct StreamNode 

pub struct StreamNode {
    pub id: String,
    pub processor: Arc<dyn StreamProcessor>,
    pub backpressure_config: BackpressureConfig,
    pub chunk_config: Option<ChunkConfig>,
    pub collect_results: bool,
}

Stream node for processing data streams

Fields

id: String

processor: Arc<dyn StreamProcessor>

backpressure_config: BackpressureConfig

chunk_config: Option<ChunkConfig>

collect_results: bool

Implementations

impl StreamNode

pub fn new(id: impl Into<String>, processor: Arc<dyn StreamProcessor>) -> Self

Create a new stream node

Examples found in repository

examples/streaming_flow.rs (line 49)

async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init();

    println!("🚀 Streaming Processing Example\n");

    // Example 1: Basic streaming with backpressure
    println!("=== Example 1: Basic Streaming with Backpressure ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("processor", processor)
        .with_backpressure(BackpressureConfig {
            buffer_size: 10,
            max_concurrent: 5,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 100 numbers with backpressure...");
    let start = std::time::Instant::now();
    let result = node.process_stream(data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  First: {:?}, Last: {:?}\n", results.first(), results.last());
    }

    // Example 2: Large dataset with chunking
    println!("=== Example 2: Large Dataset with Chunking ===\n");

    let large_data: Vec<Value> = (1..=10_000).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("chunked_processor", processor)
        .with_chunking(ChunkConfig {
            chunk_size: 1000,
            overlap: 0,
        })
        .with_backpressure(BackpressureConfig {
            buffer_size: 100,
            max_concurrent: 10,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 10,000 numbers in chunks of 1,000...");
    let start = std::time::Instant::now();
    let result = node.process_stream(large_data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  Throughput: {:.0} items/sec\n", results.len() as f64 / duration.as_secs_f64());
    }

    // Example 3: Map operator
    println!("=== Example 3: Map Operator ===\n");

    let data: Vec<Value> = (1..=10).map(|i| Value::Number(i.into())).collect();

    let map_op = Arc::new(MapOperator::new("square", |v: Value| {
        if let Some(n) = v.as_i64() {
            Value::Number((n * n).into())
        } else {
            v
        }
    }));

    let node = StreamNode::new("map_node", map_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Squaring numbers 1-10...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Results: {:?}\n", results);
    }

    // Example 4: Filter operator
    println!("=== Example 4: Filter Operator ===\n");

    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even_only", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let node = StreamNode::new("filter_node", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Filtering even numbers from 1-20...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Even numbers: {:?}\n", results);
    }

    // Example 5: Fold operator (sum)
    println!("=== Example 5: Fold Operator (Sum) ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let fold_op = Arc::new(FoldOperator::new("sum", 0i64, |acc: i64, v: Value| {
        acc + v.as_i64().unwrap_or(0)
    }));

    let node = StreamNode::new("fold_node", fold_op)
        .with_chunking(ChunkConfig {
            chunk_size: 100,
            overlap: 0,
        })
        .collect_results(false); // Only return final result

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Summing numbers 1-100...");
    let result = node.process_stream(data, &ctx).await?;

    println!("✓ Sum: {:?}\n", result);

    // Example 6: Chained operations
    println!("=== Example 6: Chained Operations ===\n");

    println!("Pipeline: numbers → filter(even) → map(square) → collect");

    // First: filter even numbers
    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let filter_node = StreamNode::new("filter", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    let filtered = filter_node.process_stream(data, &ctx).await?;

    // Then: square the numbers
    if let Value::Array(filtered_data) = filtered {
        let map_op = Arc::new(MapOperator::new("square", |v: Value| {
            if let Some(n) = v.as_i64() {
                Value::Number((n * n).into())
            } else {
                v
            }
        }));

        let map_node = StreamNode::new("square", map_op);
        let result = map_node.process_stream(filtered_data, &ctx).await?;

        println!("✓ Results: {:?}\n", result);
    }

    println!("=== Benefits of Streaming Processing ===");
    println!("  • Memory efficient - processes items incrementally");
    println!("  • Backpressure handling - prevents overwhelming consumers");
    println!("  • Large dataset support - chunking for massive data");
    println!("  • Concurrent processing - multiple items in parallel");
    println!("  • Composable operators - chain transformations");
    println!("\n🎉 Example completed!");

    Ok(())
}

pub fn with_backpressure(self, config: BackpressureConfig) -> Self

Configure backpressure

Examples found in repository

examples/streaming_flow.rs (lines 50-53)

async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init();

    println!("🚀 Streaming Processing Example\n");

    // Example 1: Basic streaming with backpressure
    println!("=== Example 1: Basic Streaming with Backpressure ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("processor", processor)
        .with_backpressure(BackpressureConfig {
            buffer_size: 10,
            max_concurrent: 5,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 100 numbers with backpressure...");
    let start = std::time::Instant::now();
    let result = node.process_stream(data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  First: {:?}, Last: {:?}\n", results.first(), results.last());
    }

    // Example 2: Large dataset with chunking
    println!("=== Example 2: Large Dataset with Chunking ===\n");

    let large_data: Vec<Value> = (1..=10_000).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("chunked_processor", processor)
        .with_chunking(ChunkConfig {
            chunk_size: 1000,
            overlap: 0,
        })
        .with_backpressure(BackpressureConfig {
            buffer_size: 100,
            max_concurrent: 10,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 10,000 numbers in chunks of 1,000...");
    let start = std::time::Instant::now();
    let result = node.process_stream(large_data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  Throughput: {:.0} items/sec\n", results.len() as f64 / duration.as_secs_f64());
    }

    // Example 3: Map operator
    println!("=== Example 3: Map Operator ===\n");

    let data: Vec<Value> = (1..=10).map(|i| Value::Number(i.into())).collect();

    let map_op = Arc::new(MapOperator::new("square", |v: Value| {
        if let Some(n) = v.as_i64() {
            Value::Number((n * n).into())
        } else {
            v
        }
    }));

    let node = StreamNode::new("map_node", map_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Squaring numbers 1-10...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Results: {:?}\n", results);
    }

    // Example 4: Filter operator
    println!("=== Example 4: Filter Operator ===\n");

    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even_only", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let node = StreamNode::new("filter_node", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Filtering even numbers from 1-20...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Even numbers: {:?}\n", results);
    }

    // Example 5: Fold operator (sum)
    println!("=== Example 5: Fold Operator (Sum) ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let fold_op = Arc::new(FoldOperator::new("sum", 0i64, |acc: i64, v: Value| {
        acc + v.as_i64().unwrap_or(0)
    }));

    let node = StreamNode::new("fold_node", fold_op)
        .with_chunking(ChunkConfig {
            chunk_size: 100,
            overlap: 0,
        })
        .collect_results(false); // Only return final result

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Summing numbers 1-100...");
    let result = node.process_stream(data, &ctx).await?;

    println!("✓ Sum: {:?}\n", result);

    // Example 6: Chained operations
    println!("=== Example 6: Chained Operations ===\n");

    println!("Pipeline: numbers → filter(even) → map(square) → collect");

    // First: filter even numbers
    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let filter_node = StreamNode::new("filter", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    let filtered = filter_node.process_stream(data, &ctx).await?;

    // Then: square the numbers
    if let Value::Array(filtered_data) = filtered {
        let map_op = Arc::new(MapOperator::new("square", |v: Value| {
            if let Some(n) = v.as_i64() {
                Value::Number((n * n).into())
            } else {
                v
            }
        }));

        let map_node = StreamNode::new("square", map_op);
        let result = map_node.process_stream(filtered_data, &ctx).await?;

        println!("✓ Results: {:?}\n", result);
    }

    println!("=== Benefits of Streaming Processing ===");
    println!("  • Memory efficient - processes items incrementally");
    println!("  • Backpressure handling - prevents overwhelming consumers");
    println!("  • Large dataset support - chunking for massive data");
    println!("  • Concurrent processing - multiple items in parallel");
    println!("  • Composable operators - chain transformations");
    println!("\n🎉 Example completed!");

    Ok(())
}

pub fn with_chunking(self, config: ChunkConfig) -> Self

Enable chunked processing for large datasets

Examples found in repository

examples/streaming_flow.rs (lines 76-79)

async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init();

    println!("🚀 Streaming Processing Example\n");

    // Example 1: Basic streaming with backpressure
    println!("=== Example 1: Basic Streaming with Backpressure ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("processor", processor)
        .with_backpressure(BackpressureConfig {
            buffer_size: 10,
            max_concurrent: 5,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 100 numbers with backpressure...");
    let start = std::time::Instant::now();
    let result = node.process_stream(data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  First: {:?}, Last: {:?}\n", results.first(), results.last());
    }

    // Example 2: Large dataset with chunking
    println!("=== Example 2: Large Dataset with Chunking ===\n");

    let large_data: Vec<Value> = (1..=10_000).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("chunked_processor", processor)
        .with_chunking(ChunkConfig {
            chunk_size: 1000,
            overlap: 0,
        })
        .with_backpressure(BackpressureConfig {
            buffer_size: 100,
            max_concurrent: 10,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 10,000 numbers in chunks of 1,000...");
    let start = std::time::Instant::now();
    let result = node.process_stream(large_data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  Throughput: {:.0} items/sec\n", results.len() as f64 / duration.as_secs_f64());
    }

    // Example 3: Map operator
    println!("=== Example 3: Map Operator ===\n");

    let data: Vec<Value> = (1..=10).map(|i| Value::Number(i.into())).collect();

    let map_op = Arc::new(MapOperator::new("square", |v: Value| {
        if let Some(n) = v.as_i64() {
            Value::Number((n * n).into())
        } else {
            v
        }
    }));

    let node = StreamNode::new("map_node", map_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Squaring numbers 1-10...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Results: {:?}\n", results);
    }

    // Example 4: Filter operator
    println!("=== Example 4: Filter Operator ===\n");

    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even_only", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let node = StreamNode::new("filter_node", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Filtering even numbers from 1-20...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Even numbers: {:?}\n", results);
    }

    // Example 5: Fold operator (sum)
    println!("=== Example 5: Fold Operator (Sum) ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let fold_op = Arc::new(FoldOperator::new("sum", 0i64, |acc: i64, v: Value| {
        acc + v.as_i64().unwrap_or(0)
    }));

    let node = StreamNode::new("fold_node", fold_op)
        .with_chunking(ChunkConfig {
            chunk_size: 100,
            overlap: 0,
        })
        .collect_results(false); // Only return final result

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Summing numbers 1-100...");
    let result = node.process_stream(data, &ctx).await?;

    println!("✓ Sum: {:?}\n", result);

    // Example 6: Chained operations
    println!("=== Example 6: Chained Operations ===\n");

    println!("Pipeline: numbers → filter(even) → map(square) → collect");

    // First: filter even numbers
    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let filter_node = StreamNode::new("filter", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    let filtered = filter_node.process_stream(data, &ctx).await?;

    // Then: square the numbers
    if let Value::Array(filtered_data) = filtered {
        let map_op = Arc::new(MapOperator::new("square", |v: Value| {
            if let Some(n) = v.as_i64() {
                Value::Number((n * n).into())
            } else {
                v
            }
        }));

        let map_node = StreamNode::new("square", map_op);
        let result = map_node.process_stream(filtered_data, &ctx).await?;

        println!("✓ Results: {:?}\n", result);
    }

    println!("=== Benefits of Streaming Processing ===");
    println!("  • Memory efficient - processes items incrementally");
    println!("  • Backpressure handling - prevents overwhelming consumers");
    println!("  • Large dataset support - chunking for massive data");
    println!("  • Concurrent processing - multiple items in parallel");
    println!("  • Composable operators - chain transformations");
    println!("\n🎉 Example completed!");

    Ok(())
}

pub fn collect_results(self, collect: bool) -> Self

Set whether to collect all results (default: true) If false, only the last result is stored

Examples found in repository

examples/streaming_flow.rs (line 159)

async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init();

    println!("🚀 Streaming Processing Example\n");

    // Example 1: Basic streaming with backpressure
    println!("=== Example 1: Basic Streaming with Backpressure ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("processor", processor)
        .with_backpressure(BackpressureConfig {
            buffer_size: 10,
            max_concurrent: 5,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 100 numbers with backpressure...");
    let start = std::time::Instant::now();
    let result = node.process_stream(data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  First: {:?}, Last: {:?}\n", results.first(), results.last());
    }

    // Example 2: Large dataset with chunking
    println!("=== Example 2: Large Dataset with Chunking ===\n");

    let large_data: Vec<Value> = (1..=10_000).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("chunked_processor", processor)
        .with_chunking(ChunkConfig {
            chunk_size: 1000,
            overlap: 0,
        })
        .with_backpressure(BackpressureConfig {
            buffer_size: 100,
            max_concurrent: 10,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 10,000 numbers in chunks of 1,000...");
    let start = std::time::Instant::now();
    let result = node.process_stream(large_data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  Throughput: {:.0} items/sec\n", results.len() as f64 / duration.as_secs_f64());
    }

    // Example 3: Map operator
    println!("=== Example 3: Map Operator ===\n");

    let data: Vec<Value> = (1..=10).map(|i| Value::Number(i.into())).collect();

    let map_op = Arc::new(MapOperator::new("square", |v: Value| {
        if let Some(n) = v.as_i64() {
            Value::Number((n * n).into())
        } else {
            v
        }
    }));

    let node = StreamNode::new("map_node", map_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Squaring numbers 1-10...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Results: {:?}\n", results);
    }

    // Example 4: Filter operator
    println!("=== Example 4: Filter Operator ===\n");

    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even_only", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let node = StreamNode::new("filter_node", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Filtering even numbers from 1-20...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Even numbers: {:?}\n", results);
    }

    // Example 5: Fold operator (sum)
    println!("=== Example 5: Fold Operator (Sum) ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let fold_op = Arc::new(FoldOperator::new("sum", 0i64, |acc: i64, v: Value| {
        acc + v.as_i64().unwrap_or(0)
    }));

    let node = StreamNode::new("fold_node", fold_op)
        .with_chunking(ChunkConfig {
            chunk_size: 100,
            overlap: 0,
        })
        .collect_results(false); // Only return final result

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Summing numbers 1-100...");
    let result = node.process_stream(data, &ctx).await?;

    println!("✓ Sum: {:?}\n", result);

    // Example 6: Chained operations
    println!("=== Example 6: Chained Operations ===\n");

    println!("Pipeline: numbers → filter(even) → map(square) → collect");

    // First: filter even numbers
    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let filter_node = StreamNode::new("filter", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    let filtered = filter_node.process_stream(data, &ctx).await?;

    // Then: square the numbers
    if let Value::Array(filtered_data) = filtered {
        let map_op = Arc::new(MapOperator::new("square", |v: Value| {
            if let Some(n) = v.as_i64() {
                Value::Number((n * n).into())
            } else {
                v
            }
        }));

        let map_node = StreamNode::new("square", map_op);
        let result = map_node.process_stream(filtered_data, &ctx).await?;

        println!("✓ Results: {:?}\n", result);
    }

    println!("=== Benefits of Streaming Processing ===");
    println!("  • Memory efficient - processes items incrementally");
    println!("  • Backpressure handling - prevents overwhelming consumers");
    println!("  • Large dataset support - chunking for massive data");
    println!("  • Concurrent processing - multiple items in parallel");
    println!("  • Composable operators - chain transformations");
    println!("\n🎉 Example completed!");

    Ok(())
}

pub async fn process_stream( &self, data: Vec<Value>, ctx: &Context, ) -> RuleResult

Process a stream of data

Examples found in repository

examples/streaming_flow.rs (line 61)

async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init();

    println!("🚀 Streaming Processing Example\n");

    // Example 1: Basic streaming with backpressure
    println!("=== Example 1: Basic Streaming with Backpressure ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("processor", processor)
        .with_backpressure(BackpressureConfig {
            buffer_size: 10,
            max_concurrent: 5,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 100 numbers with backpressure...");
    let start = std::time::Instant::now();
    let result = node.process_stream(data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  First: {:?}, Last: {:?}\n", results.first(), results.last());
    }

    // Example 2: Large dataset with chunking
    println!("=== Example 2: Large Dataset with Chunking ===\n");

    let large_data: Vec<Value> = (1..=10_000).map(|i| Value::Number(i.into())).collect();

    let processor = Arc::new(NumberProcessor);
    let node = StreamNode::new("chunked_processor", processor)
        .with_chunking(ChunkConfig {
            chunk_size: 1000,
            overlap: 0,
        })
        .with_backpressure(BackpressureConfig {
            buffer_size: 100,
            max_concurrent: 10,
        });

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Processing 10,000 numbers in chunks of 1,000...");
    let start = std::time::Instant::now();
    let result = node.process_stream(large_data, &ctx).await?;
    let duration = start.elapsed();

    if let Value::Array(results) = result {
        println!("✓ Processed {} items in {:?}", results.len(), duration);
        println!("  Throughput: {:.0} items/sec\n", results.len() as f64 / duration.as_secs_f64());
    }

    // Example 3: Map operator
    println!("=== Example 3: Map Operator ===\n");

    let data: Vec<Value> = (1..=10).map(|i| Value::Number(i.into())).collect();

    let map_op = Arc::new(MapOperator::new("square", |v: Value| {
        if let Some(n) = v.as_i64() {
            Value::Number((n * n).into())
        } else {
            v
        }
    }));

    let node = StreamNode::new("map_node", map_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Squaring numbers 1-10...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Results: {:?}\n", results);
    }

    // Example 4: Filter operator
    println!("=== Example 4: Filter Operator ===\n");

    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even_only", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let node = StreamNode::new("filter_node", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Filtering even numbers from 1-20...");
    let result = node.process_stream(data, &ctx).await?;

    if let Value::Array(results) = result {
        println!("✓ Even numbers: {:?}\n", results);
    }

    // Example 5: Fold operator (sum)
    println!("=== Example 5: Fold Operator (Sum) ===\n");

    let data: Vec<Value> = (1..=100).map(|i| Value::Number(i.into())).collect();

    let fold_op = Arc::new(FoldOperator::new("sum", 0i64, |acc: i64, v: Value| {
        acc + v.as_i64().unwrap_or(0)
    }));

    let node = StreamNode::new("fold_node", fold_op)
        .with_chunking(ChunkConfig {
            chunk_size: 100,
            overlap: 0,
        })
        .collect_results(false); // Only return final result

    let ctx = Context {
        data: HashMap::new(),
    };

    println!("Summing numbers 1-100...");
    let result = node.process_stream(data, &ctx).await?;

    println!("✓ Sum: {:?}\n", result);

    // Example 6: Chained operations
    println!("=== Example 6: Chained Operations ===\n");

    println!("Pipeline: numbers → filter(even) → map(square) → collect");

    // First: filter even numbers
    let data: Vec<Value> = (1..=20).map(|i| Value::Number(i.into())).collect();

    let filter_op = Arc::new(FilterOperator::new("even", |v: &Value| {
        v.as_i64().map(|n| n % 2 == 0).unwrap_or(false)
    }));

    let filter_node = StreamNode::new("filter", filter_op);
    let ctx = Context {
        data: HashMap::new(),
    };

    let filtered = filter_node.process_stream(data, &ctx).await?;

    // Then: square the numbers
    if let Value::Array(filtered_data) = filtered {
        let map_op = Arc::new(MapOperator::new("square", |v: Value| {
            if let Some(n) = v.as_i64() {
                Value::Number((n * n).into())
            } else {
                v
            }
        }));

        let map_node = StreamNode::new("square", map_op);
        let result = map_node.process_stream(filtered_data, &ctx).await?;

        println!("✓ Results: {:?}\n", result);
    }

    println!("=== Benefits of Streaming Processing ===");
    println!("  • Memory efficient - processes items incrementally");
    println!("  • Backpressure handling - prevents overwhelming consumers");
    println!("  • Large dataset support - chunking for massive data");
    println!("  • Concurrent processing - multiple items in parallel");
    println!("  • Composable operators - chain transformations");
    println!("\n🎉 Example completed!");

    Ok(())
}

Trait Implementations

impl Clone for StreamNode

fn clone(&self) -> StreamNode

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for StreamNode

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

Formats the value using the given formatter.

impl Node for StreamNode

fn id(&self) -> &str

fn node_type(&self) -> NodeType

fn run<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 mut Context, ) -> Pin<Box<dyn Future<Output = RuleResult> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,