ora 0.12.7

Part of the Ora scheduler framework.
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

use std::{pin::pin, sync::Arc};

use tokio::spawn;
use wgroup::{WaitGroup, WaitGroupHandle};

use crate::executor::implementation::{
    Executor, ExecutorHandle, ExecutorJobQueue, connection::connection_loop,
    executions::executor_loop,
};

impl<T> Executor<T>
where
    T: tonic::client::GrpcService<tonic::body::Body> + Send + 'static,
    T::Error: Into<tonic::codegen::StdError>,
    T::ResponseBody:
        tonic::codegen::Body<Data = tonic::codegen::Bytes> + std::marker::Send + 'static,
    <T::ResponseBody as tonic::codegen::Body>::Error:
        Into<tonic::codegen::StdError> + std::marker::Send,
    <T as tonic::client::GrpcService<tonic::body::Body>>::Future: std::marker::Send,
{
    /// Spawn the executor on the tokio runtime.
    ///
    /// The executor will run until it is stopped
    /// and will attempt to handle all errors, including
    /// reconnection in case the connection is lost with the server.
    pub fn spawn(self) -> ExecutorHandle {
        self.spawn_impl(None)
    }

    /// Spawn the executor on the tokio runtime.
    ///
    /// The executor will run until it is stopped
    /// and will attempt to handle all errors, including
    /// reconnection in case the connection is lost with the server.
    ///
    /// The executor will use the provided wait group for shutdown.
    pub fn spawn_with_wg(self, wg_handle: WaitGroupHandle) {
        _ = self.spawn_impl(Some(wg_handle));
    }

    fn spawn_impl(self, wg_handle: Option<WaitGroupHandle>) -> ExecutorHandle {
        let wg = if wg_handle.is_none() {
            Some(WaitGroup::new())
        } else {
            None
        };

        let executor_wg = match (wg_handle, &wg) {
            (Some(handle), None) => handle.add_with("executor"),
            (None, Some(wg)) => wg.add_with("executor"),
            _ => unreachable!(),
        };
        let running_executions_group = WaitGroup::new();

        spawn(async move {
            let mut client = self.client;
            let queues = Arc::<[ExecutorJobQueue]>::from(self.queues);
            let executor_name = Arc::<str>::from(self.options.name.as_str());

            loop {
                let (out_send, out_recv) = flume::unbounded();
                let (in_send, in_recv) = flume::unbounded();

                spawn(executor_loop(
                    in_recv,
                    out_send,
                    executor_name.clone(),
                    queues.clone(),
                    self.options.cancellation_grace_period,
                    self.on_execution_failed.clone(),
                    running_executions_group.add(),
                ));

                let mut conn_loop = pin!(connection_loop(&mut client, in_send, out_recv));

                tokio::select! {
                    _ = &mut conn_loop => {},
                    _ = executor_wg.waiting() => {
                        tracing::info!("executor is shutting down...");
                        running_executions_group.all_done().await;
                        return;
                    }
                }

                tracing::warn!("reconnecting in 5 seconds...");
                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
            }
        });

        ExecutorHandle { _wg: wg }
    }
}