Renamed ProcessorInstruction to AbstractProcessorInstruction

This commit is contained in:
Lennart Hensler committed 2015-03-21 13:21:17 +01:00
1 parent bb65e01ed4
commit dfb1753782
15 files changed
+35 -66

No files matched your search

@@ -6,7 +6,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
import com.sap.sse.datamining.impl.components.ProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends
AbstractSimpleParallelProcessor<InputType, ResultType> {
@@ -17,8 +17,8 @@ public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends
}
@Override
protected ProcessorInstruction<ResultType> createInstruction(final InputType element) {
return new ProcessorInstruction<ResultType>(this) {
protected AbstractProcessorInstruction<ResultType> createInstruction(final InputType element) {
return new AbstractProcessorInstruction<ResultType>(this) {
@Override
public ResultType computeResult() {
@@ -31,7 +31,7 @@ import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.impl.components.ParallelFilteringProcessor;
import com.sap.sse.datamining.impl.components.ProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.criterias.AbstractFilterCriterion;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.QueryResult;
@@ -287,8 +287,8 @@ public class TestProcessorQuery {
ConcurrencyTestsUtil.getExecutor(),
resultReceivers) {
@Override
protected ProcessorInstruction<Map<GroupKey, Double>> createInstruction(final Iterable<Number> element) {
return new ProcessorInstruction<Map<GroupKey,Double>>(this) {
protected AbstractProcessorInstruction<Map<GroupKey, Double>> createInstruction(final Iterable<Number> element) {
return new AbstractProcessorInstruction<Map<GroupKey,Double>>(this) {
@Override
public Map<GroupKey, Double> computeResult() {
Map<GroupKey, Double> result = new HashMap<>();
@@ -69,8 +69,8 @@ public class TestAbstractParallelProcessorFinishing {
private AbstractSimpleParallelProcessor<Integer, Integer> createProcessor(Collection<Processor<Integer, ?>> receivers) {
return new AbstractSimpleParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
@Override
protected ProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new ProcessorInstruction<Integer>(this) {
protected AbstractProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new AbstractProcessorInstruction<Integer>(this) {
@Override
public Integer computeResult() {
while (instructionIsWorking) {
@@ -38,8 +38,8 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
receivers.add(receiver);
processor = new AbstractSimpleParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
@Override
protected ProcessorInstruction<Integer> createInstruction(final Integer element) {
return new ProcessorInstruction<Integer>(this) {
protected AbstractProcessorInstruction<Integer> createInstruction(final Integer element) {
return new AbstractProcessorInstruction<Integer>(this) {
@Override
public Integer computeResult() {
return element;
@@ -27,8 +27,8 @@ public class TestPartitioningOfAbstractSimpleParallelProcessor {
}
@Override
protected ProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new ProcessorInstruction<Integer>(this) {
protected AbstractProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new AbstractProcessorInstruction<Integer>(this) {
@Override
public Integer computeResult() {
return 0;
@@ -1,31 +0,0 @@
package com.sap.sse.datamining.test.util.components;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
public class BlockingInstructionWrapper implements Runnable {
private Runnable innerInstruction;
private boolean workIsInProgress;
public BlockingInstructionWrapper(Runnable innerInstruction) {
this.innerInstruction = innerInstruction;
workIsInProgress = true;
}
@Override
public void run() {
while (workIsInProgress) {
ConcurrencyTestsUtil.sleepFor(100);
}
innerInstruction.run();
}
/**
* Tells the instruction, that the work is now done. It won't block it's inner instruction anymore and calls the run method of it.
*/
public void workIsNowFinished() {
workIsInProgress = false;
}
}
@@ -6,7 +6,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
import com.sap.sse.datamining.impl.components.ProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
public abstract class BlockingProcessor<InputType, ResultType> extends AbstractSimpleParallelProcessor<InputType, ResultType> {
private final long timeToBlockInMillis;
@@ -19,8 +19,8 @@ public abstract class BlockingProcessor<InputType, ResultType> extends AbstractS
}
@Override
protected ProcessorInstruction<ResultType> createInstruction(InputType element) {
return new ProcessorInstruction<ResultType>(this) {
protected AbstractProcessorInstruction<ResultType> createInstruction(InputType element) {
return new AbstractProcessorInstruction<ResultType>(this) {
@Override
public ResultType computeResult() throws Exception {
Thread.sleep(timeToBlockInMillis);
@@ -46,8 +46,8 @@ public abstract class AbstractParallelMultiDimensionalNestingGroupingProcessor<D
}
@Override
protected ProcessorInstruction<GroupedDataEntry<DataType>> createInstruction(final DataType element) {
return new ProcessorInstruction<GroupedDataEntry<DataType>>(this, ProcessorInstructionPriority.Grouping) {
protected AbstractProcessorInstruction<GroupedDataEntry<DataType>> createInstruction(final DataType element) {
return new AbstractProcessorInstruction<GroupedDataEntry<DataType>>(this, ProcessorInstructionPriority.Grouping) {
@Override
public GroupedDataEntry<DataType> computeResult() {
return new GroupedDataEntry<DataType>(createCompoundKeyFor(element,
@@ -36,7 +36,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
public void processElement(InputType element) {
if (!isFinished && !isAborted) {
for (WorkingType partialElement : partitionElement(element)) {
final ProcessorInstruction<ResultType> instruction = createInstruction(partialElement);
final AbstractProcessorInstruction<ResultType> instruction = createInstruction(partialElement);
if (isInstructionValid(instruction)) {
unfinishedInstructionsCounter.getAndIncrement();
try {
@@ -51,7 +51,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
}
}
private boolean isInstructionValid(ProcessorInstruction<ResultType> instruction) {
private boolean isInstructionValid(AbstractProcessorInstruction<ResultType> instruction) {
return instruction != null;
}
@@ -76,7 +76,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
}
}
protected abstract ProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement);
protected abstract AbstractProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement);
protected abstract Iterable<WorkingType> partitionElement(InputType element);
@@ -1,6 +1,6 @@
package com.sap.sse.datamining.impl.components;
public abstract class ProcessorInstruction<ResultType> implements Runnable, Comparable<ProcessorInstruction<?>> {
public abstract class AbstractProcessorInstruction<ResultType> implements Runnable, Comparable<AbstractProcessorInstruction<?>> {
private final AbstractPartitioningParallelProcessor<?, ?, ResultType> processor;
private final int priority;
@@ -15,7 +15,7 @@ public abstract class ProcessorInstruction<ResultType> implements Runnable, Comp
*
* @param processor The processor, that processes this instruction
*/
public ProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor) {
public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor) {
this(processor, 0);
}
@@ -25,7 +25,7 @@ public abstract class ProcessorInstruction<ResultType> implements Runnable, Comp
* @param processor The processor, that processes this instruction
* @param priority The priority of this instruction
*/
public ProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
ProcessorInstructionPriority priority) {
this(processor, priority.asIntValue());
}
@@ -40,7 +40,7 @@ public abstract class ProcessorInstruction<ResultType> implements Runnable, Comp
* @param processor The processor, that processes this instruction
* @param priority The priority of this instruction. <code>0</code> is the highest priority.
*/
public ProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
int priority) {
this.processor = processor;
this.priority = priority;
@@ -65,7 +65,7 @@ public abstract class ProcessorInstruction<ResultType> implements Runnable, Comp
protected abstract ResultType computeResult() throws Exception;
@Override
public int compareTo(ProcessorInstruction<?> instruction) {
public int compareTo(AbstractProcessorInstruction<?> instruction) {
return Integer.compare(priority, instruction.priority);
}
@@ -32,8 +32,8 @@ public abstract class AbstractRetrievalProcessor<InputType, WorkingType, ResultT
}
@Override
protected ProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement) {
return new ProcessorInstruction<ResultType>(this, ProcessorInstructionPriority.createRetrievalPriority(retrievalLevel)) {
protected AbstractProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement) {
return new AbstractProcessorInstruction<ResultType>(this, ProcessorInstructionPriority.createRetrievalPriority(retrievalLevel)) {
@Override
public ResultType computeResult() {
incrementRetrievedDataAmount();
@@ -21,6 +21,6 @@ public abstract class AbstractSimpleParallelProcessor<InputType, ResultType>
//Redefinition of the method to set the parameter name to element instead of partial element.
//This makes the implementation of sub classes more fluent.
@Override
protected abstract ProcessorInstruction<ResultType> createInstruction(InputType element);
protected abstract AbstractProcessorInstruction<ResultType> createInstruction(InputType element);
}
@@ -23,8 +23,8 @@ public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParalle
}
@Override
protected ProcessorInstruction<InputType> createInstruction(final InputType element) {
return new ProcessorInstruction<InputType>(this, ProcessorInstructionPriority.Filtration) {
protected AbstractProcessorInstruction<InputType> createInstruction(final InputType element) {
return new AbstractProcessorInstruction<InputType>(this, ProcessorInstructionPriority.Filtration) {
@Override
public InputType computeResult() {
if (filterCriterion.matches(element)) {
@@ -30,8 +30,8 @@ public class ParallelGroupedElementsValueExtractionProcessor<DataType, FunctionR
}
@Override
protected ProcessorInstruction<GroupedDataEntry<FunctionReturnType>> createInstruction(final GroupedDataEntry<DataType> element) {
return new ProcessorInstruction<GroupedDataEntry<FunctionReturnType>>(this, ProcessorInstructionPriority.Extraction) {
protected AbstractProcessorInstruction<GroupedDataEntry<FunctionReturnType>> createInstruction(final GroupedDataEntry<DataType> element) {
return new AbstractProcessorInstruction<GroupedDataEntry<FunctionReturnType>>(this, ProcessorInstructionPriority.Extraction) {
@Override
public GroupedDataEntry<FunctionReturnType> computeResult() {
FunctionReturnType value = extractionFunction.tryToInvoke(element.getDataEntry(), parameterProvider);
@@ -8,7 +8,7 @@ import java.util.concurrent.locks.ReentrantLock;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
import com.sap.sse.datamining.impl.components.ProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
public abstract class AbstractParallelStoringAggregationProcessor<InputType, AggregatedType>
@@ -28,8 +28,8 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
}
@Override
protected ProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
return new ProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
protected AbstractProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
return new AbstractProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
@Override
public AggregatedType computeResult() {
storeLock.lock();