async_deferred/lib.rs
1use std::sync::Arc;
2
3use tokio::sync::OnceCell;
4use tokio::task::JoinHandle;
5use std::future::Future;
6
7/// Represents the current state of an asynchronous task.
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub enum State {
10 /// The task hasn't been started yet.
11 NotInitialized,
12 /// The task is currently running.
13 Pending,
14 /// The task has completed successfully and returned a value.
15 Completed,
16 /// The task panicked during execution.
17 TaskPanicked(String),
18 /// The callback panicked during execution.
19 CallbackPanicked(String),
20
21}
22
23/// A handle to an asynchronous computation that allows for deferred result retrieval.
24///
25/// This struct supports the "fire-and-forget" pattern with the ability to later query the result
26/// or detect if the task panicked. It's useful when you want to start a computation but don't
27/// need the result immediately.
28///
29/// # Examples
30///
31/// ```rust
32/// use async_deferred::Deferred;
33///
34/// # tokio_test::block_on(async {
35/// // Start a computation immediately
36/// let mut deferred = Deferred::start(|| async {
37/// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
38/// 42
39/// });
40///
41/// // Later, get the result
42/// let result = deferred.join().await.try_get();
43/// assert_eq!(result, Some(&42));
44/// # })
45/// ```
46///
47/// ```
48/// // Create and start manually
49/// use async_deferred::Deferred;
50///
51/// # tokio_test::block_on(async {
52/// let mut deferred = Deferred::new();
53/// deferred.begin(|| async { "Hello, World!" });
54///
55/// // Check if ready without blocking (might not be ready yet)
56/// match deferred.try_get() {
57/// Some(result) => println!("Result: {}", result),
58/// None => println!("Not ready yet"),
59/// }
60/// # })
61/// ```
62///
63/// # Thread Safety
64///
65/// `Deferred<T>` is `Send` and `Sync` when `T` is `Send` and `Sync`, making it safe to
66/// share across threads and async tasks.
67///
68/// # Memory Management
69///
70/// Internally, the task runs on Tokio's async runtime, and the result is stored in a `OnceCell`.
71/// The computation runs independently and the result is cached until retrieved via `take()` or
72/// the `Deferred` is dropped.
73#[derive(Debug)]
74pub struct Deferred<T> {
75 value: Arc<OnceCell<T>>,
76 task_handle: Option<JoinHandle<()>>,
77 panic_message: Option<String>,
78 is_callback_panic: bool,
79}
80
81impl<T> Deferred<T>
82where
83 T: Send + Sync + 'static,
84{
85 /// Creates a new empty `Deferred` instance.
86 ///
87 /// The task must be started manually using [`begin`](Self::begin) or [`begin_with_callback`](Self::begin_with_callback).
88 ///
89 /// # Examples
90 ///
91 /// ```rust
92 /// use async_deferred::Deferred;
93 ///
94 /// # tokio_test::block_on(async {
95 /// let mut deferred = Deferred::new();
96 /// assert!(deferred.is_not_initialized());
97 ///
98 /// deferred.begin(|| async { 42 });
99 /// assert!(!deferred.is_not_initialized());
100 /// # })
101 /// ```
102 pub fn new() -> Self {
103 Self {
104 value: Arc::new(OnceCell::new()),
105 task_handle: None,
106 panic_message: None,
107 is_callback_panic: false
108 }
109 }
110
111 /// Starts a new asynchronous task and returns a `Deferred` handle.
112 ///
113 /// This is a convenient way to construct and start the deferred task in one step.
114 ///
115 /// # Arguments
116 /// * `computation` - An async function or closure that returns the result of the task.
117 ///
118 /// # Examples
119 ///
120 /// ```
121 /// use async_deferred::Deferred;
122 ///
123 /// # tokio_test::block_on(async {
124 /// let deferred = Deferred::start(|| async {
125 /// // Some computation
126 /// 42
127 /// });
128 ///
129 /// // Task is already running
130 /// assert!(deferred.is_pending() || deferred.is_ready());
131 /// # })
132 /// ```
133 pub fn start<F, Fut>(computation: F) -> Self
134 where
135 F: FnOnce() -> Fut + Send + 'static,
136 Fut: Future<Output = T> + Send + 'static,
137 {
138 let mut deferred = Self::new();
139 deferred.begin(computation);
140 deferred
141 }
142
143 /// Starts a new asynchronous task with a callback that is called once the task completes.
144 ///
145 /// The callback is executed after the computation finishes, regardless of whether it
146 /// succeeded or panicked. This is useful for logging, cleanup, or notifications.
147 ///
148 /// # Arguments
149 /// * `computation` - An async function or closure that returns the result of the task.
150 /// * `callback` - A closure that will be run after the task completes.
151 ///
152 /// # Examples
153 ///
154 /// ```rust
155 /// use async_deferred::Deferred;
156 /// # tokio_test::block_on(async {
157 /// let deferred = Deferred::start_with_callback(
158 /// || async { 42 },
159 /// || println!("Computation finished!")
160 /// );
161 /// # })
162 /// ```
163 pub fn start_with_callback<F, C, Fut>(computation: F, callback: C) -> Self
164 where
165 F: FnOnce() -> Fut + Send + 'static,
166 C: FnOnce() + Send + 'static,
167 Fut: Future<Output = T> + Send + 'static,
168 {
169 let mut deferred = Self::new();
170 deferred.begin_with_callback(computation, callback);
171 deferred
172 }
173
174 /// Begins execution of the deferred task.
175 ///
176 /// Returns `true` if the task was successfully started, or `false` if it has already
177 /// been started or completed.
178 ///
179 /// # Arguments
180 /// * `computation` - An async function or closure that returns the result of the task.
181 ///
182 /// # Examples
183 ///
184 /// ```rust
185 /// use async_deferred::Deferred;
186 /// # tokio_test::block_on(async {
187 /// let mut deferred = Deferred::new();
188 ///
189 /// assert!(deferred.begin(|| async { 42 })); // Started successfully
190 /// assert!(!deferred.begin(|| async { 24 })); // Already started, returns false
191 ///
192 /// # })
193 /// ```
194 pub fn begin<F, Fut>(&mut self, computation: F) -> bool
195 where
196 F: FnOnce() -> Fut + Send + 'static,
197 Fut: Future<Output = T> + Send + 'static,
198 {
199 self._begin(computation, None::<fn()>)
200 }
201
202 /// Begins execution of the deferred task and executes a callback after completion.
203 ///
204 /// Returns `true` if the task was successfully started, or `false` if it has already
205 /// been started or completed.
206 ///
207 /// # Arguments
208 /// * `computation` - An async function or closure that returns the result of the task.
209 /// * `callback` - A closure that will be run after the task completes.
210 ///
211 /// # Examples
212 ///
213 /// ```rust
214 /// use async_deferred::Deferred;
215 ///
216 /// # tokio_test::block_on(async {
217 /// let mut deferred = Deferred::new();
218 ///
219 /// let started = deferred.begin_with_callback(
220 /// || async { 42 },
221 /// || println!("Task completed!")
222 /// );
223 /// assert!(started);
224 /// # })
225 /// ```
226 pub fn begin_with_callback<F, C, Fut>(&mut self, computation: F, callback: C) -> bool
227 where
228 F: FnOnce() -> Fut + Send + 'static,
229 C: FnOnce() + Send + 'static,
230 Fut: Future<Output = T> + Send + 'static,
231 {
232 self._begin(computation, Some(callback))
233 }
234
235 /// Internal implementation for starting tasks with optional callbacks.
236 fn _begin<F, C, Fut>(&mut self, computation: F, callback: Option<C>) -> bool
237 where
238 F: FnOnce() -> Fut + Send + 'static,
239 C: FnOnce() + Send + 'static,
240 Fut: Future<Output = T> + Send + 'static,
241 {
242 if !self.is_not_initialized() {
243 return false;
244 }
245
246 let cell = self.value.clone();
247
248 let handle = tokio::spawn(async move {
249 let result = computation().await;
250 let _ = cell.set(result);
251 if let Some(callback) = callback {
252 callback();
253 }
254 });
255
256 self.task_handle = Some(handle);
257 true
258 }
259
260 /// Returns the current state of the task.
261 ///
262 /// # Returns
263 /// - `State::NotInitialized` - The task hasn't been started yet
264 /// - `State::Pending` - The task is currently running
265 /// - `State::Completed` - The task completed successfully or the callback panicked
266 /// - `State::TaskPanicked` - The task panicked during execution
267 ///
268 /// [`state`](Self::state) only detects if the task has panicked and not the callback.
269 /// If you need this information, use [`join`](Self::join) or [`state_async`](Self::state_async).
270 ///
271 /// # Examples
272 ///
273 /// ```rust
274 /// use async_deferred::{Deferred, State};
275 ///
276 /// # tokio_test::block_on(async {
277 /// let mut deferred: Deferred<u32> = Deferred::new();
278 /// assert!(matches!(deferred.state(), State::NotInitialized));
279 /// deferred.begin(|| async {
280 /// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
281 /// 42
282 /// });
283 /// assert!(matches!(deferred.state(), State::Pending));
284 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
285 /// assert!(matches!(deferred.state(), State::Completed));
286 /// # })
287 /// ```
288 /// If panicked due to the callback, [`state`](Self::state) will return State::Completed.
289 /// ```rust
290 /// use async_deferred::{Deferred, State};
291 ///
292 /// # tokio_test::block_on(async {
293 /// let mut deferred: Deferred<u32> = Deferred::start_with_callback(
294 /// || async {42},
295 /// || {panic!("the callback panicked");},
296 /// );
297 /// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
298 ///
299 /// // state only detects if the task has panicked and not the callback
300 /// assert!(matches!(deferred.state(), State::Completed));
301 /// # })
302 /// ```
303 pub fn state(&self) -> State {
304 // If we already detected a panic, return the appropriate state
305 if let Some(panic_msg) = &self.panic_message {
306 return if self.is_callback_panic {
307 State::CallbackPanicked(panic_msg.clone())
308 } else {
309 State::TaskPanicked(panic_msg.clone())
310 };
311 }
312
313 // Check if we have a task handle
314 if let Some(handle) = &self.task_handle {
315 if handle.is_finished() {
316 // Task is finished, but we can't check for panic synchronously
317 // We'll need to await the result in an async context
318 // For now, if the task is finished but we don't have a result, assume it panicked
319 if self.is_ready() {
320 State::Completed
321 } else {
322 // Task finished but no result - likely panicked
323 // We'll set a default message until we can properly check
324 State::TaskPanicked("Task may have panicked (use join() to get details)".to_string())
325 }
326 } else if self.is_ready() {
327 // Task is still running but result is available (shouldn't happen)
328 State::Completed
329 } else {
330 // Task is still running
331 State::Pending
332 }
333 } else if self.is_ready() {
334 // handle may be none if the handle has been consumed by the join
335 State::Completed
336 } else {
337 State::NotInitialized
338 }
339 }
340
341
342 /// Returns the current state of the task. This method is able to correctly
343 /// diagnose when a given callback panics, as opposed to [`state`](Self::state).
344 /// Though asynchronous, it will not block until the task completes.
345 ///
346 /// # Returns
347 /// - `State::NotInitialized` - The task hasn't been started yet
348 /// - `State::Pending` - The task is currently running
349 /// - `State::Completed` - The task completed successfully
350 /// - `State::TaskPanicked` - The task panicked during execution
351 /// - `State::CallbackPanicked` - The callback panicked during execution
352 ///
353 /// # Examples
354 /// ```rust
355 /// use async_deferred::{Deferred, State};
356 ///
357 /// # tokio_test::block_on(async {
358 /// let mut deferred: Deferred<u32> = Deferred::new();
359 /// assert!(matches!(deferred.state_async().await, State::NotInitialized));
360 /// deferred.begin(|| async {
361 /// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
362 /// 42
363 /// });
364 ///
365 /// // The task is not done yet
366 /// assert!(matches!(deferred.state_async().await, State::Pending));
367 ///
368 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
369 ///
370 /// // The task is completed
371 /// assert!(matches!(deferred.state_async().await, State::Completed));
372 /// # })
373 /// ```
374 ///
375 /// ```rust
376 /// use async_deferred::{Deferred, State};
377 ///
378 /// # tokio_test::block_on(async {
379 /// let mut deferred: Deferred<u32> = Deferred::start_with_callback(
380 /// || async { panic!("the task panicked"); 42 },
381 /// || { panic!("callback will not be called since the task panicked"); },
382 /// );
383 /// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
384 ///
385 /// assert!(matches!(deferred.state_async().await, State::TaskPanicked(_)));
386 /// # })
387 /// ```
388 ///
389 /// ```rust
390 /// use async_deferred::{Deferred, State};
391 ///
392 /// # tokio_test::block_on(async {
393 /// let mut deferred: Deferred<u32> = Deferred::start_with_callback(
394 /// || async { 42 },
395 /// || { panic!("the callback panicked"); },
396 /// );
397 /// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
398 ///
399 /// assert!(matches!(deferred.state_async().await, State::CallbackPanicked(_)));
400 /// # })
401 pub async fn state_async(&mut self) -> State {
402 // If we already detected a panic, return the appropriate state
403 if let Some(panic_msg) = &self.panic_message {
404 return if self.is_callback_panic {
405 State::CallbackPanicked(panic_msg.clone())
406 } else {
407 State::TaskPanicked(panic_msg.clone())
408 };
409 }
410
411 // Check if we have a task handle
412 if let Some(handle) = &self.task_handle {
413 if handle.is_finished() {
414 // We can safely await the finished task
415 self.join().await;
416
417 // Now check the state again
418 if let Some(panic_msg) = &self.panic_message {
419 if self.is_callback_panic {
420 State::CallbackPanicked(panic_msg.clone())
421 } else {
422 State::TaskPanicked(panic_msg.clone())
423 }
424 } else if self.is_ready() {
425 State::Completed
426 } else {
427 // This shouldn't happen
428 State::Pending
429 }
430 } else if self.is_ready() {
431 State::Completed
432 } else {
433 State::Pending
434 }
435 } else {
436 State::NotInitialized
437 }
438 }
439 /// Attempts to retrieve the result of the computation if available.
440 ///
441 /// Returns `Some(&T)` if the task has completed successfully, or `None` if the task
442 /// is still pending, panicked, or not started.
443 ///
444 /// This is a non-blocking operation that returns immediately.
445 ///
446 /// # Examples
447 ///
448 /// ```rust
449 /// use async_deferred::Deferred;
450 /// # tokio_test::block_on(async {
451 /// let deferred = Deferred::start(|| async { 42 });
452 ///
453 /// // Might return None if task hasn't finished yet
454 /// if let Some(result) = deferred.try_get() {
455 /// println!("Result: {}", result);
456 /// }
457 /// # })
458 /// ```
459 pub fn try_get(&self) -> Option<&T> {
460 self.value.get()
461 }
462
463 /// Waits for the task to complete and returns a reference to self.
464 ///
465 /// If the task is still pending, this will await its completion (either success or panic).
466 /// If the task has already completed or panicked, this returns immediately.
467 ///
468 /// This method allows for method chaining: `deferred.join().await.try_get()`.
469 ///
470 /// # Examples
471 ///
472 /// ```rust
473 /// use async_deferred::Deferred;
474 ///
475 /// # tokio_test::block_on(async {
476 /// let mut deferred = Deferred::start(|| async {
477 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
478 /// 42
479 /// });
480 ///
481 /// // Wait for completion and get result
482 /// let result = deferred.join().await.try_get();
483 /// assert_eq!(result, Some(&42));
484 /// # })
485 /// ```
486 pub async fn join(&mut self) -> &Self {
487 if let Some(handle) = std::mem::take(&mut self.task_handle) {
488 let result = handle.await;
489 match result {
490 Ok(_) => {
491 // Task completed successfully
492 },
493 Err(join_err) => {
494 // Check if it's a callback panic and extract message
495 let (panic_msg, is_callback_panic) = if join_err.is_panic() {
496 if let Ok(panic_payload) = join_err.try_into_panic() {
497 let panic_msg = if let Some(s) = panic_payload.downcast_ref::<&str>() {
498 format!("Panic message: {}", s)
499 } else if let Some(s) = panic_payload.downcast_ref::<String>() {
500 format!("Panic message: {}", s)
501 } else {
502 "Panic occurred, but couldn't get the message.".to_string()
503 };
504
505 (panic_msg, self.is_ready())
506 } else {
507 ("Panic occurred, but couldn't extract panic info.".to_string(), false)
508 }
509 } else if join_err.is_cancelled() {
510 ("Task was cancelled.".to_string(), false)
511 } else {
512 ("Unknown join error.".to_string(), false)
513 };
514
515 self.panic_message = Some(panic_msg);
516 self.is_callback_panic = is_callback_panic;
517 }
518 }
519 }
520 self
521 }
522
523 /// Attempts to take ownership of the computed value.
524 ///
525 /// Returns `Some(T)` if the task has completed successfully, consuming the stored result.
526 /// Returns `None` if the task is still pending, panicked, or not started.
527 ///
528 /// After calling this method successfully, subsequent calls will return `None` since
529 /// the value has been moved out.
530 ///
531 /// # Examples
532 ///
533 /// ```rust
534 /// use async_deferred::Deferred;
535 ///
536 /// # tokio_test::block_on(async {
537 /// let mut deferred = Deferred::start(|| async { vec![1, 2, 3] });
538 ///
539 /// // Wait for completion
540 /// deferred.join().await;
541 ///
542 /// // Take ownership of the result
543 /// let result = deferred.take();
544 /// assert_eq!(result, Some(vec![1, 2, 3]));
545 ///
546 /// // Subsequent calls return None
547 /// assert_eq!(deferred.take(), None);
548 /// # })
549 /// ```
550 ///
551 /// [`take`](Self::take) will not consume if it is still pending
552 /// ```rust
553 /// use async_deferred::Deferred;
554 ///
555 /// # tokio_test::block_on(async {
556 /// let mut deferred = Deferred::start(|| async {
557 /// tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
558 /// 42
559 /// });
560 /// // If still pending, doesn't consume it
561 /// assert_eq!(deferred.take(), None);
562 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
563 /// assert!(matches!(deferred.try_get(), Some(42)));
564 /// # })
565 /// ```
566 ///
567 pub fn take(&mut self) -> Option<T> {
568 if self.is_ready() {
569 let value = std::mem::take(&mut self.value);
570 self.task_handle = None;
571 Arc::try_unwrap(value).ok()?.into_inner()
572 } else {
573 None
574 }
575 }
576
577
578 /// Returns `true` if the task panicked during execution.
579 ///
580 /// A task is considered panicked if it was started but the `JoinHandle` indicates
581 /// it finished without storing a result in the `OnceCell`.
582 ///
583 /// # Examples
584 ///
585 /// ```rust
586 /// use async_deferred::Deferred;
587 ///
588 /// # tokio_test::block_on(async {
589 /// let mut deferred = Deferred::start(|| async {
590 /// panic!("Something went wrong!");
591 /// });
592 ///
593 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
594 ///
595 /// assert!(deferred.has_task_panicked());
596 /// # })
597 /// ```
598 pub fn has_task_panicked(&self) -> bool {
599 matches!(self.state(), State::TaskPanicked(_))
600 }
601
602
603 /// Returns `true` if the data is available.
604 ///
605 /// # Examples
606 ///
607 /// ```rust
608 /// use async_deferred::Deferred;
609 ///
610 /// # tokio_test::block_on(async {
611 /// let mut deferred = Deferred::start(|| async {
612 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
613 /// 42
614 /// });
615 ///
616 /// // Initially not ready
617 /// assert!(!deferred.is_ready());
618 ///
619 /// deferred.join().await;
620 /// assert!(deferred.is_ready());
621 /// # })
622 /// ```
623 pub fn is_ready(&self) -> bool {
624 self.value.get().is_some()
625 }
626
627 /// Returns `true` if the task has completed successfully.
628 ///
629 /// This means the computation finished without panicking and a result is available
630 /// via [`try_get`](Self::try_get) or [`take`](Self::take).
631 ///
632 /// # Examples
633 ///
634 /// ```rust
635 /// use async_deferred::Deferred;
636 ///
637 /// # tokio_test::block_on(async {
638 /// let mut deferred = Deferred::start(|| async {
639 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
640 /// 42
641 /// });
642 ///
643 /// // Initially not ready
644 /// assert!(!deferred.is_complete());
645 ///
646 /// deferred.join().await;
647 /// println!("{:?}", deferred.state());
648 /// assert!(deferred.is_complete());
649 /// # })
650 pub fn is_complete(&self) -> bool {
651 matches!(self.state(), State::Completed)
652 }
653
654 /// Returns `true` if the task is currently running.
655 ///
656 /// A task is pending if it has been started but hasn't completed or panicked yet.
657 ///
658 /// # Examples
659 ///
660 /// ```rust
661 /// use async_deferred::Deferred;
662 /// # tokio_test::block_on(async {
663 /// let deferred = Deferred::start(|| async {
664 /// tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
665 /// 42
666 /// });
667 ///
668 /// assert!(deferred.is_pending()); // Task is running
669 /// # })
670 /// ```
671 pub fn is_pending(&self) -> bool {
672 matches!(self.state(), State::Pending)
673 }
674
675 /// Returns `true` if the task hasn't been initialized yet.
676 ///
677 /// This is the initial state of a `Deferred` created with [`new`](Self::new).
678 ///
679 /// # Examples
680 ///
681 /// ```rust
682 /// use async_deferred::Deferred;
683 /// # tokio_test::block_on(async {
684 /// let deferred: Deferred<i32> = Deferred::new();
685 /// assert!(deferred.is_not_initialized());
686 ///
687 /// let started_deferred = Deferred::start(|| async { 42 });
688 /// assert!(!started_deferred.is_not_initialized());
689 /// # })
690 /// ```
691 pub fn is_not_initialized(&self) -> bool {
692 matches!(self.state(), State::NotInitialized)
693 }
694}
695
696impl<T> Default for Deferred<T>
697where
698 T: Send + Sync + 'static,
699{
700 /// Creates a new empty `Deferred` instance.
701 ///
702 /// Equivalent to [`Deferred::new()`](Self::new).
703 fn default() -> Self {
704 Self::new()
705 }
706}