Struct TaskFlow 

pub struct TaskFlow { /* private fields */ }

Implementations

impl TaskFlow

pub async fn new(config: TaskFlowConfig) -> Result<Self>

Examples found in repository

examples/basic_usage.rs (line 10)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/custom_handler.rs (line 173)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn from_yaml_file<P: AsRef<Path>>(path: P) -> Result<Self>

Examples found in repository

examples/yaml_config_usage.rs (line 10)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn from_yaml_str(config_content: &str) -> Result<Self>

pub async fn register_handler(&self, handler: Arc<dyn TaskHandler>)

Examples found in repository

examples/custom_handler.rs (line 175)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn submit_task(&self, definition: TaskDefinition) -> Result<String>

Examples found in repository

examples/basic_usage.rs (line 36)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/custom_handler.rs (line 187)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn submit_http_task( &self, name: &str, url: &str, method: Option<&str>, ) -> Result<String>

Examples found in repository

examples/yaml_config_usage.rs (line 35)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/basic_usage.rs (line 15)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn submit_shell_task( &self, name: &str, command: &str, args: Vec<&str>, ) -> Result<String>

Examples found in repository

examples/yaml_config_usage.rs (line 41)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/basic_usage.rs (line 21)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn get_task_status(&self, task_id: &str) -> Result<Option<TaskStatus>>

pub async fn cancel_task(&self, task_id: &str) -> Result<()>

pub async fn list_tasks(&self, status: Option<TaskStatus>) -> Result<Vec<Task>>

Examples found in repository

examples/yaml_config_usage.rs (line 73)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/basic_usage.rs (line 66)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/custom_handler.rs (line 254)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn start(&self) -> Result<()>

Examples found in repository

examples/yaml_config_usage.rs (line 50)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/basic_usage.rs (line 43)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/custom_handler.rs (line 231)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn shutdown(&self) -> Result<()>

Examples found in repository

examples/yaml_config_usage.rs (line 85)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/basic_usage.rs (line 78)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/custom_handler.rs (line 291)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

pub async fn wait_for_completion( &self, task_id: &str, timeout_seconds: Option<u64>, ) -> Result<Task>

pub async fn get_task_metrics(&self) -> Result<TaskMetrics>

Examples found in repository

examples/yaml_config_usage.rs (line 58)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    // Method 1: Load from YAML file
    let taskflow = TaskFlow::from_yaml_file("examples/config.yaml").await?;
    println!("TaskFlow framework started with YAML configuration!");

    // Method 2: You can also load from YAML string directly
    /*
    let yaml_config = r#"
scheduler:
  poll_interval_seconds: 2
  max_concurrent_tasks: 30
  enable_dependency_resolution: true
  cleanup_completed_tasks_after_hours: 24

executor:
  worker_id: "custom-worker-001"
  max_concurrent_tasks: 15
  task_timeout_seconds: 300
  heartbeat_interval_seconds: 10

storage_type: InMemory
"#;
    let taskflow = TaskFlow::from_yaml_str(yaml_config).await?;
    "#;
    */

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/basic_usage.rs (line 51)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    println!("TaskFlow framework started!");

    let task_id = taskflow
        .submit_http_task("fetch_example", "https://httpbin.org/get", Some("GET"))
        .await?;

    println!("Submitted HTTP task: {}", task_id);

    let shell_task_id = taskflow
        .submit_shell_task("list_files", "ls", vec!["-la"])
        .await?;

    println!("Submitted shell task: {}", shell_task_id);

    let dependent_task = TaskDefinition::new("dependent_task", "shell_command")
        .with_payload("command", serde_json::Value::String("echo".to_string()))
        .with_payload(
            "args",
            serde_json::Value::Array(vec![serde_json::Value::String(
                "This task depends on the shell task".to_string(),
            )]),
        )
        .with_dependencies(vec![shell_task_id.clone()]);

    let dependent_task_id = taskflow.submit_task(dependent_task).await?;
    println!("Submitted dependent task: {}", dependent_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(Duration::from_secs(2)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}, total={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed, metrics.total
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    println!("All tasks completed!");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!("Task: {} - Status: {:?}", task.definition.name, task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!("  Output: {:?}", result.output);
            } else {
                println!("  Error: {:?}", result.error);
            }
        }
    }

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}

