Struct Message

pub struct Message {
    pub id: String,
    pub data: Value,
    pub payload: Value,
    pub metadata: Value,
    pub temp_data: Value,
    pub audit_trail: Vec<AuditTrail>,
    pub errors: Vec<ErrorInfo>,
}

Fields

id: String

data: Value

payload: Value

metadata: Value

temp_data: Value

audit_trail: Vec<AuditTrail>

errors: Vec<ErrorInfo>

Errors that occurred during message processing

Implementations

impl Message

pub fn new(payload: &Value) -> Self

Examples found in repository

examples/benchmark.rs (line 183)

async fn run_async_benchmark(
    engine: &Engine,
    sample_user_data: &Value,
    num_iterations: usize,
) -> Result<BenchmarkResults, Box<dyn std::error::Error>> {
    let mut total_duration = Duration::new(0, 0);
    let mut min_duration = Duration::new(u64::MAX, 0);
    let mut max_duration = Duration::new(0, 0);
    let mut all_durations = Vec::with_capacity(num_iterations);
    let mut error_count = 0;

    println!("Starting async benchmark with {num_iterations} iterations...");

    for i in 0..num_iterations {
        let mut message = Message::new(&json!({}));
        message.temp_data = sample_user_data.clone();
        message.data = json!({});
        message.metadata = json!({
            "timestamp": chrono::Utc::now().to_rfc3339(),
            "iteration": i
        });

        let start = Instant::now();
        match engine.process_message(&mut message).await {
            Ok(_) => {
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);

                // Check for processing errors
                if message.has_errors() {
                    error_count += 1;
                    if error_count <= 5 {
                        // Only print first 5 errors
                        println!("Processing errors in iteration {}: {:?}", i, message.errors);
                    }
                }
            }
            Err(e) => {
                error_count += 1;
                if error_count <= 5 {
                    println!("Error in iteration {i}: {e:?}");
                }
                // Still record the time even for errors
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);
            }
        }

        if (i + 1) % 1000 == 0 {
            println!("Completed {} async iterations", i + 1);
        }
    }

    if error_count > 0 {
        println!("Total errors encountered: {error_count}");
    }

    // Sort durations for percentile calculations
    all_durations.sort();

    let p95_idx = (num_iterations as f64 * 0.95) as usize;
    let p99_idx = (num_iterations as f64 * 0.99) as usize;
    let p95 = all_durations.get(p95_idx).unwrap_or(&Duration::ZERO);
    let p99 = all_durations.get(p99_idx).unwrap_or(&Duration::ZERO);
    let avg_duration = total_duration / num_iterations as u32;

    println!("\nAsync Benchmark Results (v{VERSION}):");
    println!("  Iterations: {num_iterations}");
    println!("  Errors: {error_count}");
    println!("  Min time: {min_duration:?}");
    println!("  Max time: {max_duration:?}");
    println!("  Avg time: {avg_duration:?}");
    println!("  95th percentile: {p95:?}");
    println!("  99th percentile: {p99:?}");
    println!("  Total time: {total_duration:?}");

    Ok(BenchmarkResults {
        iterations: num_iterations,
        min_time: min_duration,
        max_time: max_duration,
        avg_time: avg_duration,
        p95: *p95,
        p99: *p99,
        total_time: total_duration,
    })
}

