Struct Workflow 

pub struct Workflow {
    pub id: String,
    pub name: String,
    pub priority: u32,
    pub description: Option<String>,
    pub condition: Value,
    pub condition_index: Option<usize>,
    pub tasks: Vec<Task>,
}

Workflow represents a collection of tasks that execute sequentially

Fields

id: String

name: String

priority: u32

description: Option<String>

condition: Value

condition_index: Option<usize>

tasks: Vec<Task>

Implementations

impl Workflow

pub fn new() -> Self

pub fn from_json(json_str: &str) -> Result<Self>

Load workflow from JSON string

Examples found in repository

examples/complete_workflow.rs (line 112)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Define a workflow that:
    // 1. Prepares sample user data
    // 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 enrich -> validate flow",
        "tasks": [
            {
                "id": "initialize_user",
                "name": "Initialize User Structure",
                "description": "Create empty user object in data",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.user",
                                "logic": {}
                            }
                        ]
                    }
                }
            },
            {
                "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 the workflow
    let workflow = Workflow::from_json(workflow_json)?;

    // Create the workflow engine with the workflow (built-in functions are auto-registered by default)
    let engine = Engine::new(vec![workflow], None, None);

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

    // Add sample user data to temp_data (simulating what would come from an API)
    message.temp_data = json!({
        "body": {
            "id": 1,
            "name": "John Doe",
            "email": "john.doe@example.com",
            "address": {
                "street": "123 Main St",
                "city": "New York"
            },
            "company": {
                "name": "Acme Corp"
            }
        }
    });

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

    match engine.process_message(&mut message) {
        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/benchmark.rs (line 89)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("========================================");
    println!("DATAFLOW ENGINE BENCHMARK");
    println!("========================================\n");
    println!(
        "Running {} iterations on single-threaded engine\n",
        ITERATIONS
    );

    // Define a simple workflow with data transformation
    let workflow_json = r#"
    {
        "id": "benchmark_workflow",
        "name": "Benchmark Workflow",
        "description": "Simple workflow for performance testing",
        "priority": 1,
        "tasks": [
            {
                "id": "transform_data",
                "name": "Transform Data",
                "description": "Map data fields",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.user.id", 
                                "logic": { "var": "temp_data.id" }
                            },
                            {
                                "path": "data.user.name", 
                                "logic": { "var": "temp_data.name" }
                            },
                            {
                                "path": "data.user.email", 
                                "logic": { "var": "temp_data.email" }
                            },
                            {
                                "path": "data.user.age",
                                "logic": { "+": [{ "var": "temp_data.age" }, 1] }
                            },
                            {
                                "path": "data.user.status",
                                "logic": { 
                                    "if": [
                                        { ">": [{ "var": "temp_data.age" }, 18] },
                                        "adult",
                                        "minor"
                                    ]
                                }
                            }
                        ]
                    }
                }
            },
            {
                "id": "validate_data",
                "name": "Validate Data",
                "description": "Validate transformed data",
                "function": {
                    "name": "validate",
                    "input": {
                        "rules": [
                            {
                                "path": "data",
                                "logic": { "!!": { "var": "data.user.id" } },
                                "message": "User ID is required"
                            },
                            {
                                "path": "data",
                                "logic": { "!!": { "var": "data.user.email" } },
                                "message": "User email is required"
                            }
                        ]
                    }
                }
            }
        ]
    }
    "#;

    // Parse the workflow
    let workflow = Workflow::from_json(workflow_json)?;

    // Create the engine with built-in functions
    let engine = Engine::new(vec![workflow], None, None);

    // Sample data for benchmarking
    let sample_data = json!({
        "id": 12345,
        "name": "John Doe",
        "email": "john.doe@example.com",
        "age": 25,
        "department": "Engineering"
    });

    // Warm-up run
    println!("Warming up...");
    for _ in 0..1000 {
        let mut message = Message::new(&json!({}));
        message.temp_data = sample_data.clone();
        let _ = engine.process_message(&mut message);
    }

    // Benchmark run
    println!("Starting benchmark...\n");

    let mut all_durations = Vec::with_capacity(ITERATIONS);
    let mut success_count = 0;
    let mut error_count = 0;

    let benchmark_start = Instant::now();

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

        let iteration_start = Instant::now();
        match engine.process_message(&mut message) {
            Ok(_) => {
                success_count += 1;
                if message.has_errors() {
                    error_count += 1;
                }
            }
            Err(_) => {
                error_count += 1;
            }
        }
        let iteration_duration = iteration_start.elapsed();
        all_durations.push(iteration_duration);

        // Progress indicator every 10k iterations
        if (i + 1) % 10000 == 0 {
            print!(".");
            use std::io::Write;
            std::io::stdout().flush()?;
        }
    }

    let total_time = benchmark_start.elapsed();
    println!("\n\nBenchmark Complete!");
    println!("==========================================\n");

    // Calculate statistics
    all_durations.sort_unstable();
    let p50 = all_durations[ITERATIONS * 50 / 100];
    let p90 = all_durations[ITERATIONS * 90 / 100];
    let p95 = all_durations[ITERATIONS * 95 / 100];
    let p99 = all_durations[ITERATIONS * 99 / 100];
    let throughput = ITERATIONS as f64 / total_time.as_secs_f64();

    // Display results
    println!("📊 PERFORMANCE METRICS");
    println!("──────────────────────────────────────────");
    println!("Total iterations:    {:>10}", ITERATIONS);
    println!("Successful:          {:>10}", success_count);
    println!("Errors:              {:>10}", error_count);
    println!(
        "Total time:          {:>10.3} seconds",
        total_time.as_secs_f64()
    );
    println!();

    println!("Messages/second:     {:>10.0}", throughput);
    println!();

    println!("📉 LATENCY PERCENTILES");
    println!("──────────────────────────────────────────");
    println!("P50:                 {:>10.3} μs", p50.as_micros() as f64);
    println!("P90:                 {:>10.3} μs", p90.as_micros() as f64);
    println!("P95:                 {:>10.3} μs", p95.as_micros() as f64);
    println!("P99:                 {:>10.3} μs", p99.as_micros() as f64);
    println!();

    Ok(())
}

