Opened the abstract processor

This commit is contained in:
Lennart Hensler committed 2014-02-21 22:22:30 +01:00
1 parent 85eefeab5b
commit e0613b2189
2 files changed
+38 -10

No files matched your search

@@ -9,31 +9,36 @@ import java.util.logging.Logger;
import com.sap.sse.datamining.components.Processor; import com.sap.sse.datamining.components.Processor;
public abstract class AbstractParallelProcessor<InputType, ResultType> implements Processor<InputType> { public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingType, ResultType> implements Processor<InputType> {
private static final Logger LOGGER = Logger.getLogger(AbstractParallelProcessor.class.getName()); private static final Logger LOGGER = Logger.getLogger(AbstractPartitioningParallelProcessor.class.getName());
private Set<Processor<ResultType>> resultReceivers; private Set<Processor<ResultType>> resultReceivers;
private Executor executor; private Executor executor;
private int openInstructions; private int openInstructions;
public AbstractParallelProcessor(Executor executor, Collection<Processor<ResultType>> resultReceivers) { public AbstractPartitioningParallelProcessor(Executor executor, Collection<Processor<ResultType>> resultReceivers) {
this.executor = executor; this.executor = executor;
this.resultReceivers = new HashSet<Processor<ResultType>>(resultReceivers); this.resultReceivers = new HashSet<Processor<ResultType>>(resultReceivers);
} }
@Override @Override
public void onElement(InputType element) { public void onElement(InputType element) {
Runnable instruction = createInstruction(element); for (WorkingType partialElement : partitionElement(element)) {
if (instructionIsValid(instruction)) { Runnable instruction = createInstruction(partialElement);
NotifyingInstruction notifyingInstruction = new NotifyingInstruction(instruction); if (instructionIsValid(instruction)) {
openInstructions++; NotifyingInstruction notifyingInstruction = new NotifyingInstruction(
executor.execute(notifyingInstruction); instruction);
openInstructions++;
executor.execute(notifyingInstruction);
}
} }
} }
protected abstract Runnable createInstruction(InputType element); protected abstract Runnable createInstruction(WorkingType partialElement);
protected abstract Iterable<WorkingType> partitionElement(InputType element);
private boolean instructionIsValid(Runnable instruction) { private boolean instructionIsValid(Runnable instruction) {
return instruction != null; return instruction != null;
@@ -80,7 +85,7 @@ public abstract class AbstractParallelProcessor<InputType, ResultType> implement
@Override @Override
public void run() { public void run() {
innerInstruction.run(); innerInstruction.run();
AbstractParallelProcessor.this.instructionCompleted(); AbstractPartitioningParallelProcessor.this.instructionCompleted();
} }
} }
@@ -0,0 +1,23 @@
package com.sap.sse.datamining.impl.components;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.Executor;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractSimpleParallelProcessor<InputType, ResultType>
extends
AbstractPartitioningParallelProcessor<InputType, InputType, ResultType> {
public AbstractSimpleParallelProcessor(Executor executor,
Collection<Processor<ResultType>> resultReceivers) {
super(executor, resultReceivers);
}
@Override
protected Iterable<InputType> partitionElement(InputType element) {
return Collections.singleton(element);
}
}