async fn run_sync_benchmark(
    engine: &Engine,
    sample_user_data: &Value,
    num_iterations: usize,
) -> Result<BenchmarkResults, Box<dyn std::error::Error>> {
    let mut total_duration = Duration::new(0, 0);
    let mut min_duration = Duration::new(u64::MAX, 0);
    let mut max_duration = Duration::new(0, 0);
    let mut all_durations = Vec::with_capacity(num_iterations);
    let mut error_count = 0;

    println!("Starting sync-style benchmark with {num_iterations} iterations...");

    for i in 0..num_iterations {
        let mut message = Message::new(&json!({}));
        message.temp_data = sample_user_data.clone();
        message.data = json!({});
        message.metadata = json!({
            "timestamp": chrono::Utc::now().to_rfc3339(),
            "iteration": i
        });

        let start = Instant::now();
        // Use tokio::task::block_in_place to simulate sync behavior
        let result = tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(engine.process_message(&mut message))
        });

        match result {
            Ok(_) => {
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);

                if message.has_errors() {
                    error_count += 1;
                }
            }
            Err(e) => {
                error_count += 1;
                if error_count <= 5 {
                    println!("Sync error in iteration {i}: {e:?}");
                }
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);
            }
        }

        if (i + 1) % 1000 == 0 {
            println!("Completed {} sync iterations", i + 1);
        }
    }

    if error_count > 0 {
        println!("Total sync errors encountered: {error_count}");
    }

    all_durations.sort();

    let p95_idx = (num_iterations as f64 * 0.95) as usize;
    let p99_idx = (num_iterations as f64 * 0.99) as usize;
    let p95 = all_durations.get(p95_idx).unwrap_or(&Duration::ZERO);
    let p99 = all_durations.get(p99_idx).unwrap_or(&Duration::ZERO);
    let avg_duration = total_duration / num_iterations as u32;

    println!("\nSync Benchmark Results (v{VERSION}):");
    println!("  Iterations: {num_iterations}");
    println!("  Errors: {error_count}");
    println!("  Min time: {min_duration:?}");
    println!("  Max time: {max_duration:?}");
    println!("  Avg time: {avg_duration:?}");
    println!("  95th percentile: {p95:?}");
    println!("  99th percentile: {p99:?}");
    println!("  Total time: {total_duration:?}");

    Ok(BenchmarkResults {
        iterations: num_iterations,
        min_time: min_duration,
        max_time: max_duration,
        avg_time: avg_duration,
        p95: *p95,
        p99: *p99,
        total_time: total_duration,
    })
}

examples/complete_workflow.rs (line 136)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create the workflow engine (built-in functions are auto-registered)
    let mut engine = Engine::new();

    // Define a workflow that:
    // 1. Fetches data from a public API
    // 2. Enriches the message with transformed data
    // 3. Validates the enriched data
    let workflow_json = r#"
    {
        "id": "complete_workflow",
        "name": "Complete Workflow Example",
        "priority": 0,
        "description": "Demonstrates fetch -> enrich -> validate flow",
        "condition": { "==": [true, true] },
        "tasks": [
            {
                "id": "fetch_user_data",
                "name": "Fetch User Data",
                "description": "Get user data from a public API",
                "function": {
                    "name": "http",
                    "input": {
                        "url": "https://jsonplaceholder.typicode.com/users/1",
                        "method": "GET",
                        "headers": {
                            "Accept": "application/json"
                        }
                    }
                }
            },
            {
                "id": "initialize_user",
                "name": "Initialize User Structure",
                "description": "Create empty user object in data",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data",
                                "logic": { "preserve": {"user": {}} }
                            }
                        ]
                    }
                }
            },
            {
                "id": "transform_data",
                "name": "Transform Data",
                "description": "Map API response to our data model",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.user.id", 
                                "logic": { "var": "temp_data.body.id" }
                            },
                            {
                                "path": "data.user.name", 
                                "logic": { "var": "temp_data.body.name" }
                            },
                            {
                                "path": "data.user.email", 
                                "logic": { "var": "temp_data.body.email" }
                            },
                            {
                                "path": "data.user.address", 
                                "logic": {
                                    "cat": [
                                        { "var": "temp_data.body.address.street" },
                                        ", ",
                                        { "var": "temp_data.body.address.city" }
                                    ]
                                }
                            },
                            {
                                "path": "data.user.company", 
                                "logic": { "var": "temp_data.body.company.name" }
                            }
                        ]
                    }
                }
            },
            {
                "id": "validate_user_data",
                "name": "Validate User Data",
                "description": "Ensure the user data meets our requirements",
                "function": {
                    "name": "validate",
                    "input": {
                        "rules": [
                            {
                                "path": "data",
                                "logic": { "!!": { "var": "data.user.id" } },
                                "message": "User ID is required"
                            },
                            {
                                "path": "data",
                                "logic": { "!!": { "var": "data.user.name" } },
                                "message": "User name is required"
                            },
                            {
                                "path": "data",
                                "logic": { "!!": { "var": "data.user.email" } },
                                "message": "User email is required"
                            },
                            {
                                "path": "data",
                                "logic": {
                                    "in": [
                                        "@",
                                        { "var": "data.user.email" }
                                    ]
                                },
                                "message": "Email must be valid format"
                            }
                        ]
                    }
                }
            }
        ]
    }
    "#;

    // Parse and add the workflow to the engine
    let workflow = Workflow::from_json(workflow_json)?;
    engine.add_workflow(&workflow);

    // Create a message to process with properly initialized data structure
    let mut message = Message::new(&json!({}));

    // Process the message through the workflow asynchronously
    println!("Processing message through workflow...");

    match engine.process_message(&mut message).await {
        Ok(_) => {
            println!("Workflow completed successfully!");
        }
        Err(e) => {
            eprintln!("Error executing workflow: {e:?}");
            if !message.errors.is_empty() {
                println!("\nErrors recorded in message:");
                for err in &message.errors {
                    println!(
                        "- Workflow: {:?}, Task: {:?}, Error: {:?}",
                        err.workflow_id, err.task_id, err.error_message
                    );
                }
            }
        }
    }

    println!(
        "\nFull message structure:\n{}",
        serde_json::to_string_pretty(&message)?
    );

    Ok(())
}

