oximedia_workflow/
workflow_compose.rs

1//! Workflow composition: combine smaller workflows into larger meta-workflows.
2//!
3//! Provides tools for composing multiple independent workflows into a single
4//! execution unit with cross-workflow dependencies, shared variables, and
5//! unified lifecycle management.
6//!
7//! # Example
8//!
9//! ```rust
10//! use oximedia_workflow::workflow_compose::{MetaWorkflow, SubWorkflowRef, CrossDependency};
11//! use oximedia_workflow::workflow::Workflow;
12//!
13//! let ingest_wf = Workflow::new("ingest");
14//! let transcode_wf = Workflow::new("transcode");
15//!
16//! let mut meta = MetaWorkflow::new("full-pipeline");
17//! let ingest_ref = meta.add_sub_workflow(ingest_wf);
18//! let transcode_ref = meta.add_sub_workflow(transcode_wf);
19//! meta.add_dependency(CrossDependency::new(ingest_ref, transcode_ref));
20//!
21//! let order = meta.execution_order().expect("should succeed");
22//! assert_eq!(order.len(), 2);
23//! ```
24
25use crate::workflow::Workflow;
26use serde::{Deserialize, Serialize};
27use std::collections::{HashMap, HashSet, VecDeque};
28
29/// Reference handle to a sub-workflow within a meta-workflow.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
31pub struct SubWorkflowRef(usize);
32
33impl SubWorkflowRef {
34    /// Get the numeric index.
35    #[must_use]
36    pub const fn index(self) -> usize {
37        self.0
38    }
39}
40
41impl std::fmt::Display for SubWorkflowRef {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        write!(f, "sub-{}", self.0)
44    }
45}
46
47/// A dependency between two sub-workflows.
48#[derive(Debug, Clone, Serialize, Deserialize)]
49pub struct CrossDependency {
50    /// The sub-workflow that must complete first.
51    pub predecessor: SubWorkflowRef,
52    /// The sub-workflow that depends on the predecessor.
53    pub successor: SubWorkflowRef,
54    /// Optional condition: only start successor if this condition on the
55    /// predecessor's final variables is met.
56    pub condition: Option<String>,
57    /// Whether to pass variables from predecessor to successor.
58    pub pass_variables: bool,
59}
60
61impl CrossDependency {
62    /// Create a simple ordering dependency.
63    #[must_use]
64    pub fn new(predecessor: SubWorkflowRef, successor: SubWorkflowRef) -> Self {
65        Self {
66            predecessor,
67            successor,
68            condition: None,
69            pass_variables: false,
70        }
71    }
72
73    /// Add a condition to this dependency.
74    #[must_use]
75    pub fn with_condition(mut self, condition: impl Into<String>) -> Self {
76        self.condition = Some(condition.into());
77        self
78    }
79
80    /// Enable variable passing from predecessor to successor.
81    #[must_use]
82    pub fn with_variable_passing(mut self) -> Self {
83        self.pass_variables = true;
84        self
85    }
86}
87
88/// Error from meta-workflow operations.
89#[derive(Debug, thiserror::Error)]
90pub enum ComposeError {
91    /// Cycle detected between sub-workflows.
92    #[error("Cycle detected in meta-workflow dependencies")]
93    CycleDetected,
94
95    /// Referenced sub-workflow does not exist.
96    #[error("Sub-workflow not found: {0}")]
97    SubWorkflowNotFound(SubWorkflowRef),
98
99    /// Duplicate sub-workflow name.
100    #[error("Duplicate sub-workflow name: {0}")]
101    DuplicateName(String),
102}
103
104/// Execution status of a sub-workflow within the meta-workflow.
105#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
106pub enum SubWorkflowStatus {
107    /// Not yet started.
108    Pending,
109    /// Currently executing.
110    Running,
111    /// Completed successfully.
112    Completed,
113    /// Failed.
114    Failed(String),
115    /// Skipped due to unsatisfied condition.
116    Skipped,
117}
118
119impl SubWorkflowStatus {
120    /// Whether this status is terminal.
121    #[must_use]
122    pub fn is_terminal(&self) -> bool {
123        matches!(self, Self::Completed | Self::Failed(_) | Self::Skipped)
124    }
125}
126
127/// A meta-workflow that composes multiple sub-workflows.
128#[derive(Debug, Clone, Serialize, Deserialize)]
129pub struct MetaWorkflow {
130    /// Name of the meta-workflow.
131    pub name: String,
132    /// Description.
133    pub description: String,
134    /// Sub-workflows in insertion order.
135    pub sub_workflows: Vec<Workflow>,
136    /// Cross-workflow dependencies.
137    pub dependencies: Vec<CrossDependency>,
138    /// Per-sub-workflow execution status.
139    pub statuses: HashMap<SubWorkflowRef, SubWorkflowStatus>,
140    /// Shared variables accessible by all sub-workflows.
141    pub shared_variables: HashMap<String, serde_json::Value>,
142    /// Metadata.
143    pub metadata: HashMap<String, String>,
144}
145
146impl MetaWorkflow {
147    /// Create a new empty meta-workflow.
148    #[must_use]
149    pub fn new(name: impl Into<String>) -> Self {
150        Self {
151            name: name.into(),
152            description: String::new(),
153            sub_workflows: Vec::new(),
154            dependencies: Vec::new(),
155            statuses: HashMap::new(),
156            shared_variables: HashMap::new(),
157            metadata: HashMap::new(),
158        }
159    }
160
161    /// Set description.
162    #[must_use]
163    pub fn with_description(mut self, description: impl Into<String>) -> Self {
164        self.description = description.into();
165        self
166    }
167
168    /// Add a sub-workflow, returning its reference handle.
169    pub fn add_sub_workflow(&mut self, workflow: Workflow) -> SubWorkflowRef {
170        let idx = self.sub_workflows.len();
171        let ref_handle = SubWorkflowRef(idx);
172        self.statuses.insert(ref_handle, SubWorkflowStatus::Pending);
173        self.sub_workflows.push(workflow);
174        ref_handle
175    }
176
177    /// Add a cross-workflow dependency.
178    ///
179    /// # Errors
180    ///
181    /// Returns `ComposeError` if either reference is invalid.
182    pub fn add_dependency(&mut self, dep: CrossDependency) -> Result<(), ComposeError> {
183        if dep.predecessor.0 >= self.sub_workflows.len() {
184            return Err(ComposeError::SubWorkflowNotFound(dep.predecessor));
185        }
186        if dep.successor.0 >= self.sub_workflows.len() {
187            return Err(ComposeError::SubWorkflowNotFound(dep.successor));
188        }
189        self.dependencies.push(dep);
190        Ok(())
191    }
192
193    /// Set a shared variable.
194    pub fn set_variable(&mut self, key: impl Into<String>, value: serde_json::Value) {
195        self.shared_variables.insert(key.into(), value);
196    }
197
198    /// Get a sub-workflow by reference.
199    #[must_use]
200    pub fn get_sub_workflow(&self, ref_handle: SubWorkflowRef) -> Option<&Workflow> {
201        self.sub_workflows.get(ref_handle.0)
202    }
203
204    /// Get a mutable sub-workflow.
205    pub fn get_sub_workflow_mut(&mut self, ref_handle: SubWorkflowRef) -> Option<&mut Workflow> {
206        self.sub_workflows.get_mut(ref_handle.0)
207    }
208
209    /// Total number of sub-workflows.
210    #[must_use]
211    pub fn sub_workflow_count(&self) -> usize {
212        self.sub_workflows.len()
213    }
214
215    /// Total number of tasks across all sub-workflows.
216    #[must_use]
217    pub fn total_task_count(&self) -> usize {
218        self.sub_workflows.iter().map(|w| w.tasks.len()).sum()
219    }
220
221    /// Compute execution order (topological sort of sub-workflows).
222    ///
223    /// # Errors
224    ///
225    /// Returns `ComposeError::CycleDetected` if dependencies form a cycle.
226    pub fn execution_order(&self) -> Result<Vec<SubWorkflowRef>, ComposeError> {
227        let n = self.sub_workflows.len();
228        let mut in_degree = vec![0usize; n];
229        let mut adj: Vec<Vec<usize>> = vec![Vec::new(); n];
230
231        for dep in &self.dependencies {
232            adj[dep.predecessor.0].push(dep.successor.0);
233            in_degree[dep.successor.0] += 1;
234        }
235
236        let mut queue: VecDeque<usize> = VecDeque::new();
237        for (i, &deg) in in_degree.iter().enumerate() {
238            if deg == 0 {
239                queue.push_back(i);
240            }
241        }
242
243        let mut order = Vec::with_capacity(n);
244        while let Some(idx) = queue.pop_front() {
245            order.push(SubWorkflowRef(idx));
246            for &succ in &adj[idx] {
247                in_degree[succ] -= 1;
248                if in_degree[succ] == 0 {
249                    queue.push_back(succ);
250                }
251            }
252        }
253
254        if order.len() != n {
255            return Err(ComposeError::CycleDetected);
256        }
257
258        Ok(order)
259    }
260
261    /// Check if the dependency graph has a cycle.
262    #[must_use]
263    pub fn has_cycle(&self) -> bool {
264        self.execution_order().is_err()
265    }
266
267    /// Return predecessors of a sub-workflow.
268    #[must_use]
269    pub fn predecessors(&self, ref_handle: SubWorkflowRef) -> Vec<SubWorkflowRef> {
270        self.dependencies
271            .iter()
272            .filter(|d| d.successor == ref_handle)
273            .map(|d| d.predecessor)
274            .collect()
275    }
276
277    /// Return successors of a sub-workflow.
278    #[must_use]
279    pub fn successors(&self, ref_handle: SubWorkflowRef) -> Vec<SubWorkflowRef> {
280        self.dependencies
281            .iter()
282            .filter(|d| d.predecessor == ref_handle)
283            .map(|d| d.successor)
284            .collect()
285    }
286
287    /// Get root sub-workflows (no predecessors).
288    #[must_use]
289    pub fn root_sub_workflows(&self) -> Vec<SubWorkflowRef> {
290        let has_pred: HashSet<usize> = self.dependencies.iter().map(|d| d.successor.0).collect();
291        (0..self.sub_workflows.len())
292            .filter(|i| !has_pred.contains(i))
293            .map(SubWorkflowRef)
294            .collect()
295    }
296
297    /// Get leaf sub-workflows (no successors).
298    #[must_use]
299    pub fn leaf_sub_workflows(&self) -> Vec<SubWorkflowRef> {
300        let has_succ: HashSet<usize> = self.dependencies.iter().map(|d| d.predecessor.0).collect();
301        (0..self.sub_workflows.len())
302            .filter(|i| !has_succ.contains(i))
303            .map(SubWorkflowRef)
304            .collect()
305    }
306
307    /// Mark a sub-workflow status.
308    pub fn set_status(&mut self, ref_handle: SubWorkflowRef, status: SubWorkflowStatus) {
309        self.statuses.insert(ref_handle, status);
310    }
311
312    /// Get the status of a sub-workflow.
313    #[must_use]
314    pub fn get_status(&self, ref_handle: SubWorkflowRef) -> Option<&SubWorkflowStatus> {
315        self.statuses.get(&ref_handle)
316    }
317
318    /// Check whether all sub-workflows are complete (or skipped).
319    #[must_use]
320    pub fn is_complete(&self) -> bool {
321        self.statuses.values().all(SubWorkflowStatus::is_terminal)
322    }
323
324    /// Merge a finished meta-workflow's sub-workflows into a single flat `Workflow`.
325    ///
326    /// This creates a new workflow containing all tasks from all sub-workflows,
327    /// preserving internal edges and adding edges between leaf tasks of
328    /// predecessors and root tasks of successors as per dependencies.
329    #[must_use]
330    pub fn flatten(&self) -> Workflow {
331        use crate::task::TaskId;
332
333        let mut combined = Workflow::new(&self.name);
334        combined.description = self.description.clone();
335        combined.metadata = self.metadata.clone();
336
337        // Track task ID remapping per sub-workflow
338        let mut sub_task_maps: Vec<HashMap<TaskId, TaskId>> = Vec::new();
339
340        for wf in &self.sub_workflows {
341            let mut id_map = HashMap::new();
342
343            for task in wf.tasks.values() {
344                let mut new_task = task.clone();
345                let new_id = TaskId::new();
346                new_task.id = new_id;
347                new_task.name = format!("{}/{}", wf.name, task.name);
348                id_map.insert(task.id, new_id);
349                combined.tasks.insert(new_id, new_task);
350            }
351
352            // Re-create internal edges with new IDs
353            for edge in &wf.edges {
354                if let (Some(&new_from), Some(&new_to)) =
355                    (id_map.get(&edge.from), id_map.get(&edge.to))
356                {
357                    combined.edges.push(crate::workflow::Edge {
358                        from: new_from,
359                        to: new_to,
360                        condition: edge.condition.clone(),
361                    });
362                }
363            }
364
365            sub_task_maps.push(id_map);
366        }
367
368        // Add cross-dependency edges: leaf tasks of predecessor -> root tasks of successor
369        for dep in &self.dependencies {
370            let pred_wf = &self.sub_workflows[dep.predecessor.0];
371            let succ_wf = &self.sub_workflows[dep.successor.0];
372            let pred_map = &sub_task_maps[dep.predecessor.0];
373            let succ_map = &sub_task_maps[dep.successor.0];
374
375            let pred_leaves = pred_wf.get_leaf_tasks();
376            let succ_roots = succ_wf.get_root_tasks();
377
378            for &leaf_id in &pred_leaves {
379                for &root_id in &succ_roots {
380                    if let (Some(&new_leaf), Some(&new_root)) =
381                        (pred_map.get(&leaf_id), succ_map.get(&root_id))
382                    {
383                        combined.edges.push(crate::workflow::Edge {
384                            from: new_leaf,
385                            to: new_root,
386                            condition: dep.condition.clone(),
387                        });
388                    }
389                }
390            }
391        }
392
393        combined
394    }
395
396    /// Compute summary metrics for the meta-workflow.
397    #[must_use]
398    pub fn summary(&self) -> MetaWorkflowSummary {
399        let total_sub_workflows = self.sub_workflows.len();
400        let total_tasks = self.total_task_count();
401        let completed = self
402            .statuses
403            .values()
404            .filter(|s| matches!(s, SubWorkflowStatus::Completed))
405            .count();
406        let failed = self
407            .statuses
408            .values()
409            .filter(|s| matches!(s, SubWorkflowStatus::Failed(_)))
410            .count();
411        let pending = self
412            .statuses
413            .values()
414            .filter(|s| matches!(s, SubWorkflowStatus::Pending))
415            .count();
416        let running = self
417            .statuses
418            .values()
419            .filter(|s| matches!(s, SubWorkflowStatus::Running))
420            .count();
421        let skipped = self
422            .statuses
423            .values()
424            .filter(|s| matches!(s, SubWorkflowStatus::Skipped))
425            .count();
426
427        MetaWorkflowSummary {
428            name: self.name.clone(),
429            total_sub_workflows,
430            total_tasks,
431            completed,
432            failed,
433            pending,
434            running,
435            skipped,
436            is_complete: self.is_complete(),
437        }
438    }
439}
440
441/// Summary of a meta-workflow's execution state.
442#[derive(Debug, Clone, Serialize, Deserialize)]
443pub struct MetaWorkflowSummary {
444    /// Name.
445    pub name: String,
446    /// Total sub-workflows.
447    pub total_sub_workflows: usize,
448    /// Total tasks across all sub-workflows.
449    pub total_tasks: usize,
450    /// Completed sub-workflows.
451    pub completed: usize,
452    /// Failed sub-workflows.
453    pub failed: usize,
454    /// Pending sub-workflows.
455    pub pending: usize,
456    /// Running sub-workflows.
457    pub running: usize,
458    /// Skipped sub-workflows.
459    pub skipped: usize,
460    /// Whether all sub-workflows are terminal.
461    pub is_complete: bool,
462}
463
464#[cfg(test)]
465mod tests {
466    use super::*;
467    use crate::task::{Task, TaskType};
468    use std::time::Duration;
469
470    fn make_workflow(name: &str, task_count: usize) -> Workflow {
471        let mut wf = Workflow::new(name);
472        let mut prev_id = None;
473        for i in 0..task_count {
474            let task = Task::new(
475                format!("{name}-task-{i}"),
476                TaskType::Wait {
477                    duration: Duration::from_secs(1),
478                },
479            );
480            let id = wf.add_task(task);
481            if let Some(prev) = prev_id {
482                wf.add_edge(prev, id).expect("add edge");
483            }
484            prev_id = Some(id);
485        }
486        wf
487    }
488
489    #[test]
490    fn test_meta_workflow_creation() {
491        let meta = MetaWorkflow::new("pipeline");
492        assert_eq!(meta.name, "pipeline");
493        assert_eq!(meta.sub_workflow_count(), 0);
494    }
495
496    #[test]
497    fn test_add_sub_workflow() {
498        let mut meta = MetaWorkflow::new("pipeline");
499        let wf = make_workflow("ingest", 3);
500        let ref_handle = meta.add_sub_workflow(wf);
501
502        assert_eq!(ref_handle.index(), 0);
503        assert_eq!(meta.sub_workflow_count(), 1);
504        assert!(meta.get_sub_workflow(ref_handle).is_some());
505    }
506
507    #[test]
508    fn test_cross_dependency() {
509        let mut meta = MetaWorkflow::new("pipeline");
510        let r1 = meta.add_sub_workflow(make_workflow("ingest", 2));
511        let r2 = meta.add_sub_workflow(make_workflow("transcode", 2));
512
513        assert!(meta.add_dependency(CrossDependency::new(r1, r2)).is_ok());
514        assert_eq!(meta.predecessors(r2), vec![r1]);
515        assert_eq!(meta.successors(r1), vec![r2]);
516    }
517
518    #[test]
519    fn test_invalid_dependency() {
520        let mut meta = MetaWorkflow::new("pipeline");
521        let r1 = meta.add_sub_workflow(make_workflow("ingest", 1));
522        let bad_ref = SubWorkflowRef(99);
523
524        assert!(meta
525            .add_dependency(CrossDependency::new(r1, bad_ref))
526            .is_err());
527    }
528
529    #[test]
530    fn test_execution_order_linear() {
531        let mut meta = MetaWorkflow::new("pipeline");
532        let r1 = meta.add_sub_workflow(make_workflow("ingest", 1));
533        let r2 = meta.add_sub_workflow(make_workflow("transcode", 1));
534        let r3 = meta.add_sub_workflow(make_workflow("deliver", 1));
535
536        meta.add_dependency(CrossDependency::new(r1, r2))
537            .expect("dep");
538        meta.add_dependency(CrossDependency::new(r2, r3))
539            .expect("dep");
540
541        let order = meta.execution_order().expect("order");
542        assert_eq!(order.len(), 3);
543
544        let pos1 = order.iter().position(|&x| x == r1).expect("find r1");
545        let pos2 = order.iter().position(|&x| x == r2).expect("find r2");
546        let pos3 = order.iter().position(|&x| x == r3).expect("find r3");
547        assert!(pos1 < pos2);
548        assert!(pos2 < pos3);
549    }
550
551    #[test]
552    fn test_execution_order_cycle_detection() {
553        let mut meta = MetaWorkflow::new("cycle");
554        let r1 = meta.add_sub_workflow(make_workflow("a", 1));
555        let r2 = meta.add_sub_workflow(make_workflow("b", 1));
556
557        meta.add_dependency(CrossDependency::new(r1, r2))
558            .expect("dep");
559        meta.add_dependency(CrossDependency::new(r2, r1))
560            .expect("dep");
561
562        assert!(meta.has_cycle());
563        assert!(meta.execution_order().is_err());
564    }
565
566    #[test]
567    fn test_root_and_leaf_sub_workflows() {
568        let mut meta = MetaWorkflow::new("pipeline");
569        let r1 = meta.add_sub_workflow(make_workflow("a", 1));
570        let r2 = meta.add_sub_workflow(make_workflow("b", 1));
571        let r3 = meta.add_sub_workflow(make_workflow("c", 1));
572
573        meta.add_dependency(CrossDependency::new(r1, r2))
574            .expect("dep");
575        meta.add_dependency(CrossDependency::new(r2, r3))
576            .expect("dep");
577
578        assert_eq!(meta.root_sub_workflows(), vec![r1]);
579        assert_eq!(meta.leaf_sub_workflows(), vec![r3]);
580    }
581
582    #[test]
583    fn test_total_task_count() {
584        let mut meta = MetaWorkflow::new("pipeline");
585        meta.add_sub_workflow(make_workflow("ingest", 3));
586        meta.add_sub_workflow(make_workflow("transcode", 5));
587
588        assert_eq!(meta.total_task_count(), 8);
589    }
590
591    #[test]
592    fn test_status_tracking() {
593        let mut meta = MetaWorkflow::new("pipeline");
594        let r1 = meta.add_sub_workflow(make_workflow("a", 1));
595        let r2 = meta.add_sub_workflow(make_workflow("b", 1));
596
597        assert!(!meta.is_complete());
598
599        meta.set_status(r1, SubWorkflowStatus::Completed);
600        assert!(!meta.is_complete());
601
602        meta.set_status(r2, SubWorkflowStatus::Completed);
603        assert!(meta.is_complete());
604    }
605
606    #[test]
607    fn test_status_failed() {
608        let mut meta = MetaWorkflow::new("pipeline");
609        let r1 = meta.add_sub_workflow(make_workflow("a", 1));
610
611        meta.set_status(r1, SubWorkflowStatus::Failed("timeout".to_string()));
612        assert!(meta.is_complete());
613
614        let status = meta.get_status(r1).expect("status");
615        assert!(matches!(status, SubWorkflowStatus::Failed(msg) if msg == "timeout"));
616    }
617
618    #[test]
619    fn test_flatten_creates_combined_workflow() {
620        let mut meta = MetaWorkflow::new("pipeline");
621        let r1 = meta.add_sub_workflow(make_workflow("ingest", 2));
622        let r2 = meta.add_sub_workflow(make_workflow("transcode", 2));
623
624        meta.add_dependency(CrossDependency::new(r1, r2))
625            .expect("dep");
626
627        let flat = meta.flatten();
628        assert_eq!(flat.tasks.len(), 4);
629        // Internal edges (2 * 1) + cross edges (1 leaf * 1 root = 1)
630        assert_eq!(flat.edges.len(), 3);
631        assert!(!flat.has_cycle());
632    }
633
634    #[test]
635    fn test_flatten_preserves_names() {
636        let mut meta = MetaWorkflow::new("pipeline");
637        meta.add_sub_workflow(make_workflow("ingest", 1));
638
639        let flat = meta.flatten();
640        let task = flat.tasks.values().next().expect("task");
641        assert!(task.name.starts_with("ingest/"));
642    }
643
644    #[test]
645    fn test_flatten_conditional_dependency() {
646        let mut meta = MetaWorkflow::new("cond");
647        let r1 = meta.add_sub_workflow(make_workflow("check", 1));
648        let r2 = meta.add_sub_workflow(make_workflow("process", 1));
649
650        let dep = CrossDependency::new(r1, r2).with_condition("status == success");
651        meta.add_dependency(dep).expect("dep");
652
653        let flat = meta.flatten();
654        let cross_edge = flat
655            .edges
656            .iter()
657            .find(|e| e.condition.is_some())
658            .expect("conditional edge");
659        assert_eq!(cross_edge.condition, Some("status == success".to_string()));
660    }
661
662    #[test]
663    fn test_shared_variables() {
664        let mut meta = MetaWorkflow::new("pipeline");
665        let input = std::env::temp_dir()
666            .join("oximedia-workflow-compose-input.mp4")
667            .to_string_lossy()
668            .into_owned();
669        meta.set_variable("input_path", serde_json::json!(input));
670
671        assert_eq!(
672            meta.shared_variables.get("input_path"),
673            Some(&serde_json::json!(input))
674        );
675    }
676
677    #[test]
678    fn test_summary() {
679        let mut meta = MetaWorkflow::new("pipeline");
680        let r1 = meta.add_sub_workflow(make_workflow("a", 3));
681        let _r2 = meta.add_sub_workflow(make_workflow("b", 2));
682
683        meta.set_status(r1, SubWorkflowStatus::Completed);
684
685        let summary = meta.summary();
686        assert_eq!(summary.total_sub_workflows, 2);
687        assert_eq!(summary.total_tasks, 5);
688        assert_eq!(summary.completed, 1);
689        assert_eq!(summary.pending, 1);
690        assert!(!summary.is_complete);
691    }
692
693    #[test]
694    fn test_diamond_dependency() {
695        let mut meta = MetaWorkflow::new("diamond");
696        let r1 = meta.add_sub_workflow(make_workflow("start", 1));
697        let r2 = meta.add_sub_workflow(make_workflow("branch_a", 1));
698        let r3 = meta.add_sub_workflow(make_workflow("branch_b", 1));
699        let r4 = meta.add_sub_workflow(make_workflow("join", 1));
700
701        meta.add_dependency(CrossDependency::new(r1, r2))
702            .expect("dep");
703        meta.add_dependency(CrossDependency::new(r1, r3))
704            .expect("dep");
705        meta.add_dependency(CrossDependency::new(r2, r4))
706            .expect("dep");
707        meta.add_dependency(CrossDependency::new(r3, r4))
708            .expect("dep");
709
710        assert!(!meta.has_cycle());
711        let order = meta.execution_order().expect("order");
712        assert_eq!(order.len(), 4);
713
714        let pos1 = order.iter().position(|&x| x == r1).expect("find");
715        let pos2 = order.iter().position(|&x| x == r2).expect("find");
716        let pos3 = order.iter().position(|&x| x == r3).expect("find");
717        let pos4 = order.iter().position(|&x| x == r4).expect("find");
718        assert!(pos1 < pos2 && pos1 < pos3);
719        assert!(pos2 < pos4 && pos3 < pos4);
720    }
721
722    #[test]
723    fn test_sub_workflow_ref_display() {
724        let r = SubWorkflowRef(5);
725        assert_eq!(r.to_string(), "sub-5");
726    }
727}
oximedia_workflow/workflow_compose.rs

oximedia_workflow/
workflow_compose.rs
