librqbit_core/
spawn_utils.rs

1use std::{borrow::Cow, fmt::Display};
2
3use tokio_util::sync::CancellationToken;
4use tracing::{Instrument, debug, error, trace};
5
6/// Spawns a future with tracing instrumentation.
7#[track_caller]
8pub fn spawn<E: Display + Send + 'static>(
9    span: tracing::Span,
10    name: impl Into<Cow<'static, str>>,
11    fut: impl std::future::Future<Output = Result<(), E>> + Send + 'static,
12) -> tokio::task::JoinHandle<()> {
13    let name = name.into();
14    let fut = async move {
15        trace!("started");
16        tokio::pin!(fut);
17        let mut trace_interval = tokio::time::interval(std::time::Duration::from_secs(5));
18
19        loop {
20            tokio::select! {
21                _ = trace_interval.tick() => {
22                    trace!("still running");
23                },
24                r = &mut fut => {
25                    match r {
26                        Ok(_) => {
27                            trace!("finished");
28                        }
29                        Err(e) => {
30                            error!("{name} finished with error: {:#}", e)
31                        }
32                    }
33                    return;
34                }
35            }
36        }
37    }
38    .instrument(span);
39    tokio::task::spawn(fut)
40}
41
42#[track_caller]
43pub fn spawn_with_cancel<E: Display + Send + 'static>(
44    span: tracing::Span,
45    name: impl Into<Cow<'static, str>>,
46    cancellation_token: CancellationToken,
47    fut: impl std::future::Future<Output = Result<(), E>> + Send + 'static,
48) -> tokio::task::JoinHandle<()> {
49    spawn(span, name, async move {
50        tokio::select! {
51            _ = cancellation_token.cancelled() => {
52                debug!("task cancelled");
53                Ok(())
54            },
55            r = fut => r
56        }
57    })
58}
librqbit_core/spawn_utils.rs

librqbit_core/
spawn_utils.rs