examples/custom_function.rs (line 394)

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

    // Define a workflow that uses our custom functions
    let workflow_json = r#"
    {
        "id": "custom_function_demo",
        "name": "Custom Function Demo",
        "description": "Demonstrates custom functions in workflow",
        "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 the first workflow
    let workflow = Workflow::from_json(workflow_json)?;

    // 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",
        "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)?;

    // Prepare custom functions
    let mut custom_functions = HashMap::new();
    custom_functions.insert(
        "statistics".to_string(),
        Box::new(StatisticsFunction::new()) as Box<dyn FunctionHandler + Send + Sync>,
    );
    custom_functions.insert(
        "enrich_data".to_string(),
        Box::new(DataEnrichmentFunction::new()) as Box<dyn FunctionHandler + Send + Sync>,
    );
    // Note: map and validate are now built-in to the Engine and will be used automatically

    // Create engine with custom functions and built-ins (map/validate are always included internally)
    let engine = Engine::new(
        vec![workflow, workflow2],
        Some(custom_functions),
        None, // Use default (includes built-ins)
    );

    // Create sample data for first message
    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 first 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) {
        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:?}");
        }
    }

    // Process second message
    match engine.process_message(&mut message2) {
        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 from_file<P: AsRef<Path>>(path: P) -> Result<Self>

Load workflow from JSON file

pub fn validate(&self) -> Result<()>

Validate the workflow structure

Trait Implementations

impl Clone for Workflow

fn clone(&self) -> Workflow

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for Workflow

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

Formats the value using the given formatter.

impl Default for Workflow

fn default() -> Self

Returns the “default value” for a type.

impl<'de> Deserialize<'de> for Workflow

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

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.