examples/custom_function.rs (line 399)

async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    println!("=== Custom Function Example ===\n");

    // Create engine without built-in functions to demonstrate custom ones
    let mut engine = Engine::new_empty();

    // Register our custom functions
    engine.register_task_function(
        "statistics".to_string(),
        Box::new(StatisticsFunction::new()),
    );

    engine.register_task_function(
        "enrich_data".to_string(),
        Box::new(DataEnrichmentFunction::new()),
    );

    // Also register built-in map function for data preparation
    engine.register_task_function(
        "map".to_string(),
        Box::new(dataflow_rs::engine::functions::MapFunction::new()),
    );

    // Define a workflow that uses our custom functions
    let workflow_json = r#"
    {
        "id": "custom_function_demo",
        "name": "Custom Function Demo",
        "description": "Demonstrates custom async functions in workflow",
        "condition": { "==": [true, true] },
        "tasks": [
            {
                "id": "prepare_data",
                "name": "Prepare Data",
                "description": "Extract and prepare data for analysis",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.numbers",
                                "logic": { "var": "temp_data.measurements" }
                            },
                            {
                                "path": "data.user_id",
                                "logic": { "var": "temp_data.user_id" }
                            }
                        ]
                    }
                }
            },
            {
                "id": "calculate_stats",
                "name": "Calculate Statistics",
                "description": "Calculate statistical measures from numeric data",
                "function": {
                    "name": "statistics",
                    "input": {
                        "data_path": "data.numbers",
                        "output_path": "data.stats"
                    }
                }
            },
            {
                "id": "enrich_user_data",
                "name": "Enrich User Data",
                "description": "Add additional user information",
                "function": {
                    "name": "enrich_data",
                    "input": {
                        "lookup_field": "user_id",
                        "lookup_value": "user_123",
                        "output_path": "data.user_info"
                    }
                }
            }
        ]
    }
    "#;

    // Parse and add the workflow
    let workflow = Workflow::from_json(workflow_json)?;
    engine.add_workflow(&workflow);

    // Create sample data
    let sample_data = json!({
        "measurements": [10.5, 15.2, 8.7, 22.1, 18.9, 12.3, 25.6, 14.8, 19.4, 16.7],
        "user_id": "user_123",
        "timestamp": "2024-01-15T10:30:00Z"
    });

    // Create and process message
    let mut message = dataflow_rs::engine::message::Message::new(&json!({}));
    message.temp_data = sample_data;
    message.data = json!({});

    println!("Processing message with custom functions...\n");

    // Process the message through our custom workflow
    match engine.process_message(&mut message).await {
        Ok(_) => {
            println!("✅ Message processed successfully!\n");

            println!("📊 Final Results:");
            println!("{}\n", serde_json::to_string_pretty(&message.data)?);

            println!("📋 Audit Trail:");
            for (i, audit) in message.audit_trail.iter().enumerate() {
                println!(
                    "{}. Task: {} (Status: {})",
                    i + 1,
                    audit.task_id,
                    audit.status_code
                );
                println!("   Timestamp: {}", audit.timestamp);
                println!("   Changes: {} field(s) modified", audit.changes.len());
            }

            if message.has_errors() {
                println!("\n⚠️  Errors encountered:");
                for error in &message.errors {
                    println!(
                        "   - {}: {:?}",
                        error.task_id.as_ref().unwrap_or(&"unknown".to_string()),
                        error.error_message
                    );
                }
            }
        }
        Err(e) => {
            println!("❌ Error processing message: {e:?}");
        }
    }

    // Demonstrate another example with different data
    let separator = "=".repeat(50);
    println!("\n{separator}");
    println!("=== Second Example with Different User ===\n");

    let mut message2 = dataflow_rs::engine::message::Message::new(&json!({}));
    message2.temp_data = json!({
        "measurements": [5.1, 7.3, 9.8, 6.2, 8.5],
        "user_id": "user_456",
        "timestamp": "2024-01-15T11:00:00Z"
    });
    message2.data = json!({});

    // Create a workflow for the second user
    let workflow2_json = r#"
    {
        "id": "custom_function_demo_2",
        "name": "Custom Function Demo 2",
        "description": "Second demo with different user",
        "condition": { "==": [true, true] },
        "tasks": [
            {
                "id": "prepare_data",
                "name": "Prepare Data",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.numbers",
                                "logic": { "var": "temp_data.measurements" }
                            },
                            {
                                "path": "data.user_id",
                                "logic": { "var": "temp_data.user_id" }
                            }
                        ]
                    }
                }
            },
            {
                "id": "calculate_stats",
                "name": "Calculate Statistics",
                "function": {
                    "name": "statistics",
                    "input": {
                        "data_path": "data.numbers",
                        "output_path": "data.analysis"
                    }
                }
            },
            {
                "id": "enrich_user_data",
                "name": "Enrich User Data",
                "function": {
                    "name": "enrich_data",
                    "input": {
                        "lookup_field": "user_id",
                        "lookup_value": "user_456",
                        "output_path": "data.employee_details"
                    }
                }
            }
        ]
    }
    "#;

    let workflow2 = Workflow::from_json(workflow2_json)?;
    engine.add_workflow(&workflow2);

    match engine.process_message(&mut message2).await {
        Ok(_) => {
            println!("✅ Second message processed successfully!\n");
            println!("📊 Results for user_456:");
            println!("{}", serde_json::to_string_pretty(&message2.data)?);
        }
        Err(e) => {
            println!("❌ Error processing second message: {e:?}");
        }
    }

    println!("\n🎉 Custom function examples completed!");

    Ok(())
}

