uhrwerk 0.1.0

use crate::{
	AsyncJob, Interval, Job,
	async_job::JobFuture,
	timeprovider::{DefaultTimeProvider, TimeProvider, local_offset}
};
use std::{future::Future, pin::Pin, task::Poll};
use time::UtcOffset;

/// An asynchronous job scheduler, for use with `Future`s.
///
/// The asynchronous scheduler works almost identically to the [synchronous one](crate::Scheduler), except that
/// instead of taking functions or closures returning `()`, it takes functions or closures returning values implementing `Future<Output = ()>`.
///
/// Unlike the synchronous version, there is no [`watch_thread`](crate::Scheduler::watch_thread) method, as it would tie
/// this crate to a specific runtime, and also because it's trivial to implement by hand.
/// For example, using [`smol`](https://crates.io/crates/smol):
///
/// ```no_run
/// # use uhrwerk::*;
/// # use std::time::Duration;
/// # let mut scheduler = AsyncScheduler::new();
/// smol::spawn(async move {
/// 	loop {
/// 		scheduler.run_pending().await;
/// 		smol::Timer::after(Duration::from_millis(100)).await;
/// 	}
/// });
/// ```
///
/// Or when using [`tokio`](https://crates.io/crates/tokio):
///
/// ```no_run
/// # use uhrwerk::*;
/// # use std::time::Duration;
/// # let mut scheduler = AsyncScheduler::new();
/// tokio::spawn(async move {
/// 	loop {
/// 		scheduler.run_pending().await;
/// 		tokio::time::sleep(Duration::from_millis(100)).await;
/// 	}
/// });
/// ```
///
/// ### Usage examples
///
/// The examples below are intended to demonstrate how to work with various types of Future.
/// See [synchronous examples](crate::Scheduler) for more examples of how to schedule tasks.
///
/// ```rust
/// // Scheduler, trait for .seconds(), .minutes(), etc., and trait with job scheduling methods
/// use uhrwerk::{AsyncScheduler, TimeUnits, Job};
/// // Import week days and Workday
/// use uhrwerk::Interval::*;
/// use std::time::Duration;
/// # use std::future::Future;
/// # use std::pin::Pin;
/// # async fn some_async_fn() {}
/// # fn returns_boxed_future() -> Box<dyn Future<Output=()> + Send> { Box::new(some_async_fn()) }
/// # fn returns_pinned_boxed_future() -> Pin<Box<dyn Future<Output=()> + Send>> { Box::pin(some_async_fn()) }
///
/// // Create a new scheduler
/// let mut scheduler = AsyncScheduler::new();
/// // Add some tasks to it
/// scheduler
/// 	.every(10.minutes())
/// 	.plus(30.seconds())
/// 	.run(|| async { println!("Simplest is just using an async block"); });
/// scheduler
/// 	.every(1.day())
/// 	.at("3:20 pm")
/// 	.run(|| some_async_fn());
/// scheduler
/// 	.every(Wednesday)
/// 	.at("14:20:17")
/// 	.run(some_async_fn);
/// scheduler
/// 	.every(Tuesday)
/// 	.at("14:20:17")
/// 	.and_every(Thursday)
/// 	.at("15:00")
/// 	.run(|| Pin::from(returns_boxed_future()));
/// scheduler
/// 	.every(Workday)
/// 	.run(|| returns_pinned_boxed_future());
/// scheduler
/// 	.every(1.day())
/// 	.at("3:20 pm")
/// 	.run(returns_pinned_boxed_future).once();
/// # smol::block_on(async move {
/// // Manually run the scheduler forever
/// loop {
/// 	scheduler.run_pending().await;
/// 	smol::Timer::after(Duration::from_millis(10)).await;
/// 	# break;
/// }
///
/// // Or spawn a task to run it forever
/// smol::spawn(async move {
/// 	loop {
/// 		scheduler.run_pending().await;
/// 		smol::Timer::after(Duration::from_millis(100)).await;
/// 	}
/// });
/// # });
/// ```
#[derive(Debug)]
pub struct AsyncScheduler<TP = DefaultTimeProvider> {
	jobs: Vec<AsyncJob<TP>>,
	utc_offset: UtcOffset
}

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

