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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! Tasks that come separated by time pauses.
//! A task can also be forced to be started at any time, but only during a pause.
//! If a task is forced to be started, the schedule of pauses modifies to accommodate this task.
//!
//! This code is maybe more a demo than a serious module.

use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use tokio::spawn;
use tokio::sync::{mpsc, Mutex};
use tokio::sync::mpsc::Sender;
use tokio::task::JoinHandle;
use tokio::time::timeout;
use async_trait::async_trait;
use tokio_interruptible_future::{InterruptError, interruptible_sendable};

#[allow(dead_code)]
#[derive(Clone)]
pub struct TasksWithRegularPausesData {
    sudden_tx: Arc<Mutex<Option<Sender<()>>>>,
}

impl TasksWithRegularPausesData {
    #[allow(dead_code)]
    pub fn new() -> Self {
        Self {
            sudden_tx: Arc::new(Mutex::new(None)),
        }
    }
}

/// See module documentation.
#[async_trait]
pub trait TasksWithRegularPauses<Task: Future<Output = ()> + Send>: Send + Sync + 'static {
    fn data(&self) -> &TasksWithRegularPausesData;
    fn data_mut(&mut self) -> &mut TasksWithRegularPausesData;
    async fn next_task(&self) -> Option<Task>;
    fn sleep_duration(&self) -> Duration;
    async fn _task(&mut self) -> Result<(), InterruptError> { // `InterruptError` here is a hack.
        loop {
            // It is time to run a task.
            // let this1 = this.lock().await;
            let fut = self.next_task().await;
            if let Some(fut) = fut {
                fut.await;
            } else {
                break;
            }

            // Signal that may interrupt the task.
            let (sudden_tx, mut sudden_rx) = mpsc::channel(1);
            self.data_mut().sudden_tx = Arc::new(Mutex::new(Some(sudden_tx)));

            // Re-execute by a signal, or timeout (whichever comes first)
            let sleep_duration = self.sleep_duration(); // lock for one line
            let _ = timeout(sleep_duration, sudden_rx.recv()).await;
        }
        Ok(())
    }
    fn spawn(&'static mut self, interrupt_notifier: async_channel::Receiver<()>) -> JoinHandle<Result<(), InterruptError>> {
        spawn( interruptible_sendable(interrupt_notifier, Box::pin(Self::_task(self))))
    }
    async fn suddenly(&self) -> Result<(), tokio::sync::mpsc::error::TrySendError<()>>{
        let sudden_tx = self.data().sudden_tx.lock().await.take(); // atomic operation
        if let Some(sudden_tx) = sudden_tx {
            sudden_tx.try_send(())?;
        }
        Ok(())
    }
}

/// Object-safe variation of TaskQueue
// pub struct ObjectSafeTasksWithRegularPauses<Task: Future<Output = ()> + Send>: Send + Sync + 'static {
//     base: Arc<Mutex<TasksWithRegularPauses<Task>>>,
// }
//
// impl<Task: Future<Output = ()> + Send> ObjectSafeTasksWithRegularPauses<Task> {
//     pub fn new() -> Self {
//         Self {
//             base: Arc::new(Mutex::new(TasksWithRegularPauses::new())),
//         }
//     }
//     pub async fn get_arc(&self) -> &Arc<Mutex<TasksWithRegularPauses>> {
//         &self.base
//     }
//     pub async fn get_arc_mut(&mut self) -> &Arc<Mutex<TasksWithRegularPauses>> {
//         &mut self.base
//     }
// }

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;
    use async_channel::bounded;
    use tokio::sync::Mutex;
    use async_trait::async_trait;
    use tokio::runtime::Runtime;
    use tokio_interruptible_future::InterruptError;
    use crate::TaskItem;
    use crate::tasks_with_regular_pauses::{TasksWithRegularPauses, TasksWithRegularPausesData};

    #[derive(Clone)]
    struct OurTaskQueue {
        data: TasksWithRegularPausesData,
    }

    impl OurTaskQueue {
        pub fn new() -> Self {
            Self {
                data: TasksWithRegularPausesData::new(),
            }
        }
    }

    #[async_trait]
    impl<'a> TasksWithRegularPauses<TaskItem> for OurTaskQueue where Self: 'static {
        fn data(&self) -> &TasksWithRegularPausesData {
            &self.data
        }
        fn data_mut(&mut self) -> &mut TasksWithRegularPausesData {
            &mut self.data
        }
        async fn next_task(&self) -> Option<TaskItem> {
            Some(Box::pin(async { () }))
        }
        fn sleep_duration(&self) -> Duration {
            Duration::from_millis(1)
        }
    }

    #[test]
    fn empty() -> Result<(), InterruptError> {
        let queue = OurTaskQueue::new();
        let (interrupt_notifier_tx, interrupt_notifier_rx) = bounded(1);
        let rt  = Runtime::new().unwrap();
        rt.block_on(async {
            OurTaskQueue::spawn(queue.clone(), interrupt_notifier_rx);
            let _ = interrupt_notifier_tx.send(()).await;
            queue.clone().suddenly().await.unwrap();
        });
        Ok(())
    }
}