flowbuilder-core 0.1.0

Core flow building functionality for FlowBuilder
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156

use crate::Flow;
use anyhow::Result;
use flowbuilder_context::{FlowContext, SharedContext};
use std::{future::Future, pin::Pin, time::Duration};

/// Type alias for step functions
pub type StepFuture = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
pub type Step = Box<dyn FnOnce(SharedContext) -> StepFuture + Send>;

/// Builder for creating flows with a fluent API
pub struct FlowBuilder {
    steps: Vec<Step>,
}

impl Default for FlowBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl FlowBuilder {
    /// Creates a new FlowBuilder
    pub fn new() -> Self {
        Self { steps: Vec::new() }
    }

    /// Adds a simple step to the flow
    pub fn step<Fut, F>(mut self, mut f: F) -> Self
    where
        F: FnMut(SharedContext) -> Fut + Send + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        self.steps.push(Box::new(move |ctx| Box::pin(f(ctx))));
        self
    }

    /// Adds a named step to the flow with automatic logging
    pub fn named_step<Fut, F>(mut self, name: &'static str, mut f: F) -> Self
    where
        F: FnMut(SharedContext) -> Fut + Send + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        self.steps.push(Box::new(move |ctx| {
            let ctx2 = ctx.clone();
            Box::pin(async move {
                // Start step logging
                {
                    let mut guard = ctx2.lock().await;
                    guard.start_step(name.to_string());
                }

                let result = f(ctx2.clone()).await;

                // End step logging
                {
                    let mut guard = ctx2.lock().await;
                    match &result {
                        Ok(()) => guard.end_step_success(name),
                        Err(e) => guard.end_step_failed(name, &e.to_string()),
                    }
                }

                result
            })
        }));
        self
    }

    /// Adds a conditional step that only executes if the condition is met
    pub fn step_if<Fut, F, Cond>(mut self, cond: Cond, mut f: F) -> Self
    where
        Cond: Fn(&FlowContext) -> bool + Send + Sync + 'static,
        F: FnMut(SharedContext) -> Fut + Send + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        self.steps.push(Box::new(move |ctx| {
            let ctx2 = ctx.clone();
            Box::pin(async move {
                let guard = ctx2.lock().await;
                if cond(&guard) {
                    drop(guard);
                    f(ctx2).await
                } else {
                    let trace_id = guard.trace_id.clone();
                    drop(guard);
                    println!(
                        "[trace_id:{trace_id}] [step_if] condition not met, skipping step"
                    );
                    Ok(())
                }
            })
        }));
        self
    }

    /// Adds a wait step that waits until a condition is met
    pub fn wait_until<Cond>(
        mut self,
        cond: Cond,
        interval: Duration,
        max_retry: usize,
    ) -> Self
    where
        Cond: Fn(&FlowContext) -> bool + Send + Sync + 'static,
    {
        self.steps.push(Box::new(move |ctx| {
            Box::pin(async move {
                for attempt in 0..max_retry {
                    {
                        let guard = ctx.lock().await;
                        if cond(&guard) {
                            println!(
                                "[wait_until] condition met on attempt {}",
                                attempt + 1
                            );
                            return Ok(());
                        }
                    }

                    if attempt < max_retry - 1 {
                        tokio::time::sleep(interval).await;
                    }
                }

                anyhow::bail!(
                    "[wait_until] condition not met after {} attempts",
                    max_retry
                )
            })
        }));
        self
    }

    /// Builds the flow
    pub fn build(self) -> Flow {
        Flow::new(self.steps)
    }

    /// Access steps for runtime extensions
    pub fn into_steps(self) -> Vec<Step> {
        self.steps
    }

    /// Builds and executes the flow immediately
    pub async fn execute(self) -> Result<FlowContext> {
        self.build().execute().await
    }

    /// Builds and executes the flow with a custom context
    pub async fn execute_with_context(
        self,
        context: FlowContext,
    ) -> Result<FlowContext> {
        self.build().execute_with_context(context).await
    }
}