2b79ab390055af124c291d8e99c965c349d5003e-PR-12

Build # 2b79ab390055af124c291d8e99c965c349d5003e-PR-12

Build Type

github

Committed by GitHub

Commit Message

Merge 02f4a87d2 into 50df9778c

Pull Request Pull Request #12: 0.5.0 dev branch

Run Details

246 of 246 new or added lines in 5 files covered. (100.0%)

612 of 635 relevant lines covered (96.38%)

3.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.04

/src/immediatevalueprogress.rs

use crate::{DirectCacheAccess, Progress};
use crate::{ImmediateValuePromise, ImmediateValueState};
use std::borrow::Cow;
use std::time::Instant;
use tokio::sync::mpsc::Receiver;
use tokio::sync::mpsc::Sender;

/// A status update struct containing the issue-date, progress and a message
/// You can use any struct that can be transferred via tokio mpsc channels.
#[derive(Debug)]
pub struct Status<M> {
    /// Time when this status was created
    pub time: Instant,
    /// Current progress
    pub progress: Progress,
    /// Message
    pub message: M,
}

impl<M> Status<M> {
    /// Create a new status message with `now` as timestamp
    pub fn new(progress: Progress, message: M) -> Self {
        Self {
            progress,
            message,
            time: Instant::now(),
        }
    }
}

/// This [`Status`] typedef allows to use both: `&'static str` and `String` in a message
pub type StringStatus = Status<Cow<'static, str>>;

impl StringStatus {
    /// create a [`StringStatus`] from a `&'static str`
    pub fn from_str(progress: Progress, static_message: &'static str) -> Self {
        StringStatus {
            message: Cow::Borrowed(static_message),
            time: Instant::now(),
            progress,
        }
    }
    /// create a [`StringStatus`] from a `String`
    pub fn from_string(progress: Progress, message: String) -> Self {
        StringStatus {
            message: Cow::Owned(message),
            time: Instant::now(),
            progress,
        }
    }
}

/// # A progress and status enabling wrapper for [`ImmediateValuePromise`]
/// This struct allows to use the [`Progress`] type and any kind of status message
/// You can use this to set a computation progress and optionally attach any kind of status message.
/// Assume your action runs  for an extended period of time and you want to inform the user about the state:
///```rust, no_run
///use std::borrow::Cow;
///use std::time::Duration;
///use lazy_async_promise::{ImmediateValueState, ImmediateValuePromise, Progress, ProgressTrackedImValProm, StringStatus};
///let mut oneshot_progress = ProgressTrackedImValProm::new( |s| { ImmediateValuePromise::new(
///  async move {
///  //send some initial status
///    s.send(StringStatus::new(
///      Progress::from_percent(0.0),
///      "Initializing".into(),
///    )).await.unwrap();
///    // do some long running operation
///    for i in 0..100 {
///      tokio::time::sleep(Duration::from_millis(50)).await;
///      s.send(StringStatus::new(
///        Progress::from_percent(i as f64),
///        Cow::Borrowed("In progress"))).await.unwrap();
///    }
///    Ok(34)
///  })}, 2000);
///  assert!(matches!(
///    oneshot_progress.poll_state(),
///    ImmediateValueState::Updating));
///   //waiting and polling will yield "In progress" now :)
/// ```
///
pub struct ProgressTrackedImValProm<T: Send, M> {
    promise: ImmediateValuePromise<T>,
    status: Vec<Status<M>>,
    receiver: Receiver<Status<M>>,
}