pub fn add_error(&mut self, error: ErrorInfo)

Add an error to the message

pub fn has_errors(&self) -> bool

Check if message has errors

Examples found in repository

examples/benchmark.rs (line 201)

async fn run_async_benchmark(
    engine: &Engine,
    sample_user_data: &Value,
    num_iterations: usize,
) -> Result<BenchmarkResults, Box<dyn std::error::Error>> {
    let mut total_duration = Duration::new(0, 0);
    let mut min_duration = Duration::new(u64::MAX, 0);
    let mut max_duration = Duration::new(0, 0);
    let mut all_durations = Vec::with_capacity(num_iterations);
    let mut error_count = 0;

    println!("Starting async benchmark with {num_iterations} iterations...");

    for i in 0..num_iterations {
        let mut message = Message::new(&json!({}));
        message.temp_data = sample_user_data.clone();
        message.data = json!({});
        message.metadata = json!({
            "timestamp": chrono::Utc::now().to_rfc3339(),
            "iteration": i
        });

        let start = Instant::now();
        match engine.process_message(&mut message).await {
            Ok(_) => {
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);

                // Check for processing errors
                if message.has_errors() {
                    error_count += 1;
                    if error_count <= 5 {
                        // Only print first 5 errors
                        println!("Processing errors in iteration {}: {:?}", i, message.errors);
                    }
                }
            }
            Err(e) => {
                error_count += 1;
                if error_count <= 5 {
                    println!("Error in iteration {i}: {e:?}");
                }
                // Still record the time even for errors
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);
            }
        }

        if (i + 1) % 1000 == 0 {
            println!("Completed {} async iterations", i + 1);
        }
    }

    if error_count > 0 {
        println!("Total errors encountered: {error_count}");
    }

    // Sort durations for percentile calculations
    all_durations.sort();

    let p95_idx = (num_iterations as f64 * 0.95) as usize;
    let p99_idx = (num_iterations as f64 * 0.99) as usize;
    let p95 = all_durations.get(p95_idx).unwrap_or(&Duration::ZERO);
    let p99 = all_durations.get(p99_idx).unwrap_or(&Duration::ZERO);
    let avg_duration = total_duration / num_iterations as u32;

    println!("\nAsync Benchmark Results (v{VERSION}):");
    println!("  Iterations: {num_iterations}");
    println!("  Errors: {error_count}");
    println!("  Min time: {min_duration:?}");
    println!("  Max time: {max_duration:?}");
    println!("  Avg time: {avg_duration:?}");
    println!("  95th percentile: {p95:?}");
    println!("  99th percentile: {p99:?}");
    println!("  Total time: {total_duration:?}");

    Ok(BenchmarkResults {
        iterations: num_iterations,
        min_time: min_duration,
        max_time: max_duration,
        avg_time: avg_duration,
        p95: *p95,
        p99: *p99,
        total_time: total_duration,
    })
}

