package teetime.framework;

import org.junit.Test;

import teetime.stage.InitialElementProducer;

public class TerminationTest {

	@Test(timeout = 2000)
	public void doesNotGetStuckInAdd() throws InterruptedException {
		Execution<TerminationConfig> execution = new Execution<TerminationConfig>(new TerminationConfig(1));
		execution.executeNonBlocking();
		Thread.sleep(500);
		execution.abortEventually();
	}

	private class TerminationConfig extends Configuration {
		InitialElementProducer<Integer> init = new InitialElementProducer<Integer>(1, 2, 3, 4, 5, 6);
		DoesNotRetrieveElements sinkStage = new DoesNotRetrieveElements();

		public TerminationConfig(final int capacity) {
			connectPorts(init.getOutputPort(), sinkStage.getInputPort(), capacity);
			addThreadableStage(sinkStage);
		}

	}

	private class DoesNotRetrieveElements extends AbstractConsumerStage<Integer> {

		@Override
		protected void execute(final Integer element) {
			int i = 0;
			while (true) {
				i++;
				if (i > 1) {
					Thread.currentThread().interrupt();
					break;
				}
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
				}
			}

		}

		@Override
		protected void terminate() {}

	}

}

package teetime.framework;

import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;

import java.util.Arrays;

import org.junit.Test;

public class TerminationTest {

	@Test(timeout = 3000)
	public void doesNotGetStuckInAdd() throws InterruptedException {
		TerminationConfig configuration = new TerminationConfig(1);
		Execution<TerminationConfig> execution = new Execution<TerminationConfig>(configuration);
		execution.executeNonBlocking();
		Thread.sleep(100);
		execution.abortEventually();
		assertThat(configuration.sinkStage.time - 450, is(greaterThan(configuration.init.time)));
	}

	private class TerminationConfig extends Configuration {
		InitialElementProducer<Integer> init = new InitialElementProducer<Integer>(1, 2, 3, 4, 5, 6);
		DoesNotRetrieveElements sinkStage = new DoesNotRetrieveElements();

		public TerminationConfig(final int capacity) {
			connectPorts(init.getOutputPort(), sinkStage.getInputPort(), capacity);
			addThreadableStage(sinkStage);
		}

	}

	private final class InitialElementProducer<T> extends AbstractProducerStage<T> {

		private final Iterable<T> elements;
		public long time;

		public InitialElementProducer(final T... elements) {
			this.elements = Arrays.asList(elements);
		}

		@Override
		protected void execute() {
			for (final T element : this.elements) {
				this.outputPort.send(element);
			}
			this.terminate();
		}

		@Override
		protected void terminate() {
			time = System.currentTimeMillis();
			super.terminate();
		}

	}

	private class DoesNotRetrieveElements extends AbstractConsumerStage<Integer> {

		public long time;

		@Override
		protected void execute(final Integer element) {
			int i = 0;
			while (true) {
				i++;
				try {
					Thread.sleep(500);
				} catch (InterruptedException e) {
					// First sleep will throw this
				}
				if (i > 1) {
					Thread.currentThread().interrupt();
					time = System.currentTimeMillis();
					break;
				}
			}

		}

		@Override
		protected void terminate() {}

	}

}

test now somewhat representative