impl<T: Send + 'static, M> ProgressTrackedImValProm<T, M> {
    /// create a new Progress tracked immediate value promise.
    pub fn new(
        creator: impl FnOnce(Sender<Status<M>>) -> ImmediateValuePromise<T>,
        buffer: usize,
    ) -> Self {
        let (sender, receiver) = tokio::sync::mpsc::channel(buffer);
        ProgressTrackedImValProm {
            receiver,
            status: Vec::new(),
            promise: creator(sender),
        }
    }

    /// Slice of all recorded [`Status`] changes
    pub fn status_history(&self) -> &[Status<M>] {
        &self.status
    }

    /// Get the last [`Status`] if there is any
    pub fn last_status(&self) -> Option<&Status<M>> {
        self.status.last()
    }

    /// Is our future already finished?
    pub fn finished(&self) -> bool {
        self.promise.get_value().is_some()
    }

    /// Poll the state and process the messages
    pub fn poll_state(&mut self) -> &ImmediateValueState<T> {
        while let Ok(msg) = self.receiver.try_recv() {
            self.status.push(msg);
        }
        self.promise.poll_state()
    }

    /// Get the current progress
    pub fn get_progress(&self) -> Progress {
        self.status
            .last()
            .map(|p| p.progress)
            .unwrap_or(Progress::default())
    }
}

impl<T: Send + 'static, M> DirectCacheAccess<T> for ProgressTrackedImValProm<T, M> {
    fn get_value_mut(&mut self) -> Option<&mut T> {
        self.promise.get_value_mut()
    }
    fn get_value(&self) -> Option<&T> {
        self.promise.get_value()
    }
    fn take_value(&mut self) -> Option<T> {
        self.promise.take_value()
    }
}
#[cfg(test)]
mod test {
    use super::*;
    use crate::ImmediateValuePromise;
    use std::time::Duration;
    #[tokio::test]
    async fn basic_usage_cycle() {
        let mut oneshot_progress = ProgressTrackedImValProm::new(
            |s| {
                ImmediateValuePromise::new(async move {
                    s.send(StringStatus::from_str(
                        Progress::from_percent(0.0),
                        "Initializing",
                    ))
                    .await
                    .unwrap();
                    tokio::time::sleep(Duration::from_millis(25)).await;
                    s.send(StringStatus::new(
                        Progress::from_percent(50.0),
                        "processing".into(),
                    ))
                    .await
                    .unwrap();
                    tokio::time::sleep(Duration::from_millis(25)).await;

                    s.send(StringStatus::from_string(
                        Progress::from_percent(100.0),
                        format!("Done"),
                    ))
                    .await
                    .unwrap();
                    Ok(34)
                })
            },
            2000,
        );
        assert!(matches!(
            oneshot_progress.poll_state(),
            ImmediateValueState::Updating
        ));
        assert!(!oneshot_progress.finished());

        assert_eq!(*oneshot_progress.get_progress(), 0.0);
        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = oneshot_progress.poll_state();
        assert_eq!(*oneshot_progress.get_progress(), 1.0);
        let result = oneshot_progress.poll_state();

        if let ImmediateValueState::Success(val) = result {
            assert_eq!(*val, 34);
        } else {
            unreachable!();
        }
        // check finished
        assert!(oneshot_progress.finished());
        let history = oneshot_progress.status_history();
        assert_eq!(history.len(), 3);

        // check direct cache access trait
        let val = oneshot_progress.get_value().unwrap();
        assert_eq!(*val, 34);
        let val = oneshot_progress.get_value_mut().unwrap();
        *val = 33;
        assert_eq!(*oneshot_progress.get_value().unwrap(), 33);
        let val = oneshot_progress.take_value().unwrap();
        assert_eq!(val, 33);
        assert!(oneshot_progress.get_value().is_none());
    }
}