async fn run_sync_benchmark(
    engine: &Engine,
    sample_user_data: &Value,
    num_iterations: usize,
) -> Result<BenchmarkResults, Box<dyn std::error::Error>> {
    let mut total_duration = Duration::new(0, 0);
    let mut min_duration = Duration::new(u64::MAX, 0);
    let mut max_duration = Duration::new(0, 0);
    let mut all_durations = Vec::with_capacity(num_iterations);
    let mut error_count = 0;

    println!("Starting sync-style benchmark with {num_iterations} iterations...");

    for i in 0..num_iterations {
        let mut message = Message::new(&json!({}));
        message.temp_data = sample_user_data.clone();
        message.data = json!({});
        message.metadata = json!({
            "timestamp": chrono::Utc::now().to_rfc3339(),
            "iteration": i
        });

        let start = Instant::now();
        // Use tokio::task::block_in_place to simulate sync behavior
        let result = tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(engine.process_message(&mut message))
        });

        match result {
            Ok(_) => {
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);

                if message.has_errors() {
                    error_count += 1;
                }
            }
            Err(e) => {
                error_count += 1;
                if error_count <= 5 {
                    println!("Sync error in iteration {i}: {e:?}");
                }
                let duration = start.elapsed();
                all_durations.push(duration);
                total_duration += duration;
                min_duration = min_duration.min(duration);
                max_duration = max_duration.max(duration);
            }
        }

        if (i + 1) % 1000 == 0 {
            println!("Completed {} sync iterations", i + 1);
        }
    }

    if error_count > 0 {
        println!("Total sync errors encountered: {error_count}");
    }

    all_durations.sort();

    let p95_idx = (num_iterations as f64 * 0.95) as usize;
    let p99_idx = (num_iterations as f64 * 0.99) as usize;
    let p95 = all_durations.get(p95_idx).unwrap_or(&Duration::ZERO);
    let p99 = all_durations.get(p99_idx).unwrap_or(&Duration::ZERO);
    let avg_duration = total_duration / num_iterations as u32;

    println!("\nSync Benchmark Results (v{VERSION}):");
    println!("  Iterations: {num_iterations}");
    println!("  Errors: {error_count}");
    println!("  Min time: {min_duration:?}");
    println!("  Max time: {max_duration:?}");
    println!("  Avg time: {avg_duration:?}");
    println!("  95th percentile: {p95:?}");
    println!("  99th percentile: {p99:?}");
    println!("  Total time: {total_duration:?}");

    Ok(BenchmarkResults {
        iterations: num_iterations,
        min_time: min_duration,
        max_time: max_duration,
        avg_time: avg_duration,
        p95: *p95,
        p99: *p99,
        total_time: total_duration,
    })
}

examples/custom_function.rs (line 425)

