dbuff 0.1.0

Double-buffered state with async command chains, streaming, and keyed task pools for ratatui applications
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

use std::future::Future;

use async_trait::async_trait;

/// Signals whether a command chain should continue or abort.
///
/// The framework automatically maps `Ok` from [`Command::execute`] to
/// [`Continue`](ControlFlow::Continue) and `Err` to [`Break`](ControlFlow::Break).
/// [`ExecutionBuilder::go`] and [`DomainChain::go`] inspect each command's result
/// and short-circuit the chain on [`Break`](ControlFlow::Break).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ControlFlow {
    /// The command succeeded — proceed to the next command in the chain.
    Continue,
    /// The command failed — abort the remaining chain immediately.
    Break,
}

/// A domain command that can be executed asynchronously.
///
/// Implementors return [`Result<Self::Output, Self::Error>`]. The framework
/// automatically invokes the [`OnComplete`] callback and maps `Ok` to
/// [`ControlFlow::Continue`] / `Err` to [`ControlFlow::Break`].
#[async_trait]
pub trait Command<S>: Send {
    /// The type of value produced by this command.
    type Output: Send + Sync + 'static;
    /// The error type produced by this command on failure.
    type Error: Send + Sync + 'static;

    /// Execute the command with the given services.
    ///
    /// Return `Ok(output)` on success or `Err(e)` on failure.
    /// The framework handles calling the [`OnComplete`] callback and
    /// converting the result into [`ControlFlow`].
    async fn execute(self, services: S) -> Result<Self::Output, Self::Error>;
}

/// Run this after command execution completes.
pub trait OnComplete<T>: Send {
    /// `value` is the output of the command.
    fn run(self, value: T);
}

/// A builder for chaining sequential async command executions.
///
/// Each command added via [`ExecutionBuilder::new`] or [`then`](ExecutionBuilder::then)
/// is queued and executed in order when [`go`](ExecutionBuilder::go) is called.
/// Commands within a chain are guaranteed to run one after another,
/// with each completing before the next begins.
pub struct ExecutionBuilder<S> {
    tasks: Vec<std::pin::Pin<Box<dyn Future<Output = ControlFlow> + Send>>>,
    services: S,
    rt: tokio::runtime::Handle,
}

impl<S> ExecutionBuilder<S>
where
    S: Clone + Send + Sync + 'static,
{
    /// Begin a chained command execution (low-level API).
    ///
    /// Returns an [`ExecutionBuilder`] that can be used to chain additional
    /// commands via [`then`](ExecutionBuilder::then). Call [`go`](ExecutionBuilder::go)
    /// to spawn the entire chain on the tokio runtime.
    ///
    /// Commands in the chain execute sequentially — each completes before
    /// the next begins.
    pub fn new<C, F>(rt: tokio::runtime::Handle, services: S, cmd: C, on_complete: F) -> Self
    where
        C: Command<S> + 'static,
        F: OnComplete<Result<C::Output, C::Error>> + Send + 'static,
    {
        let services_clone = services.clone();
        let fut = async move {
            let result = cmd.execute(services_clone).await;
            let flow = if result.is_ok() {
                ControlFlow::Continue
            } else {
                ControlFlow::Break
            };
            on_complete.run(result);
            flow
        };
        Self {
            tasks: vec![Box::pin(fut)],
            services,
            rt,
        }
    }

    /// Append a command to the execution chain.
    ///
    /// The command will execute after all previously added commands
    /// have completed.
    #[must_use = "the builder must be consumed with `.go()` to spawn execution"]
    pub fn then<C, F>(mut self, cmd: C, on_complete: F) -> Self
    where
        C: Command<S> + 'static,
        F: OnComplete<Result<C::Output, C::Error>> + Send + 'static,
    {
        let services = self.services.clone();
        let fut = async move {
            let result = cmd.execute(services).await;
            let flow = if result.is_ok() {
                ControlFlow::Continue
            } else {
                ControlFlow::Break
            };
            on_complete.run(result);
            flow
        };
        self.tasks.push(Box::pin(fut));
        self
    }

    /// Spawn all chained commands as a single sequential task on the tokio runtime.
    ///
    /// Each command completes before the next one begins.
    /// Returns a [`JoinHandle`](tokio::task::JoinHandle) that resolves to the final
    /// [`ControlFlow`] — [`Break`](ControlFlow::Break) if any command short-circuited,
    /// otherwise [`Continue`](ControlFlow::Continue).
    pub fn go(self) -> tokio::task::JoinHandle<ControlFlow> {
        self.rt.spawn(async move {
            for task in self.tasks {
                if task.await == ControlFlow::Break {
                    return ControlFlow::Break;
                }
            }
            ControlFlow::Continue
        })
    }
}