rewrote both stages and adapted arrays

src/main/java/teetime/framework/AbstractDCStage.java

+package teetime.framework;
+
+/**
+ * Represents a stage to provide functionality for the divide and conquer paradigm
+ *
+ * @since 2.x
+ *
+ * @author Christian Wulf, Nelson Tavares de Sousa, Robin Mohr
+ *
+ * @param <I>
+ *            type of elements to be processed.
+ *
+ */
+public abstract class AbstractDCStage<I> extends AbstractStage {
+
+	// BETTER private final I[] buffer but see next TODO (l38)
+	private I leftBuffer = null;
+	private I rightBuffer = null;
+
+	private final DynamicConfigurationContext context;
+
+	protected final InputPort<I> inputPort = this.createInputPort();
+	protected final InputPort<I> leftInputPort = this.createInputPort();
+	protected final InputPort<I> rightInputPort = this.createInputPort();
+
+	protected final OutputPort<I> outputPort = this.createOutputPort();
+	protected final OutputPort<I> leftOutputPort = this.createOutputPort();
+	protected final OutputPort<I> rightOutputPort = this.createOutputPort();
+
+	/**
+	 * Divide and Conquer stages need the configuration context upon creation
+	 *
+	 */
+	public AbstractDCStage(final DynamicConfigurationContext context) {
+		if (null == context) {
+			throw new IllegalArgumentException("Context may not be null.");
+		}
+		// TODO this.buffer = (I[]) new Object[size]; but can't figure out 'size'
+		this.context = context;
+		// connect to self instead of dummy pipe upon creation
+		context.connectPorts(leftOutputPort, leftInputPort);
+		context.connectPorts(rightOutputPort, rightInputPort);
+	}
+
+	public final InputPort<I> getInputPort() {
+		return this.inputPort;
+	}
+
+	public final InputPort<I> getLeftInputPort() {
+		return this.leftInputPort;
+	}
+
+	public final InputPort<I> getRightInputPort() {
+		return this.rightInputPort;
+	}
+
+	public final OutputPort<I> getOutputPort() {
+		return this.outputPort;
+	}
+
+	public final OutputPort<I> getleftOutputPort() {
+		return this.leftOutputPort;
+	}
+
+	public final OutputPort<I> getrightOutputPort() {
+		return this.rightOutputPort;
+	}
+
+	@Override
+	protected final void executeStage() {
+		final I element = this.getInputPort().receive();
+		final I eLeft = this.getLeftInputPort().receive();
+		final I eRight = this.getRightInputPort().receive();
+
+		if (eLeft != null) {
+			this.logger.debug("Left " + eLeft.toString());
+			if (eRight != null) {
+				this.logger.debug("Right " + eRight.toString());
+				conquer(eLeft, eRight);
+			} else {
+				if (rightBuffer != null) {
+					this.logger.debug("RightB " + rightBuffer.toString());
+					conquer(eLeft, rightBuffer);
+				} else {
+					leftBuffer = eLeft;
+				}
+			}
+		} else if (eRight != null) {
+			if (leftBuffer != null) {
+				this.logger.debug("LeftB " + leftBuffer.toString());
+				conquer(leftBuffer, eRight);
+			} else {
+				rightBuffer = eRight;
+			}
+		} else if (element != null) {
+			this.logger.debug("E " + element.toString());
+			if (splitCondition(element)) {
+				this.logger.debug("[DC]" + this.getId() + "_" + "passed splitcondition_" + element.toString());
+				makeCopy(leftOutputPort, leftInputPort);
+				makeCopy(rightOutputPort, rightInputPort);
+				this.logger.debug("[DC]" + this.getId() + "_" + "DIVIDING_" + element.toString());
+				divide(element);
+			} else {
+				this.logger.debug("[DC]" + this.getId() + "_" + "SOLVING_" + element.toString());
+				solve(element);
+			}
+		} else {
+			this.logger.debug("NO ELEMENT RECEIVED!");
+			returnNoElement();
+		}
+	}
+
+	/**
+	 * A method to add a new copy (new instance) of this stage to the configuration, which should be executed in a own thread.
+	 *
+	 */
+	private void makeCopy(final OutputPort<I> out, final InputPort<I> in) {
+		final AbstractDCStage<I> newStage = debugCreateMethod();
+		context.connectPorts(out, newStage.getInputPort());
+		context.connectPorts(newStage.getOutputPort(), in);
+		context.beginThread(newStage);
+		context.sendSignals(out);
+	}
+
+	// BETTER Write the function code in this class instead
+	protected abstract AbstractDCStage<I> debugCreateMethod();
+
+	/**
+	 * Method to divide the given input and send to the left and right output ports.
+	 *
+	 * @param element
+	 *            An element to be split and further processed
+	 */
+	protected abstract void divide(final I element);
+
+	/**
+	 * Method to process the given input and send to the output port.
+	 *
+	 * @param element
+	 *            An element to be processed
+	 */
+	protected abstract void solve(final I element);
+
+	/**
+	 * Method to join the given inputs together and send to the output port.
+	 *
+	 * @param eLeft
+	 *            First half of the resulting element.
+	 * @param eRight
+	 *            Second half of the resulting element.
+	 */
+	protected abstract void conquer(final I eLeft, final I eRight);
+
+	/**
+	 * Determines whether or not to split the input problem by examining the given element
+	 *
+	 * @param element
+	 *            The element whose properties determine the split condition
+	 */
+	protected abstract boolean splitCondition(final I element);
+
+	// TODO get rid of this
+	public DynamicConfigurationContext getContext() {
+		// TODO Auto-generated method stub
+		return this.context;
+	}
+}

