sz_rust_workflow/engine/
task_manager.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-2026 SZ-Rust Team
3//
4use std::sync::Arc;
5
6use uuid::Uuid;
7
8use crate::error::WorkflowResult;
9use crate::instance::{PageRequest, PageResult, Task};
10use crate::repository::TaskRepository;
11
12/// 任务生命周期管理器。
13pub struct TaskManager {
14    task_repo: Arc<dyn TaskRepository>,
15}
16
17impl TaskManager {
18    pub fn new(task_repo: Arc<dyn TaskRepository>) -> Self {
19        Self { task_repo }
20    }
21
22    /// 为每个候选人生成待办任务。
23    pub async fn create_tasks(
24        &self,
25        instance_id: &str,
26        node_id: &str,
27        candidates: Vec<String>,
28    ) -> WorkflowResult<Vec<Task>> {
29        let mut tasks = Vec::with_capacity(candidates.len());
30        for candidate in candidates {
31            let task = Task::new_pending(
32                Uuid::new_v4().to_string(),
33                instance_id,
34                node_id,
35                vec![candidate],
36            );
37            self.task_repo.create(&task).await?;
38            tasks.push(task);
39        }
40        Ok(tasks)
41    }
42
43    /// 失效实例所有未完成任务。
44    pub async fn invalidate_by_instance(&self, instance_id: &str) -> WorkflowResult<u64> {
45        self.task_repo.invalidate_by_instance(instance_id).await
46    }
47
48    /// 分页查询候选人待办。
49    pub async fn list_pending_by_candidate(
50        &self,
51        candidate: &str,
52        page: PageRequest,
53    ) -> WorkflowResult<PageResult<Task>> {
54        self.task_repo
55            .list_pending_by_candidate(candidate, page)
56            .await
57    }
58
59    /// 获取任务。
60    pub async fn get(&self, task_id: &str) -> WorkflowResult<Option<Task>> {
61        self.task_repo.get(task_id).await
62    }
63
64    /// 更新任务。
65    pub async fn update(&self, task: &Task) -> WorkflowResult<()> {
66        self.task_repo.update(task).await
67    }
68
69    /// 列出实例所有任务。
70    pub async fn list_by_instance(&self, instance_id: &str) -> WorkflowResult<Vec<Task>> {
71        self.task_repo.list_by_instance(instance_id).await
72    }
73}
74
75#[cfg(test)]
76mod tests {
77    use super::*;
78    use crate::instance::TaskStatus;
79    use crate::repository::InMemoryTaskRepository;
80
81    #[tokio::test]
82    async fn create_tasks() {
83        let repo = Arc::new(InMemoryTaskRepository::default());
84        let mgr = TaskManager::new(repo);
85        let tasks = mgr
86            .create_tasks("i1", "n1", vec!["u1".into(), "u2".into()])
87            .await
88            .unwrap();
89        assert_eq!(tasks.len(), 2);
90        assert_eq!(tasks[0].status, TaskStatus::Pending);
91        assert_ne!(tasks[0].task_id, tasks[1].task_id);
92    }
93
94    #[tokio::test]
95    async fn invalidate() {
96        let repo = Arc::new(InMemoryTaskRepository::default());
97        let mgr = TaskManager::new(repo);
98        mgr.create_tasks("i1", "n1", vec!["u1".into(), "u2".into()])
99            .await
100            .unwrap();
101        let count = mgr.invalidate_by_instance("i1").await.unwrap();
102        assert_eq!(count, 2);
103    }
104
105    #[tokio::test]
106    async fn list_pending() {
107        let repo = Arc::new(InMemoryTaskRepository::default());
108        let mgr = TaskManager::new(repo);
109        mgr.create_tasks("i1", "n1", vec!["u1".into(), "u2".into()])
110            .await
111            .unwrap();
112        let result = mgr
113            .list_pending_by_candidate("u1", PageRequest::default())
114            .await
115            .unwrap();
116        assert_eq!(result.total, 1);
117    }
118}
sz_rust_workflow/engine/task_manager.rs

sz_rust_workflow/engine/
task_manager.rs
