uhrwerk 0.1.0

use crate::{
	Interval, Job, SyncJob,
	timeprovider::{DefaultTimeProvider, TimeProvider, local_offset}
};
use std::{
	default::Default,
	sync::{
		Arc,
		atomic::{AtomicBool, Ordering}
	},
	thread,
	time::Duration
};
use time::UtcOffset;

/// Synchronous job scheduler
///
/// ### Usage examples
///
/// ```rust
/// // Scheduler, trait for .seconds(), .minutes(), etc., and trait with job scheduling methods
/// use uhrwerk::{Scheduler, TimeUnits, Job};
/// // Import week days and Workday
/// use uhrwerk::Interval::*;
/// # use std::thread;
/// # use std::time::Duration;
/// # use time::{Time, UtcOffset};
///
/// // Create a new scheduler
/// let mut scheduler = Scheduler::new();
/// // or a scheduler with a given timezone
/// let mut scheduler = Scheduler::new_with_utc_offset(UtcOffset::UTC);
/// // Add some tasks to it
/// scheduler
/// 	.every(10.minutes())
/// 	.plus(30.seconds())
/// 	.run(|| println!("Periodic task"));
/// scheduler
/// 	.every(1.day())
/// 	.at("3:20 pm")
/// 	.run(|| println!("Daily task"));
/// scheduler
/// 	.every(Wednesday)
/// 	.at("14:20:17")
/// 	.run(|| println!("Weekly task"));
/// scheduler
/// 	.every(Tuesday)
/// 	.at("14:20:17")
/// 	.and_every(Thursday)
/// 	.at("15:00")
/// 	.run(|| println!("Biweekly task"));
/// scheduler
/// 	.every(Workday)
/// 	.run(|| println!("Every work day at midnight"));
/// scheduler
/// 	.every(1.day())
/// 	.at("3:20 pm")
/// 	.run(|| println!("I only run once")).once();
/// scheduler
/// 	.every(Workday)
/// 	.at("12:00").count(10)
/// 	.run(|| println!("Countdown"));
/// scheduler
/// 	.every(1.day())
/// 	.at("10:00 am")
/// 	.repeating_every(30.minutes())
/// 	.times(6)
/// 	.run(|| println!("I run every half hour from 10 AM to 1 PM inclusive."));
/// scheduler
/// 	.every(1.day())
/// 	.at_time(Time::from_hms(13, 12, 14).unwrap())
/// 	.run(|| println!("You can also pass time::Time to `at_time`."));
///
/// // Manually run the scheduler in an event loop
/// for _ in 1..10 {
/// 	scheduler.run_pending();
/// 	thread::sleep(Duration::from_millis(10));
/// 	# break;
/// }
///
/// // Or run it in a background thread
/// let thread_handle = scheduler.watch_thread(Duration::from_millis(100));
/// // The scheduler stops when `thread_handle` is dropped, or `stop` is called
/// thread_handle.stop();
/// ```
#[derive(Debug)]
pub struct Scheduler<TP = DefaultTimeProvider> {
	jobs: Vec<SyncJob<TP>>,
	utc_offset: UtcOffset
}

impl Default for Scheduler {
	fn default() -> Self {
		Self::new()
	}
}

impl Scheduler {
	/// Create a new scheduler. Dates and times will be interpretted using the local timezone
	pub fn new() -> Self {
		Self::new_with_utc_offset(local_offset())
	}

	/// Create a new scheduler. Dates and times will be interpretted using the specified timezone.
	pub const fn new_with_utc_offset(utc_offset: UtcOffset) -> Self {
		Self::new_with_provider(utc_offset)
	}
}

impl<TP: TimeProvider> Scheduler<TP> {
	/// Create a new scheduler. Dates and times will be interpretted using the specified timezone.
	/// In addition, you can provide an alternate time provider. This is mostly useful for writing
	/// tests.
	pub const fn new_with_provider(utc_offset: UtcOffset) -> Self {
		Self {
			jobs: Vec::new(),
			utc_offset
		}
	}

