package com.uas.platform.b2b.core.util;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.springframework.util.CollectionUtils;

import com.uas.platform.core.concurrent.ICallable;
import com.uas.platform.core.concurrent.ICallableAdapter;
import com.uas.platform.core.concurrent.IRunnable;
import com.uas.platform.core.concurrent.IRunnableAdapter;
import com.uas.platform.core.concurrent.Listener;
import com.uas.platform.core.concurrent.Listeners;

/**
 * 多线程处理工具
 */
public class ThreadUtils {

	public static <T> Runner<T> tasks(IRunnable<T> runnable, List<T> paramList) {
		return new Runner<T>(runnable).tasks(paramList);
	}

	public static <T> void run(IRunnable<T> runnable, List<T> paramList) {
		tasks(runnable, paramList).run();
	}

	public static <T> Runner<T> pool(IRunnable<T> runnable, int poolSize) {
		return new Runner<T>(runnable, poolSize);
	}

	public static <V, T> Caller<V, T> tasks(ICallable<V, T> callable, List<T> paramList) {
		return new Caller<V, T>(callable).tasks(paramList);
	}

	public static <V, T> List<V> call(ICallable<V, T> callable, List<T> paramList) {
		return tasks(callable, paramList).call();
	}

	public static <V, T> Caller<V, T> pool(ICallable<V, T> callable, int poolSize) {
		return new Caller<V, T>(callable, poolSize);
	}

	public static Executor task(Runnable runnable) {
		return new Executor().task(runnable);
	}

	public static Executor pool(int poolSize) {
		return new Executor(poolSize);
	}

	/**
	 * @param <T>
	 *            参数类型
	 */
	public static class Runner<T> {
		private final ExecutorService threadPool;
		private IRunnable<T> runner;
		private int timeout = 900;
		private int taskCount = 0;

		public Runner() {
			this.threadPool = Executors.newCachedThreadPool();
		}

		public Runner(int poolSize) {
			this.threadPool = Executors.newFixedThreadPool(poolSize);
		}

		public Runner(IRunnable<T> runner) {
			this();
			this.runner = runner;
		}

		public Runner(IRunnable<T> runner, int poolSize) {
			this(poolSize);
			this.runner = runner;
		}

		public Runner<T> setTimeout(int timeout) {
			this.timeout = timeout;
			return this;
		}

		public Runner<T> task(T param) {
			threadPool.execute(new IRunnableAdapter<T>(runner, param));
			taskCount++;
			return this;
		}

		public Runner<T> tasks(List<T> params) {
			if (!CollectionUtils.isEmpty(params)) {
				for (T param : params) {
                    task(param);
                }
			}
			return this;
		}

		public void run() {
			if (taskCount > 0) {
				threadPool.shutdown();
				try {
					threadPool.awaitTermination(timeout, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

		public Runner<T> shutdownIf(Listeners closer) {
			closer.addListener(new Listener() {

				@Override
				public void handler() {
					threadPool.shutdownNow();
				}
			});
			return this;
		}

	}

	/**
	 * @param <V>
	 *            返回类型
	 * @param <T>
	 *            参数类型
	 */
	public static class Caller<V, T> {
		private final ExecutorService threadPool;
		private final CompletionService<V> service;
		private ICallable<V, T> caller;
		private int taskCount = 0;
		private AtomicBoolean closed = new AtomicBoolean(false);

		public Caller() {
			this.threadPool = Executors.newCachedThreadPool();
			this.service = new ExecutorCompletionService<V>(threadPool);
		}

		public Caller(int poolSize) {
			this.threadPool = Executors.newFixedThreadPool(poolSize);
			this.service = new ExecutorCompletionService<V>(threadPool);
		}

		public Caller(ICallable<V, T> caller) {
			this();
			this.caller = caller;
		}

		public Caller(ICallable<V, T> caller, int poolSize) {
			this(poolSize);
			this.caller = caller;
		}

		public Caller<V, T> task(T param) {
			service.submit(new ICallableAdapter<V, T>(caller, param));
			taskCount++;
			return this;
		}

		public Caller<V, T> tasks(List<T> params) {
			if (!CollectionUtils.isEmpty(params)) {
				for (T param : params) {
                    task(param);
                }
			}
			return this;
		}

		public List<V> call() {
			threadPool.shutdown();
			int finish = 0;
			List<V> results = new ArrayList<V>();
			while (finish < taskCount && !closed.get()) {
				Future<V> val = service.poll();
				// 采用非阻塞方式
				if (null != val) {
					try {
						results.add(val.get());
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (ExecutionException e) {
						e.printStackTrace();
					}
					finish++;
				}
				try {
					TimeUnit.MILLISECONDS.sleep(50);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			closed.set(true);
			return results;
		}

		public Caller<V, T> shutdownIf(Listeners closer) {
			closer.addListener(new Listener() {

				@Override
				public void handler() {
					if (!closed.get()) {
						threadPool.shutdownNow();
						closed.set(true);
					}
				}
			});
			return this;
		}

	}

	/**
	 * 无参、无返回执行
	 */
	public static class Executor {
		private final ExecutorService threadPool;
		private int timeout = 900;// 15 min
		private int taskCount = 0;

		public Executor() {
			this.threadPool = Executors.newCachedThreadPool();
		}

		public Executor(int poolSize) {
			this.threadPool = Executors.newFixedThreadPool(poolSize);
		}

		public Executor setTimeout(int timeout) {
			this.timeout = timeout;
			return this;
		}

		public Executor task(Runnable runnable) {
			threadPool.execute(runnable);
			taskCount++;
			return this;
		}

		public Executor tasks(List<Runnable> runnables) {
			if (!CollectionUtils.isEmpty(runnables)) {
				for (Runnable runnable : runnables) {
                    task(runnable);
                }
			}
			return this;
		}

		public void run() {
			if (taskCount > 0) {
				threadPool.shutdown();
				try {
					threadPool.awaitTermination(timeout, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

		public Executor shutdownIf(Listeners closer) {
			closer.addListener(new Listener() {

				@Override
				public void handler() {
					threadPool.shutdownNow();
				}
			});
			return this;
		}

	}
}