use std::sync::Arc;
use tokio::sync::OnceCell;
use tokio::task::JoinHandle;
use std::future::Future;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum State {
NotInitialized,
Pending,
Completed,
TaskPanicked(String),
CallbackPanicked(String),
}
#[derive(Debug)]
pub struct Deferred<T> {
value: Arc<OnceCell<T>>,
task_handle: Option<JoinHandle<()>>,
panic_message: Option<String>,
is_callback_panic: bool,
}
impl<T> Deferred<T>
where
T: Send + Sync + 'static,
{
pub fn new() -> Self {
Self {
value: Arc::new(OnceCell::new()),
task_handle: None,
panic_message: None,
is_callback_panic: false
}
}
pub fn start<F, Fut>(computation: F) -> Self
where
F: FnOnce() -> Fut + Send + 'static,
Fut: Future<Output = T> + Send + 'static,
{
let mut deferred = Self::new();
deferred.begin(computation);
deferred
}
pub fn start_with_callback<F, C, Fut>(computation: F, callback: C) -> Self
where
F: FnOnce() -> Fut + Send + 'static,
C: FnOnce() + Send + 'static,
Fut: Future<Output = T> + Send + 'static,
{
let mut deferred = Self::new();
deferred.begin_with_callback(computation, callback);
deferred
}
pub fn begin<F, Fut>(&mut self, computation: F) -> bool
where
F: FnOnce() -> Fut + Send + 'static,
Fut: Future<Output = T> + Send + 'static,
{
self._begin(computation, None::<fn()>)
}
pub fn begin_with_callback<F, C, Fut>(&mut self, computation: F, callback: C) -> bool
where
F: FnOnce() -> Fut + Send + 'static,
C: FnOnce() + Send + 'static,
Fut: Future<Output = T> + Send + 'static,
{
self._begin(computation, Some(callback))
}
fn _begin<F, C, Fut>(&mut self, computation: F, callback: Option<C>) -> bool
where
F: FnOnce() -> Fut + Send + 'static,
C: FnOnce() + Send + 'static,
Fut: Future<Output = T> + Send + 'static,
{
if !self.is_not_initialized() {
return false;
}
let cell = self.value.clone();
let handle = tokio::spawn(async move {
let result = computation().await;
let _ = cell.set(result);
if let Some(callback) = callback {
callback();
}
});
self.task_handle = Some(handle);
true
}
pub fn state(&self) -> State {
if let Some(panic_msg) = &self.panic_message {
return if self.is_callback_panic {
State::CallbackPanicked(panic_msg.clone())
} else {
State::TaskPanicked(panic_msg.clone())
};
}
if let Some(handle) = &self.task_handle {
if handle.is_finished() {
if self.is_ready() {
State::Completed
} else {
State::TaskPanicked("Task may have panicked (use join() to get details)".to_string())
}
} else if self.is_ready() {
State::Completed
} else {
State::Pending
}
} else if self.is_ready() {
State::Completed
} else {
State::NotInitialized
}
}
pub async fn state_async(&mut self) -> State {
if let Some(panic_msg) = &self.panic_message {
return if self.is_callback_panic {
State::CallbackPanicked(panic_msg.clone())
} else {
State::TaskPanicked(panic_msg.clone())
};
}
if let Some(handle) = &self.task_handle {
if handle.is_finished() {
self.join().await;
if let Some(panic_msg) = &self.panic_message {
if self.is_callback_panic {
State::CallbackPanicked(panic_msg.clone())
} else {
State::TaskPanicked(panic_msg.clone())
}
} else if self.is_ready() {
State::Completed
} else {
State::Pending
}
} else if self.is_ready() {
State::Completed
} else {
State::Pending
}
} else {
State::NotInitialized
}
}
pub fn try_get(&self) -> Option<&T> {
self.value.get()
}
pub async fn join(&mut self) -> &Self {
if let Some(handle) = std::mem::take(&mut self.task_handle) {
let result = handle.await;
match result {
Ok(_) => {
},
Err(join_err) => {
let (panic_msg, is_callback_panic) = if join_err.is_panic() {
if let Ok(panic_payload) = join_err.try_into_panic() {
let panic_msg = if let Some(s) = panic_payload.downcast_ref::<&str>() {
format!("Panic message: {}", s)
} else if let Some(s) = panic_payload.downcast_ref::<String>() {
format!("Panic message: {}", s)
} else {
"Panic occurred, but couldn't get the message.".to_string()
};
(panic_msg, self.is_ready())
} else {
("Panic occurred, but couldn't extract panic info.".to_string(), false)
}
} else if join_err.is_cancelled() {
("Task was cancelled.".to_string(), false)
} else {
("Unknown join error.".to_string(), false)
};
self.panic_message = Some(panic_msg);
self.is_callback_panic = is_callback_panic;
}
}
}
self
}
pub fn cancel(&mut self) -> bool{
if self.is_pending() {
if let Some(handle) = &self.task_handle {
handle.abort();
}
self.task_handle = None;
self.value = Arc::new(OnceCell::new());
self.panic_message = None;
self.is_callback_panic = false;
true
} else {
false
}
}
pub fn take(&mut self) -> Option<T> {
if self.is_ready() {
let value = std::mem::take(&mut self.value);
self.task_handle = None;
Arc::try_unwrap(value).ok()?.into_inner()
} else {
None
}
}
pub fn has_task_panicked(&self) -> bool {
matches!(self.state(), State::TaskPanicked(_))
}
pub fn is_ready(&self) -> bool {
self.value.get().is_some()
}
pub fn is_complete(&self) -> bool {
matches!(self.state(), State::Completed)
}
pub fn is_pending(&self) -> bool {
matches!(self.state(), State::Pending)
}
pub fn is_not_initialized(&self) -> bool {
matches!(self.state(), State::NotInitialized)
}
}
impl<T> Default for Deferred<T>
where
T: Send + Sync + 'static,
{
fn default() -> Self {
Self::new()
}
}