	/// Add a new job to the scheduler to be run on the given interval
	///
	/// ```rust
	/// # use uhrwerk::*;
	/// # use uhrwerk::Interval::*;
	/// let mut scheduler = Scheduler::new();
	/// scheduler
	/// 	.every(10.minutes())
	/// 	.plus(30.seconds())
	/// 	.run(|| println!("Periodic task"));
	/// scheduler
	/// 	.every(1.day())
	/// 	.at("3:20 pm")
	/// 	.run(|| println!("Daily task"));
	/// scheduler
	/// 	.every(Wednesday)
	/// 	.at("14:20:17")
	/// 	.run(|| println!("Weekly task"));
	/// scheduler
	/// 	.every(Workday)
	/// 	.run(|| println!("Every work day at midnight"));
	/// ```
	pub fn every(&mut self, ival: Interval) -> &mut SyncJob<TP> {
		let job = SyncJob::new(ival, self.utc_offset);
		self.jobs.push(job);
		let last_index = self.jobs.len() - 1;
		&mut self.jobs[last_index]
	}

	/// Run all jobs that should run at this time.
	///
	/// This method blocks while jobs are being run. If a job takes a long time, it may prevent
	/// other tasks from running as scheduled. If you have a long-running task, you might consider
	/// having the job move the work into another thread so that it can return promptly.
	///
	/// ```rust
	/// # use uhrwerk::*;
	/// # use uhrwerk::Interval::*;
	/// use std::{thread, time::Duration};
	/// # let mut scheduler = Scheduler::new();
	/// loop {
	/// 	scheduler.run_pending();
	/// 	thread::sleep(Duration::from_millis(100));
	/// 	break;
	/// }
	/// ```
	pub fn run_pending(&mut self) {
		let now = TP::now(self.utc_offset);
		for job in &mut self.jobs {
			if job.is_pending(now) {
				job.execute(now);
			}
		}
	}
}

impl Scheduler {
	/// Start a background thread to call [Scheduler::run_pending()] repeatedly.
	///
	/// The frequency argument controls how long the thread will sleep between calls
	/// to [Scheduler::run_pending()].
	/// If the returned [ScheduleHandle] is dropped, the resulting thread will end
	/// cleanly when [Scheduler::run_pending()] would have next been called.
	///
	/// Passing large durations for `frequency` can cause long delays when [ScheduleHandle::stop()]
	/// is called, or the [ScheduleHandle] is dropped, as it waits for the thread to finish sleeping.
	/// This could affect how long it takes for the program to exit.
	///
	/// Reasonable values for `frequency` would be between 100 ms and 10 seconds.
	/// If in doubt, choose a smaller value.
	#[must_use = "The scheduler is halted when the returned handle is dropped"]
	pub fn watch_thread(self, frequency: Duration) -> ScheduleHandle {
		let stop = Arc::new(AtomicBool::new(false));
		let my_stop = stop.clone();
		let mut me = self;
		let handle = thread::spawn(move || {
			while !stop.load(Ordering::SeqCst) {
				me.run_pending();
				thread::sleep(frequency);
			}
		});
		ScheduleHandle {
			stop: my_stop,
			thread_handle: Some(handle)
		}
	}
}

/// Guard object for the scheduler background thread. The thread is terminated if this object
/// is dropped, or [ScheduleHandle::stop()] is called
pub struct ScheduleHandle {
	stop: Arc<AtomicBool>,
	thread_handle: Option<thread::JoinHandle<()>>
}

impl ScheduleHandle {
	/// Halt the scheduler background thread
	pub fn stop(self) {}
}

impl Drop for ScheduleHandle {
	fn drop(&mut self) {
		self.stop.store(true, Ordering::SeqCst);
		let handle = self.thread_handle.take();
		handle.unwrap().join().ok();
	}
}

#[cfg(test)]
mod tests {
	use super::{Job, Scheduler, TimeProvider};
	use crate::intervals::*;
	use std::sync::{
		Arc,
		atomic::{AtomicU32, Ordering}
	};
	use time::{OffsetDateTime, UtcOffset, format_description::well_known::Rfc3339};

