/**
 * 
 */
package net.sf.jabb.util.parallel;

import static org.junit.Assert.*;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Function;

import org.junit.Test;

/**
 * @author James
 *
 */
public class PipelineTest {
	ExecutorService threadPool1 = Executors.newFixedThreadPool(5);
	ExecutorService threadPool2 = Executors.newFixedThreadPool(2);
	ExecutorService threadPool3 = Executors.newFixedThreadPool(4);
	ExecutorService threadPool4 = Executors.newFixedThreadPool(5);
	
	Function<String, Integer> stringToInteger = s -> Integer.valueOf(s);
	Function<Integer, Long> integerToLong = i -> {
		try{
			Thread.sleep(1000);
		}catch(InterruptedException e){
			Thread.interrupted();
		}
		return Long.valueOf(i);
	};
	Function<Long, BigInteger> longToBigInteger = l -> BigInteger.valueOf(l);
	Function<BigInteger, Integer> bigIntegerToInteger = big -> {
		if (big.compareTo(BigInteger.valueOf(-1)) == 0){
			throw new IllegalArgumentException("value -1 is not allowed");
		}
		return big.intValue();
	};

	@Test
	public void testSyntax() {
		Pipeline<String, Integer> pipeline = createPipeline();
		assertNotNull(pipeline);
	};
	
	Pipeline<String, Integer> createPipeline(){
		Pipeline<String, Integer> pipeline = 
			new Pipeline<BigInteger, Integer>(threadPool4, bigIntegerToInteger)
				.prepend(threadPool3, longToBigInteger)
				.prepend(threadPool2, integerToLong)
				.prepend(threadPool1, stringToInteger);
		return pipeline;
	}
	
	Integer createPipelineAndFeed(String input) throws InterruptedException, ExecutionException{
		Pipeline<String, Integer> pipeline = createPipeline();
		Future<Integer> result = pipeline.feed(input);
		Integer output = result.get();
		return output;
	}
	
	@Test
	public void testResult() throws InterruptedException, ExecutionException{
		String input = "18763";
		Integer output = createPipelineAndFeed(input);
		assertEquals(input, output.toString());
	}

	@Test(expected=ExecutionException.class)
	public void testExecutionExceptionInStage1() throws InterruptedException, ExecutionException{
		String input = "18sdf763";
		createPipelineAndFeed(input);
	}

	@Test(expected=ExecutionException.class)
	public void testExecutionExceptionInStage4() throws InterruptedException, ExecutionException{
		String input = "-1";
		createPipelineAndFeed(input);
	}

	@Test(expected=CancellationException.class)
	public void testCancel() throws InterruptedException, ExecutionException{
		String input = "18763";
		Pipeline<String, Integer> pipeline = createPipeline();
		Future<Integer> result = pipeline.feed(input);
		assertTrue(result.cancel(true));
		result.get();
	}
	
	@Test
	public void testFeedMore() throws InterruptedException, ExecutionException{
		Pipeline<String, Integer> pipeline = createPipeline();
		
		int SIZE = 50;
		String[] inputs = new String[SIZE];
		List<Future<Integer>> outputs = new ArrayList<>(SIZE);
		for (int i = 0; i < SIZE; i ++){
			String input = String.valueOf(i);
			inputs[i] = input;
			outputs.add(pipeline.feed(input));
		}
		
		for (int i = 0; i < SIZE; i ++){
			assertEquals(inputs[i], outputs.get(i).get().toString());
		}
	}
}

@@ -129,24 +129,53 @@
 .ArrayList;%0A
+import java.util.LinkedList;%0A
 import java.
@@ -1525,20 +1525,16 @@
 %0A%09%09%09
-new 
 Pipeline
 %3CBig
@@ -1529,37 +1529,64 @@
 Pipeline
-%3CBigInteger, Integer%3E
+.endWith(new LinkedList%3CInteger%3E())%0A%09%09%09%09.prepend
 (threadP
@@ -1764,16 +1764,19 @@
 ne;%0A%09%7D%0A%09
+%0A%0A%09
 %0A%09Intege

test with pipeline with output collection