impl AsyncScheduler {
	/// 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> AsyncScheduler<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::*;
	/// # use std::future::Future;
	/// # use std::pin::Pin;
	/// # async fn some_async_fn() {}
	/// # fn returns_boxed_future() -> Box<dyn Future<Output=()> + Send> { Box::new(some_async_fn()) }
	/// # fn returns_pinned_boxed_future() -> Pin<Box<dyn Future<Output=()> + Send>> { Box::pin(some_async_fn()) }
	/// let mut scheduler = AsyncScheduler::new();
	/// scheduler
	/// 	.every(10.minutes())
	/// 	.plus(30.seconds())
	/// 	.run(|| async { println!("Periodic task") });
	/// scheduler
	/// 	.every(1.day())
	/// 	.at("3:20 pm")
	/// 	.run(|| some_async_fn());
	/// scheduler
	/// 	.every(Wednesday)
	/// 	.at("14:20:17")
	/// 	.run(|| Pin::from(returns_boxed_future()));
	/// scheduler
	/// 	.every(Workday)
	/// 	.run(|| returns_pinned_boxed_future());
	/// ```
	pub fn every(&mut self, ival: Interval) -> &mut AsyncJob<TP> {
		let job = AsyncJob::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 returns a future that will poll each of the tasks until they are completed.
	///
	/// ```no_run
	/// # use uhrwerk::*;
	/// # use uhrwerk::Interval::*;
	/// use std::time::Duration;
	/// # let mut scheduler = AsyncScheduler::new();
	/// # async {
	/// loop {
	/// 	scheduler.run_pending().await;
	/// 	smol::Timer::after(Duration::from_millis(100)).await;
	/// 	# break
	/// }
	/// # };
	/// ```
	///
	/// Note that while all pending jobs will run asynchronously, a long-running task can still
	/// block future executions if you `await` the future returned by this method.
	/// If you are concerned that a task might run for a long time, there are several possible approaches:
	///
	///  1. Pass the result of `scheduler.run_pending()` to your runtime's `spawn` function. This might
	///     result in multiple invocations of the same task running concurrently.
	///  2. Use `spawn` or `spawn_blocking` in your task itself. This has the same concurrent execution risk
	///     as approach 1, but limited to that specific task.
	///  3. Use `tokio::time::timeout` or equivalent to prevent `scheduler.run_pending()` or the task itself
	///     from running more than an expected amount of time. E.g.
	///
	///     ```no_run
	///     # use uhrwerk::*;
	///     # use uhrwerk::Interval::*;
	///     # async fn scrape_pages() {}
	///     use std::time::Duration;
	///     let mut scheduler = AsyncScheduler::new();
	///     scheduler.every(10.minutes()).run(|| async {
	///     	if let Err(_) =
	///     		tokio::time::timeout(Duration::from_secs(10 * 60), scrape_pages()).await
	///     	{
	///     		eprintln!("Timed out scraping pages")
	///     	}
	///     });
	///     ```
	pub fn run_pending(&mut self) -> AsyncSchedulerFuture {
		let now = TP::now(self.utc_offset);
		let mut futures = vec![];
		for job in &mut self.jobs {
			if job.is_pending(now) {
				if let Some(future) = job.execute(now) {
					futures.push(Some(future));
				}
			}
		}
		AsyncSchedulerFuture { futures }
	}
}

pub struct AsyncSchedulerFuture {
	futures: Vec<Option<JobFuture>>
}

impl Future for AsyncSchedulerFuture {
	type Output = ();

	fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
		let mut all_done = true;

		for future in &mut self.get_mut().futures {
			if let Some(this_future) = future {
				if this_future.as_mut().poll(cx) == Poll::Ready(()) {
					future.take();
				} else {
					all_done = false;
				}
			}
		}
		if all_done {
			Poll::Ready(())
		} else {
			Poll::Pending
		}
	}
}