	macro_rules! make_time_provider {
		($name:ident : $($time:literal),+) => {
			#[derive(Debug)]
			struct $name;

			static TIMES_TIME_REQUESTED: AtomicU32 = AtomicU32::new(0);

			impl TimeProvider for $name {
				fn now(utc_offset: UtcOffset) -> OffsetDateTime {
					static TIMES: &[&str] = &[$($time),+];
					let idx = TIMES_TIME_REQUESTED.fetch_add(1, Ordering::SeqCst) as usize;
					OffsetDateTime::parse(TIMES[idx], &Rfc3339).unwrap().to_offset(utc_offset)
				}
			}
		};
	}

	#[test]
	fn test_every_plus() {
		make_time_provider!(FakeTimeProvider :
			"2019-10-22T12:40:00Z",
			"2019-10-22T12:40:00Z",
			"2019-10-22T12:50:20Z",
			"2019-10-22T12:50:30Z"
		);
		let mut scheduler =
			Scheduler::<FakeTimeProvider>::new_with_provider(UtcOffset::UTC);
		let times_called = Arc::new(AtomicU32::new(0));
		{
			let times_called = times_called.clone();
			scheduler
				.every(10.minutes())
				.plus(5.seconds())
				.run(move || {
					times_called.fetch_add(1, Ordering::SeqCst);
				});
		}
		assert_eq!(1, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(0, times_called.load(Ordering::SeqCst));
		assert_eq!(2, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		assert_eq!(3, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		assert_eq!(4, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
	}

	#[test]
	fn test_every_at() {
		make_time_provider!(FakeTimeProvider:
			"2019-10-22T12:40:00Z",
			"2019-10-22T12:40:10Z",
			"2019-10-25T12:50:20Z",
			"2019-10-25T15:23:30Z",
			"2019-10-26T15:50:30Z"
		);
		let mut scheduler =
			Scheduler::<FakeTimeProvider>::new_with_provider(UtcOffset::UTC);
		let times_called = Arc::new(AtomicU32::new(0));
		{
			let times_called = times_called.clone();
			scheduler.every(3.days()).at("15:23").run(move || {
				times_called.fetch_add(1, Ordering::SeqCst);
			});
		}
		assert_eq!(1, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(0, times_called.load(Ordering::SeqCst));
		assert_eq!(2, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		assert_eq!(3, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		assert_eq!(4, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
	}

	#[test]
	fn test_every_and_every() {
		make_time_provider!(FakeTimeProvider:
			"2019-10-22T12:40:01Z",
			"2019-10-22T12:40:01Z",
			"2019-10-22T12:40:02Z",
			"2019-10-22T12:40:03Z",
			"2019-10-22T12:40:04Z",
			"2019-10-22T12:40:05Z",
			"2019-10-22T12:40:06Z"
		);
		let mut scheduler =
			Scheduler::<FakeTimeProvider>::new_with_provider(UtcOffset::UTC);
		let times_called = Arc::new(AtomicU32::new(0));
		{
			let times_called = times_called.clone();
			scheduler
				.every(5.seconds())
				.and_every(2.seconds())
				.run(move || {
					times_called.fetch_add(1, Ordering::SeqCst);
				});
		}
		assert_eq!(1, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(2, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(0, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(3, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(4, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(5, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(2, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(6, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(3, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(7, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(4, times_called.load(Ordering::SeqCst));
	}

	#[test]
	fn test_once() {
		make_time_provider!(FakeTimeProvider:
			"2019-10-22T12:40:01Z",
			"2019-10-22T12:40:01Z",
			"2019-10-22T12:40:02Z",
			"2019-10-22T12:40:03Z"
		);
		let mut scheduler =
			Scheduler::<FakeTimeProvider>::new_with_provider(UtcOffset::UTC);
		let times_called = Arc::new(AtomicU32::new(0));
		{
			let times_called = times_called.clone();
			scheduler.every(1.seconds()).once().run(move || {
				times_called.fetch_add(1, Ordering::SeqCst);
			});
		}
		assert_eq!(1, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(2, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(0, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(3, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(1, times_called.load(Ordering::SeqCst));
		scheduler.run_pending();
		assert_eq!(4, TIMES_TIME_REQUESTED.load(Ordering::SeqCst));
		assert_eq!(1, times_called.load(Ordering::SeqCst));
	}
}