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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

use dbuff::*;
use std::time::Duration;

type UserId = i32;

#[derive(Debug, Clone, PartialEq)]
struct User {
    id: UserId,
    name: String,
    email: String,
}

#[derive(Debug, Clone, Default)]
struct AppData {
    user_status: TaskStatus<User>,
    post_count: usize,
    posts_status: TaskStatus<usize>,
}

#[derive(Debug, Clone, wherror::Error)]
#[error(debug)]
struct AppError;

struct FetchUser(UserId);

#[async_trait::async_trait]
impl Command<()> for FetchUser {
    type Output = User;
    type Error = AppError;

    async fn execute(self, _: ()) -> Result<Self::Output, Self::Error> {
        tokio::time::sleep(Duration::from_millis(50)).await;
        Ok(User {
            id: self.0,
            name: format!("user_{}", self.0),
            email: format!("user_{}@example.com", self.0),
        })
    }
}

struct FetchPosts(UserId);

#[async_trait::async_trait]
impl Command<()> for FetchPosts {
    type Output = usize;
    type Error = AppError;

    async fn execute(self, _: ()) -> Result<Self::Output, Self::Error> {
        tokio::time::sleep(Duration::from_millis(50)).await;
        Ok(self.0.unsigned_abs() as usize * 10)
    }
}

#[tokio::main]
async fn main() {
    let rt = tokio::runtime::Handle::current();
    let (domain, write_handle) =
        SharedDomainData::with_coalesce(AppData::default(), Duration::from_micros(500));
    tokio::spawn(write_handle.run());

    // Show initial Idle state before spawning
    println!("user_status: {:?}", domain.read().user_status);
    println!("posts_status: {:?}", domain.read().posts_status);
    println!("---");

    // Chain: FetchUser returns User, tracked status stores it.
    // No separate user field needed — TaskStatus<User> holds the value.
    let handle = domain
        .bind((), rt.clone())
        .exec(FetchUser(42), |_, _| {})
        .tracked(|d, status| d.user_status = status)
        .then(|user| FetchPosts(user.id), |d, count| d.post_count = *count)
        .tracked(|d, status| d.posts_status = status)
        .go();

    // Poll user_status: Idle → Pending → Resolved
    loop {
        let state = domain.read();
        match &state.user_status {
            TaskStatus::Idle => {
                println!("user_status: idle");
            }
            TaskStatus::Pending => {
                println!("user_status: pending...");
            }
            TaskStatus::Resolved(user) => {
                println!("user_status: resolved ({} — {})", user.id, user.email);
                break;
            }
            TaskStatus::Error(e) => {
                println!("user_status: error ({e})");
                panic!("expected resolved, got error");
            }
            TaskStatus::Aborted => {
                println!("user_status: aborted");
                panic!("expected resolved, got aborted");
            }
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }

    // Poll posts_status: Pending → Resolved
    loop {
        let state = domain.read();
        match &state.posts_status {
            TaskStatus::Idle => {
                println!("posts_status: idle");
            }
            TaskStatus::Pending => {
                println!("posts_status: pending...");
            }
            TaskStatus::Resolved(count) => {
                println!("posts_status: resolved ({count})");
                break;
            }
            TaskStatus::Error(e) => {
                println!("posts_status: error ({e})");
                panic!("expected resolved, got error");
            }
            TaskStatus::Aborted => {
                println!("posts_status: aborted");
                panic!("expected resolved, got aborted");
            }
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }

    // Wait for full chain to complete
    let flow = handle.await.unwrap();
    assert_eq!(flow, ControlFlow::Continue);

    tokio::time::sleep(Duration::from_millis(10)).await;

    // Verify final state — read values directly from TaskStatus
    let state = domain.read();
    println!("---");
    println!("user_status = {:?}", state.user_status);
    println!("post_count = {}", state.post_count);
    println!("posts_status = {:?}", state.posts_status);

    let user = state.user_status.resolved().unwrap();
    assert_eq!(user.id, 42);
    assert_eq!(user.name, "user_42");
    assert_eq!(user.email, "user_42@example.com");
    assert_eq!(state.post_count, 420);
    assert!(state.user_status.is_resolved());
    assert!(state.posts_status.is_resolved());
    assert_eq!(*state.posts_status.resolved().unwrap(), 420);
}