hans 0.1.0

Task orchestrator.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

use std::{future::Future, pin::Pin, sync::Arc};

use anyhow::Result;
use chrono::{DateTime, Utc};
use tokio::{spawn, sync::Mutex, task::JoinHandle};

use crate::{Executor, Task, TaskStatus};

/// A Tokio executor.
pub struct TokioExecutor;

impl Executor for TokioExecutor {}

/// An implementation of task for Tokio.
pub struct TokioTask {
    closure: Arc<Box<TokioTaskClosure>>,
    finish_status: Mutex<Option<TaskStatus>>,
    job: Mutex<Option<Job>>,
}

impl TokioTask {
    /// Creates new task.
    ///
    /// - `closure`: the closure to run.
    pub fn new<
        FN: Fn() -> FUT + Send + Sync + 'static,
        FUT: Future<Output = anyhow::Result<()>> + Send + Sync,
    >(
        closure: FN,
    ) -> Self {
        let closure = Arc::new(closure);
        Self {
            closure: Arc::new(Box::new(move || {
                let closure = closure.clone();
                Box::pin(async move { closure().await })
            })),
            finish_status: Mutex::new(None),
            job: Mutex::new(None),
        }
    }
}

impl Task<TokioExecutor> for TokioTask {
    async fn delete(&self, _exec: &TokioExecutor) -> Result<()> {
        Ok(())
    }

    async fn is_deleted(&self, _exec: &TokioExecutor) -> Result<bool> {
        Ok(true)
    }

    async fn start(&self, _exec: &TokioExecutor) -> Result<TaskStatus> {
        let closure = self.closure.clone();
        let mut job = self.job.lock().await;
        let started_at = Utc::now();
        *job = Some(Job {
            handle: spawn(async move {
                match closure().await {
                    Ok(()) => TaskStatus::Succeeded {
                        finished_at: Utc::now(),
                        started_at,
                    },
                    Err(_) => TaskStatus::Failed {
                        finished_at: Utc::now(),
                        started_at,
                    },
                }
            }),
            started_at,
        });
        Ok(TaskStatus::Running {
            started_at: Utc::now(),
        })
    }

    async fn status(&self, _exec: &TokioExecutor) -> Result<TaskStatus> {
        let mut finish_status = self.finish_status.lock().await;
        let mut guard = self.job.lock().await;
        if let Some(finish_status) = *finish_status {
            Ok(finish_status)
        } else if let Some(job) = guard.take() {
            if job.handle.is_finished() {
                let status = job.handle.await?;
                *finish_status = Some(status);
                Ok(status)
            } else {
                let started_at = job.started_at;
                *guard = Some(job);
                Ok(TaskStatus::Running { started_at })
            }
        } else {
            Ok(TaskStatus::Pending)
        }
    }
}

type TokioTaskClosure =
    dyn Fn() -> Pin<Box<dyn Future<Output = Result<()>> + Send + Sync>> + Send + Sync;

struct Job {
    handle: JoinHandle<TaskStatus>,
    started_at: DateTime<Utc>,
}