package com.walker.scheduler;

import com.walker.infrastructure.arguments.ArgumentsManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
//import com.walkersoft.datagather.store.DatabaseStore;

public abstract class AbstractScheduler implements Scheduler {

	protected transient final Logger logger = LoggerFactory.getLogger(getClass());

	private int id;
	private String name;
	public void setId(int id) {
		this.id = id;
	}

	public void setName(String name) {
		this.name = name;
	}

	private Option option = null;

	// 内部时钟频率，默认：10秒
	private long timeInterval = 5000 * 1;

	protected long startTime = 0;
	private long restartTime = 0;

	// 记录线程引用，这样可以调用让线程中断睡眠
	// 2016-11-28 时克英
	private Thread kernelThread = null;

	public int nThreads = 1;
	public int nThreadQueue = 16;

	private ExecutorService executorService = null;
	private InternalTimerRunner timerRunner = new InternalTimerRunner();

	protected boolean started = false; // 是否已经启动运行

	// 如果任务执行没有返回值（调用请求没有得到数据），是否让任务终止
	private boolean taskTerminateCondition = false;

	// 任务调用失败最大值，默认：0表示不限制最大值
	private int maxFailedTimes = 0;
	private int currentFailedTime = 0;

	protected ScheduleEngine scheduleEngine = null;

	public ScheduleEngine getScheduleEngine() {
		return scheduleEngine;
	}

//	DatabaseStore store = null;

	// 是否核心调度器
	private boolean kernelScheduler = false;

	public AbstractScheduler(){}

	public AbstractScheduler(int id, String name
//			, DatabaseStore store
	){
		this.id = id;
		this.name = name;
//		this.store = store;
		if(this.executorService == null){
			ThreadPoolExecutor threadPoolExecutor =  new ThreadPoolExecutor(nThreads, nThreads,
					0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>(nThreadQueue),
					new ThreadFactory() {
						@Override
						public Thread newThread(Runnable r) {
							Thread t = new Thread(r);
							t.setDaemon(true);
							return t;
						}
					});
			this.executorService = threadPoolExecutor;
			logger.info("初始化了线程池对象'executorService', nThreads=" + nThreads + ", nThreadQueue=" + nThreadQueue);
		}
	}

	@Override
	public String getStoreId(){
//		if(store != null){
//			return store.getId();
//		}
//		return null;
		throw new UnsupportedOperationException("该属性废弃");
	}

	/**
	 * 返回采集请求数据发生错误的次数，无论是http或者数据库请求失败都会计数。<br>
	 * 当该值超过最大整数时，自动清零，并重新计数。
	 * @return
	 */
	public int getCurrentFailedTime() {
		return currentFailedTime;
	}

//	public DatabaseStore getStore() {
//		return store;
//	}

	@Override
	public int getId() {
		return this.id;
	}

	@Override
	public String getName() {
		return this.name;
	}

	@Override
	public Option getOption(){
		return this.option;
	}

	public long getTimeInterval() {
		return timeInterval;
	}

	@Override
	public long getStartTime() {
		return startTime;
	}

	public long getRestartTime() {
		return restartTime;
	}

	@Override
	public boolean isStarted() {
		return started;
	}

	@Override
	public boolean isPause(){
		if(timerRunner != null){
			return timerRunner.isPause();
		}
		return false;
	}

	@Override
	public boolean isTaskTerminateCondition() {
		return taskTerminateCondition;
	}

	@Override
	public boolean isKernelScheduler(){
		return kernelScheduler;
	}

	/**
	 * 中断执行线程，通常是在它间隔睡眠时，执行该方法唤醒。
	 */
	public void interruptKernelThread(){
		logger.debug("线程被唤醒一次：" + this.getName());
		if(this.kernelThread != null){
			kernelThread.interrupt();
		}
	}

	public boolean getStart(){
		return started;
	}

	public boolean getPause(){
		return isPause();
	}

	@Override
	public void start() {
		if(!started){
			this.checkData();
			executorService.execute(timerRunner);
			logger.info("调度器'" + name + "'启动......");
			started = true;
			if(scheduleEngine != null){
				scheduleEngine.setStatusStarted(id);
			}
		} else {
			throw new IllegalStateException("调度器已启动，调用状态错误。id = " + id);
		}
		startTime = System.currentTimeMillis();
	}

	@Override
	public void restart(){
		if(started && !timerRunner.isPause()){
			logger.warn("调度器正在执行，不能重复执行启动操作。id = " + id);
			throw new IllegalStateException("调度器正在执行，不能重复执行启动操作。id = " + id);
		}
		logger.info("调度器'" + name + "'暂停后，被重新运行......");
		timerRunner.setPause(false);

		this.checkData();
		executorService.execute(timerRunner);

		if(scheduleEngine != null){
			scheduleEngine.setStatusRestarted(id);
		}
		restartTime = System.currentTimeMillis();
	}