async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    println!("=== Custom Function Example ===\n");

    // Create engine without built-in functions to demonstrate custom ones
    let mut engine = Engine::new_empty();

    // Register our custom functions
    engine.register_task_function(
        "statistics".to_string(),
        Box::new(StatisticsFunction::new()),
    );

    engine.register_task_function(
        "enrich_data".to_string(),
        Box::new(DataEnrichmentFunction::new()),
    );

    // Also register built-in map function for data preparation
    engine.register_task_function(
        "map".to_string(),
        Box::new(dataflow_rs::engine::functions::MapFunction::new()),
    );

    // Define a workflow that uses our custom functions
    let workflow_json = r#"
    {
        "id": "custom_function_demo",
        "name": "Custom Function Demo",
        "description": "Demonstrates custom async functions in workflow",
        "condition": { "==": [true, true] },
        "tasks": [
            {
                "id": "prepare_data",
                "name": "Prepare Data",
                "description": "Extract and prepare data for analysis",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.numbers",
                                "logic": { "var": "temp_data.measurements" }
                            },
                            {
                                "path": "data.user_id",
                                "logic": { "var": "temp_data.user_id" }
                            }
                        ]
                    }
                }
            },
            {
                "id": "calculate_stats",
                "name": "Calculate Statistics",
                "description": "Calculate statistical measures from numeric data",
                "function": {
                    "name": "statistics",
                    "input": {
                        "data_path": "data.numbers",
                        "output_path": "data.stats"
                    }
                }
            },
            {
                "id": "enrich_user_data",
                "name": "Enrich User Data",
                "description": "Add additional user information",
                "function": {
                    "name": "enrich_data",
                    "input": {
                        "lookup_field": "user_id",
                        "lookup_value": "user_123",
                        "output_path": "data.user_info"
                    }
                }
            }
        ]
    }
    "#;

    // Parse and add the workflow
    let workflow = Workflow::from_json(workflow_json)?;
    engine.add_workflow(&workflow);

    // Create sample data
    let sample_data = json!({
        "measurements": [10.5, 15.2, 8.7, 22.1, 18.9, 12.3, 25.6, 14.8, 19.4, 16.7],
        "user_id": "user_123",
        "timestamp": "2024-01-15T10:30:00Z"
    });

    // Create and process message
    let mut message = dataflow_rs::engine::message::Message::new(&json!({}));
    message.temp_data = sample_data;
    message.data = json!({});

    println!("Processing message with custom functions...\n");

    // Process the message through our custom workflow
    match engine.process_message(&mut message).await {
        Ok(_) => {
            println!("✅ Message processed successfully!\n");

            println!("📊 Final Results:");
            println!("{}\n", serde_json::to_string_pretty(&message.data)?);

            println!("📋 Audit Trail:");
            for (i, audit) in message.audit_trail.iter().enumerate() {
                println!(
                    "{}. Task: {} (Status: {})",
                    i + 1,
                    audit.task_id,
                    audit.status_code
                );
                println!("   Timestamp: {}", audit.timestamp);
                println!("   Changes: {} field(s) modified", audit.changes.len());
            }

            if message.has_errors() {
                println!("\n⚠️  Errors encountered:");
                for error in &message.errors {
                    println!(
                        "   - {}: {:?}",
                        error.task_id.as_ref().unwrap_or(&"unknown".to_string()),
                        error.error_message
                    );
                }
            }
        }
        Err(e) => {
            println!("❌ Error processing message: {e:?}");
        }
    }

    // Demonstrate another example with different data
    let separator = "=".repeat(50);
    println!("\n{separator}");
    println!("=== Second Example with Different User ===\n");

    let mut message2 = dataflow_rs::engine::message::Message::new(&json!({}));
    message2.temp_data = json!({
        "measurements": [5.1, 7.3, 9.8, 6.2, 8.5],
        "user_id": "user_456",
        "timestamp": "2024-01-15T11:00:00Z"
    });
    message2.data = json!({});

    // Create a workflow for the second user
    let workflow2_json = r#"
    {
        "id": "custom_function_demo_2",
        "name": "Custom Function Demo 2",
        "description": "Second demo with different user",
        "condition": { "==": [true, true] },
        "tasks": [
            {
                "id": "prepare_data",
                "name": "Prepare Data",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.numbers",
                                "logic": { "var": "temp_data.measurements" }
                            },
                            {
                                "path": "data.user_id",
                                "logic": { "var": "temp_data.user_id" }
                            }
                        ]
                    }
                }
            },
            {
                "id": "calculate_stats",
                "name": "Calculate Statistics",
                "function": {
                    "name": "statistics",
                    "input": {
                        "data_path": "data.numbers",
                        "output_path": "data.analysis"
                    }
                }
            },
            {
                "id": "enrich_user_data",
                "name": "Enrich User Data",
                "function": {
                    "name": "enrich_data",
                    "input": {
                        "lookup_field": "user_id",
                        "lookup_value": "user_456",
                        "output_path": "data.employee_details"
                    }
                }
            }
        ]
    }
    "#;

    let workflow2 = Workflow::from_json(workflow2_json)?;
    engine.add_workflow(&workflow2);

    match engine.process_message(&mut message2).await {
        Ok(_) => {
            println!("✅ Second message processed successfully!\n");
            println!("📊 Results for user_456:");
            println!("{}", serde_json::to_string_pretty(&message2.data)?);
        }
        Err(e) => {
            println!("❌ Error processing second message: {e:?}");
        }
    }

    println!("\n🎉 Custom function examples completed!");

    Ok(())
}

Trait Implementations

impl Clone for Message

fn clone(&self) -> Message

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for Message

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

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for Message

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for Message

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.