/* ************************************************************************
#
#  DivConq
#
#  http://divconq.com/
#
#  Copyright:
#    Copyright 2012 eTimeline, LLC. All rights reserved.
#
#  License:
#    See the license.txt file in the project's top-level directory for details.
#
#  Authors:
#    * Andy White
#
************************************************************************ */
package divconq.scheduler;

import java.util.HashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.locks.ReentrantLock;

import org.joda.time.DateTimeUtils;
import org.joda.time.LocalDate;
import org.joda.time.LocalDateTime;
import org.joda.time.LocalTime;
import org.joda.time.Period;
import org.joda.time.ReadableInstant;

import divconq.lang.Hub;
import divconq.lang.TaskContext;
import divconq.scheduler.limit.LimitHelper;
import divconq.util.StringUtil;
import divconq.work.IRunnable;
import divconq.work.ITaskInfo;
import divconq.work.WorkPool;
import divconq.xml.XElement;

/**
 * Handles scheduling application tasks.
 * 
 * @author Andy White
 *
 */
// TODO add tracing settings
public class Scheduler {
	// do not queue for running more that this # per cycle of the scheduler loop
	// this limit reduces the issue of a large number of items all scheduled to 
	// happen at one time (which might steal too much time from the thread pool)
	static public final int MAX_PER_CYCLE = 10;
	
	// the first node in the list of scheduled nodes - the head of the list is moved
	// forward as the items on the list get scheduled.  List is single linked list.
	protected SchedulerNode first = null;

	// how many are currently in the linked list?
	protected long nodeCnt = 0;
	
	// lock during adding and removing of scheduled work.  all add and remove operations
	// are thread safe
	protected ReentrantLock lock = new ReentrantLock();
	
	protected ScheduledFuture<?> clock = null;
	
	protected HashMap<String,LimitHelper> batches = new HashMap<String,LimitHelper>();
	
	public LimitHelper getBatch(String name) {
		return this.batches.get(name);
	}
	
	public void init(XElement config) {
		// TODO

		if (config != null) {
			for (XElement el : config.selectAll("Batch")) {
				XElement bel = el.find("Limits");
				String name = el.getAttribute("Name");
				
				if (StringUtil.isNotEmpty(name) && (bel != null)) {
					LimitHelper h = new LimitHelper();
					h.init(bel);
					this.batches.put(name, h);
				}				
			}
		}
	}
	
	public void start() {
		int scheduleFreq = 1;	// once every second - TODO config
		
		// we check our app schedule ever 1 (or so) system seconds
		this.clock = Hub.instance.getClock().schedulePeriodicInternal(new Runnable() {
			@Override
			public void run() {
				Scheduler.this.execute();
			}
		}, scheduleFreq);
	}
	
	public void stop() {
		// TODO
		
		this.clock.cancel(false);
	}

	public long size() {
		return this.nodeCnt;
	}

	// the scheduler runs on its own thread, this is the code that starts and runs the scheduler
	// to start running the scheduler you should call Resume like you do with any TThread
	private void execute() {
		this.lock.lock();

		// TODO check work pool - if it is too busy then (75% ?) then skip this time
		
		try {
			SchedulerNode curr = this.first;

			// max tasks to queue per run of watcher
			//long maxqueue = Scheduler.MAX_PER_CYCLE;
			int maxpercent = 75;  // TODO configure
			long now = DateTimeUtils.currentTimeMillis();
			WorkPool p = Hub.instance.getWorkPool(); 

			//while ((maxqueue > 0) && (curr != null) && (curr.when <= now)) {
			while ((p.howBusy() <= maxpercent) && (curr != null) && (curr.when <= now)) {
				p.submit(curr);
				//maxqueue--;
				curr = curr.next;
				this.first = curr;
				this.nodeCnt--;
			}
		}
		catch(Exception x) {
			// TODO trace/log
		}		
		finally {
			this.lock.unlock();
		}
	}

	// add a work unit to run (almost) immediately - much the same as directly adding to the thread pool
	// any work unit submitted to the scheduler (or to any thread pool) will become owned by the
	// scheduler (or thread pool).
    public SchedulerNode runNow(IRunnable work) {
    	return this.addNode(new SimpleSchedule(work, DateTimeUtils.currentTimeMillis(), 0));
    }

	// run the work unit once in Sec seconds from now
	public SchedulerNode runIn(IRunnable work, int secs) {
		return this.addNode(new SimpleSchedule(work, DateTimeUtils.currentTimeMillis() + (1000 * secs), 0));		
	}

