package io.domisum.lib.auxiliumlib.thread.multithreader;

import io.domisum.lib.auxiliumlib.annotations.API;
import io.domisum.lib.auxiliumlib.util.ThreadUtil;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

@API
@RequiredArgsConstructor
public class Multithreader<I, O>
{
	
	private final Logger logger = LoggerFactory.getLogger(getClass());
	
	
	// SETTINGS
	private final Function<I, O> action;
	
	private final String name;
	private final int numberOfThreads;
	
	// STATUS
	private final AtomicInteger runCounter = new AtomicInteger(0);
	
	
	// INIT
	@API
	public static <I, O> Multithreader<I, O> create(Function<I, O> action, String name)
	{
		return create(action, name, defaultNumerOfThreads());
	}
	
	@API
	public static <I, O> Multithreader<I, O> overcreate(Function<I, O> action, String name, double overcreateFactor)
	{
		int numberOfThreads = (int) Math.ceil(overcreateFactor*defaultNumerOfThreads());
		return new Multithreader<>(action, name, numberOfThreads);
	}
	
	@API
	public static <I, O> Multithreader<I, O> create(Function<I, O> action, String name, int numberOfThreads)
	{
		return new Multithreader<>(action, name, numberOfThreads);
	}
	
	private static int defaultNumerOfThreads()
	{
		return Runtime.getRuntime().availableProcessors();
	}
	
	
	// RUN
	public Set<O> run(MultithreaderQueue<I> queue)
	{
		int runCount = runCounter.incrementAndGet();
		var run = new Run(runCount, queue, action);
		
		return run.run();
	}
	
	@RequiredArgsConstructor
	private class Run
	{
		
		// SETTINGS
		private final int runCount;
		private final MultithreaderQueue<I> queue;
		private final Function<I, O> action;
		
		// OUTPUT
		private final Set<O> outputs = new HashSet<>();
		
		
		// RUN
		public Set<O> run()
		{
			var executorService = createExecutorService();
			var futures = scheduleThreadLoopTasks(executorService);
			
			waitForFuturesCompletion(futures);
			executorService.shutdown();
			
			// noinspection AssignmentOrReturnOfFieldWithMutableType
			return outputs;
		}
		
		private ExecutorService createExecutorService()
		{
			var executorService = Executors.newFixedThreadPool(numberOfThreads, getThreadFactory());
			return executorService;
		}
		
		private ThreadFactory getThreadFactory()
		{
			var threadCount = new AtomicInteger(0);
			var threadGroup = Thread.currentThread().getThreadGroup();
			
			return runnable->
			{
				String threadName = name+"-run"+runCount+"-t"+threadCount.incrementAndGet();
				
				var thread = new Thread(threadGroup, runnable, threadName);
				thread.setDaemon(true);
				
				return thread;
			};
		}
		
		private ArrayList<Future<?>> scheduleThreadLoopTasks(ExecutorService executorService)
		{
			var futures = new ArrayList<Future<?>>();
			for(int i = 0; i < numberOfThreads; i++)
			{
				var future = executorService.submit(this::threadLoop);
				futures.add(future);
			}
			
			return futures;
		}
		
		private void waitForFuturesCompletion(Collection<Future<?>> futures)
		{
			try
			{
				for(var future : futures)
					future.get();
			}
			catch(InterruptedException|ExecutionException e)
			{
				throw new IllegalStateException("Failed to multithread", e);
			}
		}
		
		private void threadLoop()
		{
			while(queue.areThereMoreElements())
			{
				var elementOptional = queue.poll();
				
				if(elementOptional.isPresent())
					processElement(elementOptional.get());
				else
					ThreadUtil.sleep(Duration.ofMillis(10));
			}
		}
		
		private void processElement(I element)
		{
			try
			{
				processElementUncaught(element);
			}
			catch(Exception e)
			{
				logger.error("Exception while multithreading", e);
			}
		}
		
		private void processElementUncaught(I element)
		{
			var output = action.apply(element);
			if(output != null)
				outputs.add(output);
		}
		
	}
	
}

@@ -1013,24 +1013,156 @@
 ger(0);%0A%09%0A%09%0A
+%09// CONSTANT METHODS%0A%09private static int DEFAULT_NUMBER_OF_THREADS()%0A%09%7B%0A%09%09return Runtime.getRuntime().availableProcessors();%0A%09%7D%0A%09%0A%09%0A
 %09// INIT%0A%09@A
@@ -1251,28 +1251,62 @@
 name
-)%0A%09%7B%0A%09%09return create
+, int numberOfThreads)%0A%09%7B%0A%09%09return new Multithreader%3C%3E
 (act
@@ -1316,26 +1316,20 @@
 , name, 
-defaultN
+n
 um
+b
 erOfThre
@@ -1327,26 +1327,24 @@
 berOfThreads
-()
 );%0A%09%7D%0A%09%0A%09@AP
@@ -1387,20 +1387,16 @@
 r%3CI, O%3E 
-over
 create(F
@@ -1432,179 +1432,70 @@
 name
-, double overcreateFactor)%0A%09%7B%0A%09%09int numberOfThreads = (int) Math.ceil(overcreateFactor*defaultNumerOfThreads());%0A%09%09return new Multithreader%3C%3E(action, name, numberOfThreads
+)%0A%09%7B%0A%09%09return create(action, name, DEFAULT_NUMBER_OF_THREADS()
 );%0A%09
@@ -1538,32 +1538,36 @@
 ithreader%3CI, O%3E 
+over
 create(Function%3C
@@ -1597,195 +1597,181 @@
 me, 
-int numberOfThreads)%0A%09%7B%0A%09%09return new Multithreader%3C%3E(action, name, numberOfThreads);%0A%09%7D%0A%09%0A%09private static int defaultNumerOfThreads()%0A%09%7B%0A%09%09return Runtime.getRuntime().availableProcessors(
+double overcreateFactor)%0A%09%7B%0A%09%09int numberOfThreads = (int) Math.ceil(overcreateFactor*DEFAULT_NUMBER_OF_THREADS());%0A%09%09return new Multithreader%3C%3E(action, name, numberOfThreads
 );%0A%09

Clean up Multithreader