examples/custom_handler.rs (line 239)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    init();

    let config = TaskFlowConfig::with_in_memory();
    let taskflow = TaskFlow::new(config).await?;

    taskflow.register_handler(Arc::new(MathTaskHandler)).await;
    taskflow
        .register_handler(Arc::new(DataProcessingHandler))
        .await;

    println!("TaskFlow with custom handlers started!");

    let add_task = TaskDefinition::new("addition", "math_operation")
        .with_payload("operation", serde_json::Value::String("add".to_string()))
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(10)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(5)));

    let add_task_id = taskflow.submit_task(add_task).await?;
    println!("Submitted addition task: {}", add_task_id);

    let multiply_task = TaskDefinition::new("multiplication", "math_operation")
        .with_payload(
            "operation",
            serde_json::Value::String("multiply".to_string()),
        )
        .with_payload("a", serde_json::Value::Number(serde_json::Number::from(7)))
        .with_payload("b", serde_json::Value::Number(serde_json::Number::from(3)));

    let multiply_task_id = taskflow.submit_task(multiply_task).await?;
    println!("Submitted multiplication task: {}", multiply_task_id);

    let data_array = vec![
        serde_json::Value::Number(serde_json::Number::from(1)),
        serde_json::Value::Number(serde_json::Number::from(2)),
        serde_json::Value::Number(serde_json::Number::from(3)),
        serde_json::Value::Number(serde_json::Number::from(4)),
        serde_json::Value::Number(serde_json::Number::from(5)),
    ];

    let sum_task = TaskDefinition::new("sum_data", "data_processing")
        .with_payload("operation", serde_json::Value::String("sum".to_string()))
        .with_payload("data", serde_json::Value::Array(data_array.clone()));

    let sum_task_id = taskflow.submit_task(sum_task).await?;
    println!("Submitted sum task: {}", sum_task_id);

    let avg_task = TaskDefinition::new("average_data", "data_processing")
        .with_payload(
            "operation",
            serde_json::Value::String("average".to_string()),
        )
        .with_payload("data", serde_json::Value::Array(data_array))
        .with_dependencies(vec![sum_task_id.clone()]);

    let avg_task_id = taskflow.submit_task(avg_task).await?;
    println!("Submitted average task (depends on sum): {}", avg_task_id);

    let taskflow_clone = std::sync::Arc::new(taskflow);
    let taskflow_for_execution = taskflow_clone.clone();

    let execution_handle = tokio::spawn(async move {
        if let Err(e) = taskflow_for_execution.start().await {
            eprintln!("TaskFlow execution failed: {}", e);
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    loop {
        let metrics = taskflow_clone.get_task_metrics().await?;
        println!(
            "Task metrics: pending={}, running={}, completed={}, failed={}",
            metrics.pending, metrics.running, metrics.completed, metrics.failed
        );

        if metrics.pending == 0 && metrics.running == 0 {
            break;
        }

        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    println!("\nAll tasks completed! Results:");

    let tasks = taskflow_clone.list_tasks(None).await?;
    for task in tasks {
        println!(
            "\nTask: {} ({})",
            task.definition.name, task.definition.task_type
        );
        println!("  Status: {:?}", task.status);
        if let Some(result) = &task.result {
            if result.success {
                println!(
                    "  Result: {}",
                    result.output.as_ref().unwrap_or(&"No output".to_string())
                );
                println!("  Execution time: {}ms", result.execution_time_ms);
                if !result.metadata.is_empty() {
                    println!("  Metadata: {:?}", result.metadata);
                }
            } else {
                println!(
                    "  Error: {}",
                    result
                        .error
                        .as_ref()
                        .unwrap_or(&"Unknown error".to_string())
                );
            }
        }
    }

    let final_metrics = taskflow_clone.get_task_metrics().await?;
    println!("\nFinal metrics:");
    println!("  Total tasks: {}", final_metrics.total);
    println!(
        "  Success rate: {:.1}%",
        final_metrics.success_rate() * 100.0
    );

    taskflow_clone.shutdown().await?;
    execution_handle.abort();

    Ok(())
}