	// run the work unit once at the specified time.  If less than now then submits immediately
	// if in the distant future, it may not be run if the process is terminated.  adding working
	// to the schedule is no guarantee work will be run.
	public SchedulerNode runAt(IRunnable work, ReadableInstant time) {
		return this.addNode(new SimpleSchedule(work, time.getMillis(), 0));				
	}
	
	public SchedulerNode runAt(IRunnable work, LocalDate date, Period period) {
		LocalDateTime ldt = date.toLocalDateTime(new LocalTime(0, 0).plus(period));
		return this.runAt(work, ldt.toDateTime());
	}

	// run the work unit repeatedly, every Secs seconds - note scheduler will not own work, you have to keep track of it
	public SchedulerNode runEvery(IRunnable work, int secs) {
		return this.addNode(new SimpleSchedule(work, DateTimeUtils.currentTimeMillis() + (1000 * secs), secs));		
	}

	// add a work unit to run (almost) immediately - much the same as directly adding to the thread pool
	// any work unit submitted to the scheduler (or to any thread pool) will become owned by the
	// scheduler (or thread pool).
    public SchedulerNode runNow(ITaskInfo work) {
    	return this.addNode(new SimpleSchedule(work, DateTimeUtils.currentTimeMillis(), 0));
    }

	// run the work unit once in Sec seconds from now
	public SchedulerNode runIn(ITaskInfo work, int secs) {
		return this.addNode(new SimpleSchedule(work, DateTimeUtils.currentTimeMillis() + (1000 * secs), 0));		
	}

	// run the work unit once at the specified time.  If less than now then submits immediately
	// if in the distant future, it may not be run if the process is terminated.  adding working
	// to the schedule is no guarantee work will be run.
	public SchedulerNode runAt(ITaskInfo work, ReadableInstant time) {
		return this.addNode(new SimpleSchedule(work, time.getMillis(), 0));				
	}
	
	public SchedulerNode runAt(ITaskInfo work, LocalDate date, Period period) {
		LocalDateTime ldt = date.toLocalDateTime(new LocalTime(0, 0).plus(period));
		return this.runAt(work, ldt.toDateTime());
	}

	// run the work unit repeatedly, every Secs seconds - note scheduler will not own work, you have to keep track of it
	public SchedulerNode runEvery(ITaskInfo work, int secs) {
		return this.addNode(new SimpleSchedule(work, DateTimeUtils.currentTimeMillis() + (1000 * secs), secs));		
	}
	
	public SchedulerNode addNode(ISchedule schedule) {
		long when = schedule.when();
		
		if (when < 0)
			return null;
		
		if ((schedule.task() == null) || (schedule.task().context() == null)) {
			TaskContext.getHubContext().warn("Schedule missing task or context: " + schedule.task());
			return null;
		}
		
		this.lock.lock();

        try {
        	SchedulerNode snode = new SchedulerNode(schedule, when);

			SchedulerNode curr = this.first;
            SchedulerNode last = null;

			this.nodeCnt++;

			// loop through the scheduling linked list and find the right place to insert
			// the new TScheduleNode
            while (curr != null) {
				if (snode.when < curr.when) {
                    snode.next = curr;

					if (last == null) 
						this.first = snode;
					else 
						last.next = snode;

                    return snode;
                }

                last = curr;
				curr = curr.next;
            }

            // none found then add to end
			if (last == null) 
				this.first = snode;
			else 
				last.next = snode;

            return snode;
        }
        catch(Exception x) {
        	// TODO

            return null;
        }
        finally {
        	this.lock.unlock();
        }
	}
	
	public class SchedulerNode implements ITaskInfo {
		long when = 0;
		SchedulerNode next = null;
		ISchedule schedule = null;
		ITaskInfo task = null;

		public SchedulerNode(ISchedule schedule, long when) {
			this.schedule = schedule;
			this.task = schedule.task();
			this.when = when;
		}
		
		@Override
		public TaskContext context() {
			return this.task.context();
		}
		
		@Override
		public void cancel() {
			this.task.cancel();
		}

		@Override
		public boolean canceled() {
			return this.task.canceled();
		}

		@Override
		public boolean hasRun() {
			return this.task.hasRun();
		}

		@Override
		public void run() {
			this.task.run();
			
			if (this.task.canceled())
				return;
		
			if (this.schedule.reschedule())
				Scheduler.this.addNode(this.schedule);
		}

		@Override
		public void setThreadSlot(int slot) {
			this.task.setThreadSlot(slot);
		}
	}
}
