package org.shypl.common.concurrent;

import org.shypl.common.util.Cancelable;
import org.shypl.common.util.LinkedQueue;
import org.slf4j.LoggerFactory;

import java.util.Queue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class Worker {
	private final Object          lock  = new Object();
	private final Queue<Runnable> queue = new LinkedQueue<>();
	private final ScheduledExecutorService executor;
	private       boolean                  executing;
	
	public Worker(ScheduledExecutorService executor) {
		this.executor = executor;
	}
	
	public void addTask(Runnable task) {
		synchronized (lock) {
			if (executing) {
				queue.add(task);
				return;
			}
			executing = true;
		}
		
		runTask(task);
		processQueue();
	}
	
	public Cancelable scheduleTask(Runnable task, long delay, TimeUnit unit) {
		ScheduledTaskHolder holder = new ScheduledTaskHolder(task);
		holder.setFuture(executor.schedule(holder, delay, unit));
		return holder;
	}
	
	public Cancelable scheduleTaskPeriodic(Runnable task, long period, TimeUnit unit) {
		return scheduleTaskPeriodic(task, period, period, unit);
	}
	
	public Cancelable scheduleTaskPeriodic(Runnable task, long initialDelay, long period, TimeUnit unit) {
		ScheduledTaskHolder scheduledTask = new ScheduledTaskHolder(task);
		scheduledTask.setFuture(executor.scheduleAtFixedRate(scheduledTask, initialDelay, period, unit));
		return scheduledTask;
	}
	
	private void processQueue() {
		Runnable task;
		
		while (true) {
			synchronized (lock) {
				task = queue.poll();
				if (task == null) {
					executing = false;
					return;
				}
			}
			runTask(task);
		}
	}
	
	private void runTask(Runnable task) {
		try {
			task.run();
		}
		catch (Throwable e) {
			LoggerFactory.getLogger(Worker.class).error("Error on run task " + task.getClass().getName(), e);
		}
	}
	
	private class ScheduledTaskHolder implements Cancelable, Runnable {
		private final Runnable           task;
		private       ScheduledFuture<?> future;
		private volatile boolean actual = true;
		
		public ScheduledTaskHolder(Runnable task) {
			this.task = task;
		}
		
		@Override
		public void run() {
			if (actual) {
				addTask(task);
			}
		}
		
		@Override
		public void cancel() {
			ScheduledFuture<?> f = future;
			if (actual) {
				actual = false;
				f.cancel(false);
				future = null;
			}
		}
		
		public void setFuture(ScheduledFuture<?> future) {
			this.future = future;
		}
	}
}

@@ -307,16 +307,52 @@
 imeUnit;
+%0Aimport java.util.function.Supplier;
 %0A%0Apublic
@@ -495,16 +495,25 @@
 e final 
+Supplier%3C
 Schedule
@@ -520,32 +520,33 @@
 dExecutorService
+%3E
  executor;%0A%09priv
@@ -593,16 +593,111 @@
 ting;%0A%09%0A
+%09public Worker(Supplier%3CScheduledExecutorService%3E executor) %7B%0A%09%09this.executor = executor;%0A%09%7D%0A%09%0A
 %09public 
@@ -749,32 +749,38 @@
 %09this.executor =
+ () -%3E
  executor;%0A%09%7D%0A%09%0A
@@ -1125,32 +1125,38 @@
 Future(executor.
+get().
 schedule(holder,
@@ -1550,16 +1550,22 @@
 xecutor.
+get().
 schedule

add supplier of executor