vg-batteries 0.1.0

glimpse into my personal madness
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
157

use core::pin::pin;
use std::sync::{Arc, RwLock, RwLockReadGuard};

use dioxus::prelude::*;
use futures::{Future, Stream, StreamExt};

use tokio::sync::{oneshot, watch};



use super::periodic::SendOnDrop;

pub async fn cancellable<T>(
    kill: oneshot::Receiver<()>,
    task: impl Future<Output = T>,
) -> anyhow::Result<T> {
    tokio::select! {
        output = task => Ok(output),
        _ = kill => anyhow::bail!("Task was cancelled"),
    }
}

pub fn use_stream<'a, T, F, S, P>(
    cx: &'a Scope<P>,
    max_values: usize,
    stream_creator: impl Send + Sync + 'static + Fn() -> F,
) -> RwLockReadGuard<'a, Option<Vec<T>>>
where
    T: Send + Sync + 'static,
    F: Send + Future<Output = S>,
    S: Send + Stream<Item = T>,
{
    let (value, sender_max_vals, receiver_max_vals, _) = cx.use_hook(|| {
        let value = Arc::new(RwLock::new(None));

        let update = cx.schedule_update();

        let (sender_cancel, receiver_cancel) = oneshot::channel();
        let (sender_max_vals, receiver_max_vals) = watch::channel(max_values);

        {
            let mut receiver_max_vals = receiver_max_vals.clone();
            let value = value.clone();
            tokio::spawn(cancellable(receiver_cancel, async move {
                let mut stream = pin!(stream_creator().await);
                let mut cnt = 0;

                *value.write().unwrap() = Some(Vec::new());

                loop {
                    while cnt >= *receiver_max_vals.borrow() {
                        let Ok(_) = receiver_max_vals.changed().await else {
                            return;
                        };
                    }

                    let Some(val) = stream.next().await else {
                        return;
                    };

                    let mut opt = value.write().unwrap();
                    let vec = opt.as_mut().unwrap();
                    vec.push(val);
                    cnt += 1;

                    update();
                }
            }));
        }

        (
            value,
            sender_max_vals,
            receiver_max_vals,
            SendOnDrop::new(sender_cancel),
        )
    });

    if *receiver_max_vals.borrow() != max_values {
        sender_max_vals.send(max_values).unwrap();
    }

    value.read().unwrap()
}

pub fn use_last_stream_value<'a, Arg, T, F, S, P>(
    cx: &'a Scope<P>,
    arg: Arg,
    stream_creator: impl Send + Sync + 'static + Fn(Arg) -> F,
) -> RwLockReadGuard<'a, Option<T>>
where
    T: Send + Sync + 'static,
    F: Send + Future<Output = S>,
    S: Send + Stream<Item = T>,
    Arg: std::fmt::Debug + Eq + Clone + Send + Sync + 'static,
{
    let thearg = arg.clone();

    let (value, sender_args, receiver_args, _) = cx.use_hook(|| {
        let value = Arc::new(RwLock::new(None));

        let update = cx.schedule_update();

        let (sender_cancel, receiver_cancel) = oneshot::channel();
        let (sender_args, receiver_args) = watch::channel(arg);

        {
            let mut receiver_args = receiver_args.clone();
            let value = value.clone();

            tokio::spawn(cancellable(receiver_cancel, async move {
                loop {
                    let current_arg = receiver_args.borrow().clone();
                    let mut stream = pin!(stream_creator(current_arg.clone()).await);

                    loop {
                        tokio::select! {
                            _ = receiver_args.changed() => {
                                break;
                            }
                            val = stream.next() => {
                                match val {
                                    None => {
                                        break;
                                    }
                                    Some(val) => {
                                        *value.write().unwrap() = Some(val);
                                        update();
                                    }
                                }
                            }
                        }
                    }

                    if current_arg == *receiver_args.borrow()
                        && receiver_args.changed().await.is_err()
                    {
                        return;
                    }
                }
            }));
        }

        (
            value,
            sender_args,
            receiver_args,
            SendOnDrop::new(sender_cancel),
        )
    });

    if *receiver_args.borrow() != thearg {
        sender_args.send(thearg).unwrap();
    }

    value.read().unwrap()
}