d344fac14fd12eb31e274a93a5d726c21c38e18d

Committed 17 Oct 2024 07:25PM UTC coverage: 87.275% (-8.8%) from 96.041%

Build # d344fac14fd12eb31e274a93a5d726c21c38e18d

Build Type

push

github

Committed by

ChrisRega

Commit Message

Fix grcov

Run Details

2 of 3 new or added lines in 1 file covered. (66.67%)

80 existing lines in 5 files now uncovered.

583 of 668 relevant lines covered (87.28%)

2.36 hits per line

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

92.91

/src/immediatevalueprogress.rs

use crate::{BoxedSendError, 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_default()
    }
}

impl<T: Send + 'static, M> DirectCacheAccess<T, BoxedSendError> 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 get_result(&self) -> Option<Result<&T, &BoxedSendError>> {
        self.promise.get_result()
    }
    fn take_value(&mut self) -> Option<T> {
        self.promise.take_value()
    }
    fn take_result(&mut self) -> Option<Result<T, BoxedSendError>> {
        self.promise.take_result()
    }
}
#[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::{BoxedSendError, 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.last().map(\|p\| p.progress).unwrap_or_default()	2✔
129	}	2✔
130	}
131
132	impl<T: Send + 'static, M> DirectCacheAccess<T, BoxedSendError> for ProgressTrackedImValProm<T, M> {
133	fn get_value_mut(&mut self) -> Option<&mut T> {	1✔
134	self.promise.get_value_mut()	1✔
135	}	1✔
136	fn get_value(&self) -> Option<&T> {	3✔
137	self.promise.get_value()	3✔
138	}	3✔
UNCOV 139	fn get_result(&self) -> Option<Result<&T, &BoxedSendError>> {	×
UNCOV 140	self.promise.get_result()	×
UNCOV 141	}	×
142	fn take_value(&mut self) -> Option<T> {	1✔
143	self.promise.take_value()	1✔
144	}	1✔
UNCOV 145	fn take_result(&mut self) -> Option<Result<T, BoxedSendError>> {	×
UNCOV 146	self.promise.take_result()	×
UNCOV 147	}	×
148	}
149	#[cfg(test)]
150	mod test {
151	use super::*;
152	use crate::ImmediateValuePromise;
153	use std::time::Duration;
154	#[tokio::test]
155	async fn basic_usage_cycle() {	1✔
156	let mut oneshot_progress = ProgressTrackedImValProm::new(	1✔
157	\|s\| {	1✔
158	ImmediateValuePromise::new(async move {	1✔
159	s.send(StringStatus::from_str(	1✔
160	Progress::from_percent(0.0),	1✔
161	"Initializing",	1✔
162	))	1✔
163	.await	1✔
164	.unwrap();	1✔
165	tokio::time::sleep(Duration::from_millis(25)).await;	1✔
166	s.send(StringStatus::new(	1✔
167	Progress::from_percent(50.0),	1✔
168	"processing".into(),	1✔
169	))	1✔
170	.await	1✔
171	.unwrap();	1✔
172	tokio::time::sleep(Duration::from_millis(25)).await;	1✔
173		1✔
174	s.send(StringStatus::from_string(	1✔
175	Progress::from_percent(100.0),	1✔
176	format!("Done"),	1✔
177	))	1✔
178	.await	1✔
179	.unwrap();	1✔
180	Ok(34)	1✔
181	})	1✔
182	},	1✔
183	2000,	1✔
184	);	1✔
185	assert!(matches!(	1✔
186	oneshot_progress.poll_state(),	1✔
187	ImmediateValueState::Updating	1✔
188	));	1✔
189	assert!(!oneshot_progress.finished());	1✔
190		1✔
191	assert_eq!(*oneshot_progress.get_progress(), 0.0);	1✔
192	tokio::time::sleep(Duration::from_millis(100)).await;	1✔
193	let _ = oneshot_progress.poll_state();	1✔
194	assert_eq!(*oneshot_progress.get_progress(), 1.0);	1✔
195	let result = oneshot_progress.poll_state();	1✔
196		1✔
197	if let ImmediateValueState::Success(val) = result {	1✔
198	assert_eq!(*val, 34);	1✔
199	} else {	1✔
200	unreachable!();	1✔
201	}	1✔
202	// check finished	1✔
203	assert!(oneshot_progress.finished());	1✔
204	let history = oneshot_progress.status_history();	1✔
205	assert_eq!(history.len(), 3);	1✔
206		1✔
207	// check direct cache access trait	1✔
208	let val = oneshot_progress.get_value().unwrap();	1✔
209	assert_eq!(*val, 34);	1✔
210	let val = oneshot_progress.get_value_mut().unwrap();	1✔
211	*val = 33;	1✔
212	assert_eq!(*oneshot_progress.get_value().unwrap(), 33);	1✔
213	let val = oneshot_progress.take_value().unwrap();	1✔
214	assert_eq!(val, 33);	1✔
215	assert!(oneshot_progress.get_value().is_none());	1✔
216	}	1✔
217	}

ChrisRega / lazy_async_promise / d344fac14fd12eb31e274a93a5d726c21c38e18d

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