shinyframework_jobs 0.1.2

Shiny Jobs
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

use std::sync::Arc;
use crate::Job;
use shiny_common::context::{ContextFactory};
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use tokio_util::sync::CancellationToken;
use tracing::{Instrument};
use uuid::Uuid;
use crate::JobDefinition;
use crate::job_trigger::JobTrigger;

pub struct JobsExecutor {
    jobs: Vec<JobDefinition>,
    context_factory: Arc<dyn ContextFactory>,
}

impl JobsExecutor {
    pub fn new(
        context_factory: Arc<dyn ContextFactory>
    ) -> JobsExecutor {
        JobsExecutor {
            jobs: Vec::new(),
            context_factory,
        }
    }

    pub fn schedule(&mut self, name: String, job: Arc<dyn Job>, trigger: Arc<dyn JobTrigger>) {
        self.jobs.push(JobDefinition {
            name,
            job,
            trigger,
        });
    }

    pub async fn run(&self, cancellation_token: CancellationToken) {
        let loops = self.jobs.iter().map(|definition| {
            let trigger = definition.trigger.clone();
            let job = definition.job.clone();
            let cancellation_token = cancellation_token.child_token();
            let context_factory = self.context_factory.clone();

            async move {
                loop {
                    let task_id = Uuid::new_v4();
                    let span = tracing::span!(tracing::Level::INFO, "Task", id=?task_id);

                    let execution_flow = async {
                        tracing::debug!("awaiting trigger fire");
                        tokio::select! {
                            _ = cancellation_token.cancelled() => {
                                tracing::debug!("cancellation signal received");
                                return ExecutionFlow::Stop;
                            }
                            _ = trigger.next() => {
                                tracing::debug!("starting execution of job");
                            }
                        }

                        let child_token = cancellation_token.child_token();
                        let local_job = job.clone();
                        let context_factory = context_factory.clone();

                        tokio::spawn(async move {
                            let mut context = context_factory.create();
                            context.set_cancellation_token(child_token);

                            match local_job.execute(&mut context).await {
                                Ok(_) => tracing::info!("Job successfully executed"),
                                Err(err) => tracing::error!("Job failed {}", err)
                            }
                        }.in_current_span());

                        ExecutionFlow::Continue
                    }.instrument(span).await;

                    if let ExecutionFlow::Stop = execution_flow {
                        return;
                    }
                }
            }.instrument(tracing::span!(tracing::Level::INFO, "Job", name=definition.name))
        });

        let mut loops = FuturesUnordered::from_iter(loops);

        if loops.next().await.is_some() && cancellation_token.is_cancelled() {
            loops.collect().await
        }
    }
}


enum ExecutionFlow {
    Continue,
    Stop
}