src/main/java/teetime/stage/QuicksortStage.java

+package teetime.stage;
+
+import teetime.framework.AbstractDCStage;
+import teetime.framework.DynamicConfigurationContext;
+import teetime.stage.util.QuicksortProblem;
+
+public final class QuicksortStage extends AbstractDCStage<QuicksortProblem> {
+
+	public QuicksortStage(final DynamicConfigurationContext context) {
+		super(context);
+	}
+
+	// TODO Get rid of this
+	@Override
+	protected AbstractDCStage<QuicksortProblem> debugCreateMethod() {
+		return new QuicksortStage(super.getContext());
+	}
+
+	@Override
+	protected void divide(final QuicksortProblem qsp) {
+		final int low = qsp.getLow();
+		final int high = qsp.getHigh();
+		final int[] arr = qsp.getArr();
+
+		// pick the pivot
+		final int middle = low + (high - low) / 2;
+		final int pivot = arr[middle];
+
+		// make left < pivot and right > pivot
+		int i = low, j = high;
+		while (i <= j) {
+			while (arr[i] < pivot) {
+				i++;
+			}
+
+			while (arr[j] > pivot) {
+				j--;
+			}
+
+			if (i <= j) {
+				int temp = arr[i];
+				arr[i] = arr[j];
+				arr[j] = temp;
+				i++;
+				j--;
+			}
+		}
+
+		// recursively sort two sub parts
+		QuicksortProblem newQuicksortProblem1 = new QuicksortProblem(low, j, arr);
+		leftOutputPort.send(newQuicksortProblem1);
+		QuicksortProblem newQuicksortProblem2 = new QuicksortProblem(i, high, arr);
+		rightOutputPort.send(newQuicksortProblem2);
+
+	}
+
+	@Override
+	protected void solve(final QuicksortProblem qsp) {
+		this.outputPort.send(qsp);
+	}
+
+	@Override
+	protected void conquer(final QuicksortProblem eLeft, final QuicksortProblem eRight) {
+		final int rlow = eRight.getLow();
+		final int rhigh = eRight.getHigh();
+		final int[] arr1 = eLeft.getArr();
+		final int[] arr2 = eRight.getArr();
+		for (int j = rlow; j <= rhigh; j++) {
+			arr1[j] = arr2[j];
+		}
+		QuicksortProblem newQuicksortProblem = new QuicksortProblem(eLeft.getLow(), rhigh, arr1);
+		this.outputPort.send(newQuicksortProblem);
+
+	}
+
+	@Override
+	protected boolean splitCondition(final QuicksortProblem qsp) {
+		return (((qsp.getHigh() - qsp.getLow()) >= 1) ? true : false);
+	}
+}

src/main/java/teetime/stage/util/QuicksortProblem.java