	private void checkData(){
		if(this.option == null){
			throw new IllegalStateException("调度器'" + this.name + "'无法启动: option未设置");
		}
	}

	@Override
	public void stop() {
		timerRunner.setStop();
		executorService.shutdown();
		started = false;
		if(scheduleEngine != null){
			scheduleEngine.setStatusStoped(id);
		}
		logger.info("调度器'" + name + "'被终止运行......");
	}

	@Override
	public void pause() {
		timerRunner.setPause(true);
		if(scheduleEngine != null){
			scheduleEngine.setStatusPaused(id);
		}
		logger.info("调度器'" + name + "'暂停......");
	}

	protected void setCurrentFailedTime(int currentFailedTime) {
		this.currentFailedTime = currentFailedTime;
	}

	public void setTimeInterval(long timeInterval) {
		this.timeInterval = timeInterval;
	}

	@Override
	public void setTaskTerminateCondition(boolean boo) {
		this.taskTerminateCondition = boo;
	}

	/**
	 * 设置调度任务失败最大次数，超过该次数，自动终止任务。
	 * @param maxFailedTimes
	 */
	@Override
	public void setMaxFailedTimes(int maxFailedTimes) {
		this.maxFailedTimes = maxFailedTimes;
	}

	@Override
	public void setOption(Option option) {
		if(option == null){
			throw new IllegalArgumentException("创建调度器失败：设置的option并不存在");
		}
		this.option = option;
	}

	@Override
	public void setScheduleEngine(ScheduleEngine scheduleEngine) {
		this.scheduleEngine = scheduleEngine;
	}

	/**
	 * 运行任务，由子类执行。因为输入任务参数只有子类才知道。
	 * @return
	 * @throws Exception
	 */
	protected abstract Object runTask() throws Exception;

	private class InternalTimerRunner implements Runnable{

		private boolean pause = false;
		private boolean stop = false;

		public void setStop() {
			this.stop = true;
		}

		public void setPause(boolean pause) {
//			if(pause){
//				// 暂停
//				try {
//					this.pause = true;
//					Thread.currentThread().wait();
//				} catch (InterruptedException e) {
//					e.printStackTrace();
//					this.pause = false;
//				}
//			} else {
//				// 恢复
//				Thread.currentThread().notify();
//				this.pause = false;
//			}
			this.pause = pause;
		}

		public boolean isPause() {
			return pause;
		}

		@Override
		public void run() {
			while(!stop){
				if(maxFailedTimes >= 1 && currentFailedTime >= maxFailedTimes){
					logger.info("调度被迫终止'" + name + "': 因为调用失败超过最大值(" + maxFailedTimes + ")");
//					stop();
//					this.stop = true;
					break;
				}

				// 采用新策略，暂停后，直接停止调用它的线程，恢复后重新执行线程任务
				// 2015-1-5 by shikeying
				if(this.pause){
					break;
				}

				try{
					Option.TimeObject to = option.isAvailableTime(System.currentTimeMillis());
					if(to.isAvailable()){
//					if(option.isAvailable(System.currentTimeMillis())){
						logger.debug("======== 满足时间要求，执行任务调用一次");
						onBeforeSchedule(to);
						Object result = runTask();

						//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
						// 时克英添加，2019-01-02，如果是周期性任务，每次执行完都必须切换到下个时间
						if(option.isCycleTask()){
							option.scheduleToNext(to);
						}
						//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

						if(result == null && taskTerminateCondition){
							logger.info("调度被迫终止'" + name + "': 因为调用未返回任何数据");
//								stop();
							break;
						}
					}
				} catch(Exception ex){
					if(ex instanceof NotFoundGatherDataException){
						logger.debug("数据源没有可采集数据，不属于业务错误。");
					} else {
						if(currentFailedTime >= Integer.MAX_VALUE){
							currentFailedTime = 0;
						}
						currentFailedTime ++;
						logger.error("任务调用失败一次，scheduler = " + id, ex);
					}
				} finally {
					try {
						if(timeInterval > 0){
//							logger.debug("*************** sleep: " + timeInterval);
							TimeUnit.MILLISECONDS.sleep(timeInterval);
						}
					} catch (InterruptedException e) {}
				}
			}
			if(!pause && !stop){
				// 只有没有暂停的情况下而且没有手动停止，才算调度执行完成
				if(scheduleEngine != null){
					scheduleEngine.setStatusDone(id);
				}
				started = false;
			}
		}
	}

	/**
	 * 每次执行调度之前，调用该方法。
	 * @param to
	 */
	protected void onBeforeSchedule(Option.TimeObject to){

	}

	private ArgumentsManager argumentManager;

	@Override
	public ArgumentsManager getArgumentManager() {
		return argumentManager;
	}

	public void setArgumentManager(ArgumentsManager argumentManager){
		this.argumentManager = argumentManager;
	}
}