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

extern crate config;
extern crate riker;
extern crate uuid;

use std::time::{Duration, SystemTime};
use std::thread;
use std::sync::mpsc::{channel, Sender};

use config::Config;
use uuid::Uuid;

use riker::protocol::Message;
use riker::system::{TimerFactory, Job, OnceJob, RepeatJob};

impl<Msg: Message> TimerFactory for BasicTimer<Msg> {
    type Msg = Msg;

    fn new(config: &Config, _debug: bool) -> Sender<Job<Msg>> {
        let config = BasicTimerConfig::from(config);
        let tx = BasicTimer::start(config);

        tx
    }
}

pub struct BasicTimer<Msg: Message> {
    once_jobs: Vec<OnceJob<Msg>>,
    repeat_jobs: Vec<RepeatJob<Msg>>,
}

impl<Msg: Message> BasicTimer<Msg> {
    fn start(config: BasicTimerConfig) -> Sender<Job<Msg>> {
        let mut process = BasicTimer {
            once_jobs: Vec::new(),
            repeat_jobs: Vec::new()
        };

        let (tx, rx) = channel();
        thread::spawn(move || {
            loop {
                process.execute_once_jobs();
                process.execute_repeat_jobs();

                if let Ok(job) = rx.try_recv() {
                    match job {
                        Job::Cancel(id) => process.cancel(&id),
                        Job::Once(job) => process.schedule_once(job),
                        Job::Repeat(job) => process.schedule_repeat(job)
                    }
                }

                thread::sleep(Duration::from_millis(config.frequency_millis));
            }
        });

        tx
    }

    fn execute_once_jobs(&mut self) {
        let (send, keep): (Vec<OnceJob<Msg>>, Vec<OnceJob<Msg>>) =
            self.once_jobs.drain(..).partition(|j| SystemTime::now() >= j.send_at);

        // send those messages where the 'send_at' time has been reached or elapsed
        for job in send.into_iter() {
            job.send();
        }

        // for those messages that are not to be sent yet, just put them back on the vec
        for job in keep.into_iter() {
            self.once_jobs.push(job);
        }
    }

    fn execute_repeat_jobs(&mut self) {
        for job in self.repeat_jobs.iter_mut() {
            if SystemTime::now() >= job.send_at {
                job.send_at = SystemTime::now() + job.interval;
                job.send();
            }
        }
    }

    fn cancel(&mut self, id: &Uuid) {
        // slightly sub optimal way of canceling because we don't know the job type
        // so need to do the remove on both collections
        
        if let Some(pos) = self.once_jobs.iter().position(|job| &job.id == id) {
            self.once_jobs.remove(pos);
        }

        if let Some(pos) = self.repeat_jobs.iter().position(|job| &job.id == id) {
            self.repeat_jobs.remove(pos);
        }
    }

    fn schedule_once(&mut self, job: OnceJob<Msg>) {
        if SystemTime::now() >= job.send_at {
            job.send();
        } else {
            self.once_jobs.push(job);
        }
    }

    fn schedule_repeat(&mut self, job: RepeatJob<Msg>) {
        if SystemTime::now() >= job.send_at {
            job.send();
        } else {
            self.repeat_jobs.push(job);
        }
    }
}

struct BasicTimerConfig {
    frequency_millis: u64,
}

impl<'a> From<&'a Config> for BasicTimerConfig {
    fn from(config: &Config) -> Self {
        BasicTimerConfig {
            frequency_millis: config.get_int("scheduler.frequency_millis").unwrap() as u64
        }
    }
}