+package teetime.stage.util;
+
+/**
+ * @since 2.x
+ *
+ * @author Robin Mohr
+ *
+ */
+public final class QuicksortProblem {
+
+	private final int low;
+	private final int high;
+	private final int[] arr;
+
+	/**
+	 * An implementation of a quicksort problem.
+	 *
+	 * @param low
+	 *            Pointer to the lower bound of indices to be compared in the array
+	 * @param high
+	 *            Pointer to the upper bound of indices to be compared in the array
+	 * @param arr
+	 *            Array to be sorted
+	 */
+	public QuicksortProblem(final int low, final int high, final int[] arr) {
+		this.low = low;
+		this.high = high;
+		this.arr = arr;
+	}
+
+	public int getLow() {
+		return this.low;
+	}
+
+	public int getHigh() {
+		return this.high;
+	}
+
+	public int[] getArr() {
+		return this.arr;
+	}
+}

src/test/java/teetime/examples/quicksort/ArrayToQuicksortProblem.java

+package teetime.examples.quicksort;
+
+import teetime.framework.AbstractProducerStage;
+import teetime.stage.util.QuicksortProblem;
+
+public final class ArrayToQuicksortProblem extends AbstractProducerStage<QuicksortProblem> {
+
+	private final int[] arr;
+	private boolean firstTime = true;
+
+	public ArrayToQuicksortProblem(final int[] arr) {
+		this.arr = arr;
+	}
+
+	@Override
+	protected void execute() {
+		if (firstTime) {
+			final QuicksortProblem qsp = new QuicksortProblem(0, arr.length - 1, this.arr);
+			this.outputPort.send(qsp);
+			this.firstTime = false;
+		}
+	}
+
+}

src/test/java/teetime/examples/quicksort/QuicksortConfiguration.java

+package teetime.examples.quicksort;
+
+import teetime.framework.Configuration;
+import teetime.stage.QuicksortStage;
+
+public class QuicksortConfiguration extends Configuration {
+
+	public QuicksortConfiguration(final int[] arr) {
+
+		final ArrayToQuicksortProblem ltqsp = new ArrayToQuicksortProblem(arr);
+		final QuicksortStage qsort = new QuicksortStage(this.getDynamicContext());
+		final QuicksortProblemToList qsptl = new QuicksortProblemToList();
+
+		connectPorts(ltqsp.getOutputPort(), qsort.getInputPort());
+		connectPorts(qsort.getOutputPort(), qsptl.getInputPort());
+
+	}
+
+}

src/test/java/teetime/examples/quicksort/QuicksortProblemToList.java

+package teetime.examples.quicksort;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import teetime.framework.AbstractConsumerStage;
+import teetime.stage.util.QuicksortProblem;
+
+public final class QuicksortProblemToList extends AbstractConsumerStage<QuicksortProblem> {
+
+	@Override
+	protected void execute(final QuicksortProblem qsp) {
+		final List<Integer> list = new ArrayList<Integer>();
+		final int[] arr = qsp.getArr();
+		for (int j = 0; j < arr.length; j++) {
+			list.add(arr[j]);
+		}
+		System.out.println(list.toString());
+	}
+}

src/test/java/teetime/examples/quicksort/QuicksortTest.java

+package teetime.examples.quicksort;
+
+import java.util.Random;
+
+import org.junit.Test;
+
+import teetime.framework.Execution;
+
+public class QuicksortTest {
+
+	public QuicksortTest() {}
+
+	@Test
+	public void executeTest() {
+		final int[] arr = generateRandomNumbers(100);
+
+		final QuicksortConfiguration configuration = new QuicksortConfiguration(arr);
+		final Execution<QuicksortConfiguration> execution = new Execution<QuicksortConfiguration>(configuration);
+		execution.executeBlocking();
+
+		// Assert.assertTrue(Files.equal(new File(inputFile), new File(outputFile)));
+	}
+
+	private int[] generateRandomNumbers(final int n) {
+
+		int[] arr = new int[n];
+		Random random = new Random();
+
+		for (int i = 0; i < n; i++) {
+			arr[i] = (random.nextInt(n * 10));
+		}
+
+		return arr;
+	}
+
+}