1	use crate::{DirectCacheAccess, Progress};
2	use crate::{ImmediateValuePromise, ImmediateValueState};
3	use std::borrow::Cow;
4	use std::time::Instant;
5	use tokio::sync::mpsc::Receiver;
6	use tokio::sync::mpsc::Sender;
7
8	/// A status update struct containing the issue-date, progress and a message
9	/// You can use any struct that can be transferred via tokio mpsc channels.
10	#[derive(Debug)]	×
11	pub struct Status<M> {
12	/// Time when this status was created
13	pub time: Instant,
14	/// Current progress
15	pub progress: Progress,
16	/// Message
17	pub message: M,
18	}
19
20	impl<M> Status<M> {
21	/// Create a new status message with `now` as timestamp
22	pub fn new(progress: Progress, message: M) -> Self {	1✔
23	Self {	1✔
24	progress,	1✔
25	message,	1✔
26	time: Instant::now(),	1✔
27	}	1✔
28	}	1✔
29	}
30
31	/// This [`Status`] typedef allows to use both: `&'static str` and `String` in a message
32	pub type StringStatus = Status<Cow<'static, str>>;
33
34	impl StringStatus {
35	/// create a [`StringStatus`] from a `&'static str`
36	pub fn from_str(progress: Progress, static_message: &'static str) -> Self {	1✔
37	StringStatus {	1✔
38	message: Cow::Borrowed(static_message),	1✔
39	time: Instant::now(),	1✔
40	progress,	1✔
41	}	1✔
42	}	1✔
43	/// create a [`StringStatus`] from a `String`
44	pub fn from_string(progress: Progress, message: String) -> Self {	1✔
45	StringStatus {	1✔
46	message: Cow::Owned(message),	1✔
47	time: Instant::now(),	1✔
48	progress,	1✔
49	}	1✔
50	}	1✔
51	}
52
53	/// # A progress and status enabling wrapper for [`ImmediateValuePromise`]
54	/// This struct allows to use the [`Progress`] type and any kind of status message
55	/// You can use this to set a computation progress and optionally attach any kind of status message.
56	/// Assume your action runs for an extended period of time and you want to inform the user about the state:
57	///```rust, no_run
58	///use std::borrow::Cow;
59	///use std::time::Duration;
60	///use lazy_async_promise::{ImmediateValueState, ImmediateValuePromise, Progress, ProgressTrackedImValProm, StringStatus};
61	///let mut oneshot_progress = ProgressTrackedImValProm::new( \|s\| { ImmediateValuePromise::new(
62	/// async move {
63	/// //send some initial status
64	/// s.send(StringStatus::new(
65	/// Progress::from_percent(0.0),
66	/// "Initializing".into(),
67	/// )).await.unwrap();
68	/// // do some long running operation
69	/// for i in 0..100 {
70	/// tokio::time::sleep(Duration::from_millis(50)).await;
71	/// s.send(StringStatus::new(
72	/// Progress::from_percent(i as f64),
73	/// Cow::Borrowed("In progress"))).await.unwrap();
74	/// }
75	/// Ok(34)
76	/// })}, 2000);
77	/// assert!(matches!(
78	/// oneshot_progress.poll_state(),
79	/// ImmediateValueState::Updating));
80	/// //waiting and polling will yield "In progress" now :)
81	/// ```
82	///
83	pub struct ProgressTrackedImValProm<T: Send, M> {
84	promise: ImmediateValuePromise<T>,
85	status: Vec<Status<M>>,
86	receiver: Receiver<Status<M>>,
87	}
88
89	impl<T: Send + 'static, M> ProgressTrackedImValProm<T, M> {
90	/// create a new Progress tracked immediate value promise.
91	pub fn new(	1✔
92	creator: impl FnOnce(Sender<Status<M>>) -> ImmediateValuePromise<T>,	1✔
93	buffer: usize,	1✔
94	) -> Self {	1✔
95	let (sender, receiver) = tokio::sync::mpsc::channel(buffer);	1✔
96	ProgressTrackedImValProm {	1✔
97	receiver,	1✔
98	status: Vec::new(),	1✔
99	promise: creator(sender),	1✔
100	}	1✔
101	}	1✔
102
103	/// Slice of all recorded [`Status`] changes
104	pub fn status_history(&self) -> &[Status<M>] {	1✔
105	&self.status	1✔
106	}	1✔
107
108	/// Get the last [`Status`] if there is any
109	pub fn last_status(&self) -> Option<&Status<M>> {	×
110	self.status.last()	×
111	}	×
112
113	/// Is our future already finished?
114	pub fn finished(&self) -> bool {	2✔
115	self.promise.get_value().is_some()	2✔
116	}	2✔
117
118	/// Poll the state and process the messages
119	pub fn poll_state(&mut self) -> &ImmediateValueState<T> {	3✔
120	while let Ok(msg) = self.receiver.try_recv() {	6✔
121	self.status.push(msg);	3✔
122	}	3✔
123	self.promise.poll_state()	3✔
124	}	3✔
125
126	/// Get the current progress
127	pub fn get_progress(&self) -> Progress {	2✔
128	self.status	2✔
129	.last()	2✔
130	.map(\|p\| p.progress)	2✔
131	.unwrap_or(Progress::default())	2✔
132	}	2✔
133	}
134
135	impl<T: Send + 'static, M> DirectCacheAccess<T> for ProgressTrackedImValProm<T, M> {
136	fn get_value_mut(&mut self) -> Option<&mut T> {	1✔
137	self.promise.get_value_mut()	1✔
138	}	1✔
139	fn get_value(&self) -> Option<&T> {	3✔
140	self.promise.get_value()	3✔
141	}	3✔
142	fn take_value(&mut self) -> Option<T> {	1✔
143	self.promise.take_value()	1✔
144	}	1✔
145	}
146	#[cfg(test)]
147	mod test {
148	use super::*;
149	use crate::ImmediateValuePromise;
150	use std::time::Duration;
151	#[tokio::test]	1✔
152	async fn basic_usage_cycle() {	1✔
153	let mut oneshot_progress = ProgressTrackedImValProm::new(	1✔
154	\|s\| {	1✔
155	ImmediateValuePromise::new(async move {	1✔
156	s.send(StringStatus::from_str(	1✔
157	Progress::from_percent(0.0),	1✔
158	"Initializing",	1✔
159	))	1✔
160	.await	×
161	.unwrap();	1✔
162	tokio::time::sleep(Duration::from_millis(25)).await;	1✔
163	s.send(StringStatus::new(	1✔
164	Progress::from_percent(50.0),	1✔
165	"processing".into(),	1✔
166	))	1✔
167	.await	×
168	.unwrap();	1✔
169	tokio::time::sleep(Duration::from_millis(25)).await;	1✔
170
171	s.send(StringStatus::from_string(	1✔
172	Progress::from_percent(100.0),	1✔
173	format!("Done"),	1✔
174	))	1✔
175	.await	×
176	.unwrap();	1✔
177	Ok(34)	1✔
178	})	1✔
179	},	1✔
180	2000,	1✔
181	);	1✔
182	assert!(matches!(	1✔
183	oneshot_progress.poll_state(),	1✔
184	ImmediateValueState::Updating
185	));
186	assert!(!oneshot_progress.finished());	1✔
187
188	assert_eq!(*oneshot_progress.get_progress(), 0.0);	1✔
189	tokio::time::sleep(Duration::from_millis(100)).await;	1✔
190	let _ = oneshot_progress.poll_state();	1✔
191	assert_eq!(*oneshot_progress.get_progress(), 1.0);	1✔
192	let result = oneshot_progress.poll_state();	1✔
193
194	if let ImmediateValueState::Success(val) = result {	1✔
195	assert_eq!(*val, 34);	1✔
196	} else {
197	unreachable!();	×
198	}
199	// check finished
200	assert!(oneshot_progress.finished());	1✔
201	let history = oneshot_progress.status_history();	1✔
202	assert_eq!(history.len(), 3);	1✔
203
204	// check direct cache access trait
205	let val = oneshot_progress.get_value().unwrap();	1✔
206	assert_eq!(*val, 34);	1✔
207	let val = oneshot_progress.get_value_mut().unwrap();	1✔
208	*val = 33;	1✔
209	assert_eq!(*oneshot_progress.get_value().unwrap(), 33);	1✔
210	let val = oneshot_progress.take_value().unwrap();	1✔
211	assert_eq!(val, 33);	1✔
212	assert!(oneshot_progress.get_value().is_none());	1✔
213	}
214	}

ChrisRega / lazy_async_promise / 2b79ab390055af124c291d8e99c965c349d5003e-PR-12

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous