eclipse-sailing-analytics/sailing-analytics
1
0 Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-11-14 22:43:57 +00:00
Code Activity

Simplified the processor inheritance hierarchy

Browse Source
This commit is contained in:
Lennart Hensler committed 2015-04-17 11:07:24 +02:00
1 parent ce9a332ec9
commit 3dda7463ef
38 files changed
+157 -233
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

java/com.sap.sailing.datamining.test/src/com/sap/sailing/datamining/test/util/NullProcessor.java
+5
View File
@@ -8,6 +8,11 @@ public class NullProcessor<InputType, ResultType> extends AbstractProcessor<Inpu
public NullProcessor(Class<InputType> inputType, Class<ResultType> resultType) {
super(inputType, resultType);
}
@Override
public boolean canProcessElements() {
return true;
}
@Override
public void processElement(InputType element) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/CompetitorOfRaceInLeaderboardRetrievalProcessor.java
+3 -3
View File
@@ -4,16 +4,16 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import com.sap.sailing.datamining.data.HasRaceResultOfCompetitorContext;
import com.sap.sailing.datamining.data.HasLeaderboardContext;
import com.sap.sailing.datamining.data.HasRaceResultOfCompetitorContext;
import com.sap.sailing.datamining.impl.data.RaceResultOfCompetitorWithContext;
import com.sap.sailing.domain.base.Competitor;
import com.sap.sailing.domain.base.RaceColumn;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class CompetitorOfRaceInLeaderboardRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<HasLeaderboardContext, HasRaceResultOfCompetitorContext> {
AbstractRetrievalProcessor<HasLeaderboardContext, HasRaceResultOfCompetitorContext> {
public CompetitorOfRaceInLeaderboardRetrievalProcessor(ExecutorService executor,
Collection<Processor<HasRaceResultOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java
+2 -2
View File
@@ -12,9 +12,9 @@ import com.sap.sailing.domain.common.tracking.GPSFixMoving;
import com.sap.sailing.domain.tracking.GPSFixTrack;
import com.sap.sailing.domain.tracking.TrackedLegOfCompetitor;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class GPSFixRetrievalProcessor extends AbstractSimpleRetrievalProcessor<HasTrackedLegOfCompetitorContext, HasGPSFixContext> {
public class GPSFixRetrievalProcessor extends AbstractRetrievalProcessor<HasTrackedLegOfCompetitorContext, HasGPSFixContext> {
public GPSFixRetrievalProcessor(ExecutorService executor, Collection<Processor<HasGPSFixContext, ?>> resultReceivers, int retrievalLevel) {
super(HasTrackedLegOfCompetitorContext.class, HasGPSFixContext.class, executor, resultReceivers, retrievalLevel);
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/LeaderboardGroupRetrievalProcessor.java
+2 -3
View File
@@ -8,10 +8,9 @@ import com.sap.sailing.datamining.impl.data.LeaderboardGroupWithContext;
import com.sap.sailing.polars.PolarDataService;
import com.sap.sailing.server.RacingEventService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class LeaderboardGroupRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<RacingEventService, LeaderboardGroupWithContext> {
public class LeaderboardGroupRetrievalProcessor extends AbstractRetrievalProcessor<RacingEventService, LeaderboardGroupWithContext> {
public LeaderboardGroupRetrievalProcessor(ExecutorService executor,
Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/LeaderboardRetrievalProcessor.java
+2 -3
View File
@@ -9,10 +9,9 @@ import com.sap.sailing.datamining.impl.data.LeaderboardGroupWithContext;
import com.sap.sailing.datamining.impl.data.LeaderboardWithContext;
import com.sap.sailing.domain.leaderboard.Leaderboard;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class LeaderboardRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<LeaderboardGroupWithContext, HasLeaderboardContext> {
public class LeaderboardRetrievalProcessor extends AbstractRetrievalProcessor<LeaderboardGroupWithContext, HasLeaderboardContext> {
public LeaderboardRetrievalProcessor(ExecutorService executor,
Collection<Processor<HasLeaderboardContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/MarkPassingRetrievalProcessor.java
+3 -4
View File
@@ -1,11 +1,11 @@
package com.sap.sailing.datamining.impl.components;
import java.util.Collection;
import java.util.HashSet;
import java.util.NavigableSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.HashSet;
import com.sap.sailing.datamining.data.HasMarkPassingContext;
import com.sap.sailing.datamining.data.HasTrackedRaceContext;
import com.sap.sailing.datamining.impl.data.MarkPassingWithContext;
@@ -13,10 +13,9 @@ import com.sap.sailing.domain.base.Competitor;
import com.sap.sailing.domain.tracking.MarkPassing;
import com.sap.sailing.domain.tracking.TrackedRace;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class MarkPassingRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<HasTrackedRaceContext, HasMarkPassingContext> {
public class MarkPassingRetrievalProcessor extends AbstractRetrievalProcessor<HasTrackedRaceContext, HasMarkPassingContext> {
public MarkPassingRetrievalProcessor(ExecutorService executor,
Collection<Processor<HasMarkPassingContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/TrackedLegOfCompetitorRetrievalProcessor.java
+2 -3
View File
@@ -9,10 +9,9 @@ import com.sap.sailing.datamining.data.HasTrackedLegOfCompetitorContext;
import com.sap.sailing.datamining.impl.data.TrackedLegOfCompetitorWithContext;
import com.sap.sailing.domain.base.Competitor;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class TrackedLegOfCompetitorRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<HasTrackedLegContext, HasTrackedLegOfCompetitorContext> {
public class TrackedLegOfCompetitorRetrievalProcessor extends AbstractRetrievalProcessor<HasTrackedLegContext, HasTrackedLegOfCompetitorContext> {
public TrackedLegOfCompetitorRetrievalProcessor(ExecutorService executor,
Collection<Processor<HasTrackedLegOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/TrackedLegRetrievalProcessor.java
+2 -3
View File
@@ -9,10 +9,9 @@ import com.sap.sailing.datamining.data.HasTrackedRaceContext;
import com.sap.sailing.datamining.impl.data.TrackedLegWithContext;
import com.sap.sailing.domain.tracking.TrackedLeg;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class TrackedLegRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<HasTrackedRaceContext, HasTrackedLegContext> {
public class TrackedLegRetrievalProcessor extends AbstractRetrievalProcessor<HasTrackedRaceContext, HasTrackedLegContext> {
public TrackedLegRetrievalProcessor(ExecutorService executor,
Collection<Processor<HasTrackedLegContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/TrackedRaceRetrievalProcessor.java
+2 -3
View File
@@ -12,10 +12,9 @@ import com.sap.sailing.domain.base.RaceColumn;
import com.sap.sailing.domain.base.Regatta;
import com.sap.sailing.domain.tracking.TrackedRace;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
public class TrackedRaceRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<HasLeaderboardContext, HasTrackedRaceContext> {
public class TrackedRaceRetrievalProcessor extends AbstractRetrievalProcessor<HasLeaderboardContext, HasTrackedRaceContext> {
public TrackedRaceRetrievalProcessor(ExecutorService executor,
Collection<Processor<HasTrackedRaceContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/AbstractEnrichingProcessor.java
+2 -3
View File
@@ -5,11 +5,10 @@ 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.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends
AbstractSimpleParallelProcessor<InputType, ResultType> {
public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends AbstractParallelProcessor<InputType, ResultType> {
public AbstractEnrichingProcessor(Class<InputType> inputType, Class<ResultType> resultType,
ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/CubicRegressionPerCourseProcessor.java
+6
View File
@@ -25,6 +25,12 @@ public class CubicRegressionPerCourseProcessor implements Processor<GroupedDataE
private static final Logger logger = Logger.getLogger(CubicRegressionPerCourseProcessor.class.getName());
private final Map<GroupKey, AngleAndSpeedRegression> regressions = new HashMap<>();
@Override
public boolean canProcessElements() {
// TODO Auto-generated method stub
return true;
}
@Override
public void processElement(GroupedDataEntry<GPSFixMovingWithPolarContext> element) {
java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/MovingAverageProcessorImpl.java
+6
View File
@@ -49,6 +49,12 @@ public class MovingAverageProcessorImpl implements MovingAverageProcessor {
public MovingAverageProcessorImpl(ClusterGroup<Speed> speedClusterGroup) {
this.speedClusterGroup = speedClusterGroup;
}
@Override
public boolean canProcessElements() {
// TODO Auto-generated method stub
return true;
}
@Override
public void processElement(GroupedDataEntry<GPSFixMovingWithPolarContext> element) {
java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/SpeedRegressionPerAngleClusterProcessor.java
+6
View File
@@ -35,6 +35,12 @@ private static final Logger logger = Logger.getLogger(CubicRegressionPerCoursePr
public SpeedRegressionPerAngleClusterProcessor(ClusterGroup<Bearing> angleClusterGroup) {
this.angleClusterGroup = angleClusterGroup;
}
@Override
public boolean canProcessElements() {
// TODO Auto-generated method stub
return true;
}
@Override
public void processElement(GroupedDataEntry<GPSFixMovingWithPolarContext> element) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java
+6 -6
View File
@@ -31,9 +31,9 @@ import com.sap.sse.datamining.factories.ProcessorFactory;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.ParameterProvider;
import com.sap.sse.datamining.functions.ParameterizedFunction;
import com.sap.sse.datamining.impl.components.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.impl.components.ParallelFilteringProcessor;
import com.sap.sse.datamining.impl.criterias.AbstractFilterCriterion;
@@ -141,7 +141,7 @@ public class TestProcessorQuery {
retrievalResultReceivers.add(filtrationProcessor);
@SuppressWarnings("unchecked")
Processor<Iterable<Number>, Number> retrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Number>, Number>((Class<Iterable<Number>>)(Class<?>) Iterable.class, Number.class,
Processor<Iterable<Number>, Number> retrievalProcessor = new AbstractRetrievalProcessor<Iterable<Number>, Number>((Class<Iterable<Number>>)(Class<?>) Iterable.class, Number.class,
ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 0) {
@Override
protected Iterable<Number> retrieveData(Iterable<Number> element) {
@@ -292,7 +292,7 @@ public class TestProcessorQuery {
protected Processor<Iterable<Number>, ?> createFirstProcessor() {
Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers = new ArrayList<>();
resultReceivers.add(this.getResultReceiver());
return new AbstractSimpleParallelProcessor<Iterable<Number>, Map<GroupKey, Double>>((Class<Iterable<Number>>)(Class<?>) Iterable.class,
return new AbstractParallelProcessor<Iterable<Number>, Map<GroupKey, Double>>((Class<Iterable<Number>>)(Class<?>) Iterable.class,
(Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
ConcurrencyTestsUtil.getExecutor(),
resultReceivers) {
@@ -332,7 +332,7 @@ public class TestProcessorQuery {
protected Processor<Double, ?> createFirstProcessor() {
Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers = new ArrayList<>();
resultReceivers.add(this.getResultReceiver());
return new AbstractSimpleParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
return new AbstractParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
(Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
ConcurrencyTestsUtil.getExecutor(),
resultReceivers) {
@@ -372,7 +372,7 @@ public class TestProcessorQuery {
protected Processor<Double, ?> createFirstProcessor() {
Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers = new ArrayList<>();
resultReceivers.add(this.getResultReceiver());
return new AbstractSimpleParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
return new AbstractParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
(Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
ConcurrencyTestsUtil.getExecutor(),
resultReceivers) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorFinishing.java
+2 -2
View File
@@ -66,8 +66,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) {
private AbstractParallelProcessor<Integer, Integer> createProcessor(Collection<Processor<Integer, ?>> receivers) {
return new AbstractParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
@Override
protected AbstractProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new AbstractProcessorInstruction<Integer>(this) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorWithManySimpleInstructions.java
+1 -1
View File
@@ -36,7 +36,7 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
Collection<Processor<Integer, ?>> receivers = new ArrayList<>();
receivers.add(receiver);
processor = new AbstractSimpleParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
processor = new AbstractParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
@Override
protected AbstractProcessorInstruction<Integer> createInstruction(final Integer element) {
return new AbstractProcessorInstruction<Integer>(this) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestPartitioningOfAbstractSimpleParallelProcessor.java
-58
View File
@@ -1,58 +0,0 @@
package com.sap.sse.datamining.impl.components;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import java.util.HashSet;
import java.util.Iterator;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
public class TestPartitioningOfAbstractSimpleParallelProcessor {
private Iterable<Integer> partitionedElements;
@Test
public void testPartitioning() {
Processor<Integer, Integer> processor = new AbstractSimpleParallelProcessor<Integer, Integer>(Integer.class, Integer.class, FunctionTestsUtil.getExecutor(), new HashSet<Processor<Integer, ?>>()) {
@Override
protected Iterable<Integer> partitionElement(Integer element) {
partitionedElements = super.partitionElement(element);
return partitionedElements;
}
@Override
protected AbstractProcessorInstruction<Integer> createInstruction(Integer partialElement) {
return new AbstractProcessorInstruction<Integer>(this) {
@Override
public Integer computeResult() {
return 0;
}
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
};
processor.processElement(1);
ConcurrencyTestsUtil.sleepFor(200); //Giving the processor time to execute the instruction
verifyPartitionedElements(1);
}
private void verifyPartitionedElements(int expectedSize) {
int size = 0;
Iterator<Integer> iterator = partitionedElements.iterator();
while (iterator.hasNext()) {
iterator.next();
size++;
}
assertThat(size, is(expectedSize));
}
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestRetrieverFilterProcessorChain.java
+2 -2
View File
@@ -71,7 +71,7 @@ public class TestRetrieverFilterProcessorChain {
Collection<Processor<Iterable<Integer>, ?>> resultReceivers = new ArrayList<>();
resultReceivers.add(retrievalProcessor);
@SuppressWarnings("unchecked")
Processor<Iterable<Iterable<Integer>>, Iterable<Integer>> layeredRetrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Iterable<Integer>>, Iterable<Integer>>((Class<Iterable<Iterable<Integer>>>)(Class<?>) Iterable.class, (Class<Iterable<Integer>>)(Class<?>) Iterable.class,
Processor<Iterable<Iterable<Integer>>, Iterable<Integer>> layeredRetrievalProcessor = new AbstractRetrievalProcessor<Iterable<Iterable<Integer>>, Iterable<Integer>>((Class<Iterable<Iterable<Integer>>>)(Class<?>) Iterable.class, (Class<Iterable<Integer>>)(Class<?>) Iterable.class,
ConcurrencyTestsUtil.getExecutor(), resultReceivers, 0) {
@Override
protected Iterable<Iterable<Integer>> retrieveData(Iterable<Iterable<Integer>> element) {
@@ -114,7 +114,7 @@ public class TestRetrieverFilterProcessorChain {
Collection<Processor<Integer, ?>> retrievalResultReceivers = new ArrayList<>();
retrievalResultReceivers.add(filtrationProcessor);
retrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Integer>, Integer>((Class<Iterable<Integer>>)(Class<?>) Iterable.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 1) {
retrievalProcessor = new AbstractRetrievalProcessor<Iterable<Integer>, Integer>((Class<Iterable<Integer>>)(Class<?>) Iterable.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 1) {
@Override
protected Iterable<Integer> retrieveData(Iterable<Integer> element) {
return element;
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/BlockingProcessor.java
+2 -2
View File
@@ -5,10 +5,10 @@ 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.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
public abstract class BlockingProcessor<InputType, ResultType> extends AbstractSimpleParallelProcessor<InputType, ResultType> {
public abstract class BlockingProcessor<InputType, ResultType> extends AbstractParallelProcessor<InputType, ResultType> {
private final long timeToBlockInMillis;
public BlockingProcessor(Class<InputType> inputType, Class<ResultType> resultType,
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/NullProcessor.java
+5
View File
@@ -8,6 +8,11 @@ public class NullProcessor<InputType, ResultType> extends AbstractProcessor<Inpu
public NullProcessor(Class<InputType> inputType, Class<ResultType> resultType) {
super(inputType, resultType);
}
@Override
public boolean canProcessElements() {
return true;
}
@Override
public void processElement(InputType element) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/NullProcessorInstruction.java
+2 -2
View File
@@ -1,12 +1,12 @@
package com.sap.sse.datamining.test.util.components;
import com.sap.sse.datamining.impl.components.AbstractPartitioningParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
public class NullProcessorInstruction<ResultType> extends AbstractProcessorInstruction<ResultType> {
public NullProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
public NullProcessorInstruction(AbstractParallelProcessor<?, ResultType> processor,
ProcessorInstructionPriority priority) {
super(processor, priority);
}
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/TestLegOfCompetitorWithContextRetrievalProcessor.java
+2 -3
View File
@@ -5,15 +5,14 @@ import java.util.Collection;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
import com.sap.sse.datamining.test.functions.registry.test_classes.Test_Competitor;
import com.sap.sse.datamining.test.functions.registry.test_classes.Test_Leg;
import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasLegOfCompetitorContext;
import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasLegOfCompetitorContextImpl;
import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasRaceContext;
public class TestLegOfCompetitorWithContextRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>{
public class TestLegOfCompetitorWithContextRetrievalProcessor extends AbstractRetrievalProcessor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>{
public TestLegOfCompetitorWithContextRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_HasLegOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
super(Test_HasRaceContext.class, Test_HasLegOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/TestRaceWithContextRetrievalProcessor.java
+2 -3
View File
@@ -5,14 +5,13 @@ import java.util.Collection;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
import com.sap.sse.datamining.test.functions.registry.test_classes.Test_Race;
import com.sap.sse.datamining.test.functions.registry.test_classes.Test_Regatta;
import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasRaceContext;
import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasRaceContextImpl;
public class TestRaceWithContextRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<Test_Regatta, Test_HasRaceContext> {
public class TestRaceWithContextRetrievalProcessor extends AbstractRetrievalProcessor<Test_Regatta, Test_HasRaceContext> {
public TestRaceWithContextRetrievalProcessor(ExecutorService executor,
Collection<Processor<Test_HasRaceContext, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/TestRegattaRetrievalProcessor.java
+2 -3
View File
@@ -4,11 +4,10 @@ import java.util.Collection;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
import com.sap.sse.datamining.test.functions.registry.test_classes.Test_Regatta;
public class TestRegattaRetrievalProcessor extends
AbstractSimpleRetrievalProcessor<Collection<Test_Regatta>, Test_Regatta> {
public class TestRegattaRetrievalProcessor extends AbstractRetrievalProcessor<Collection<Test_Regatta>, Test_Regatta> {
@SuppressWarnings("unchecked")
public TestRegattaRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_Regatta, ?>> resultReceivers, int retrievalLevel) {
java/com.sap.sse.datamining/src/com/sap/sse/datamining/DataRetrieverChainDefinition.java
-1
View File
@@ -9,7 +9,6 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.DataRetrieverTypeWithInformation;
import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
import com.sap.sse.i18n.ResourceBundleStringMessages;
/**
java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/Processor.java
+5
View File
@@ -3,6 +3,11 @@ package com.sap.sse.datamining.components;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
public interface Processor<InputType, ResultType> {
/**
* @return <code>true</code>, if the processor accepts new elements to process.
*/
public boolean canProcessElements();
/**
* Processes the given element and forwards the result.
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/ProcessorQuery.java
+5
View File
@@ -223,6 +223,11 @@ public abstract class ProcessorQuery<AggregatedType, DataSourceType> implements
occuredFailures = new ArrayList<>();
}
@Override
public boolean canProcessElements() {
return true;
}
@Override
public void processElement(Map<GroupKey, AggregatedType> groupedAggregations) {
resultsLock.lock();
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelMultiDimensionalNestingGroupingProcessor.java
+1 -1
View File
@@ -13,7 +13,7 @@ import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.impl.NestingCompoundGroupKey;
public abstract class AbstractParallelMultiDimensionalNestingGroupingProcessor<DataType>
extends AbstractSimpleParallelProcessor<DataType, GroupedDataEntry<DataType>> {
extends AbstractParallelProcessor<DataType, GroupedDataEntry<DataType>> {
private Iterable<ParameterizedFunction<?>> parameterizedDimensions;
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java → java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java
+49 -30
View File
@@ -12,10 +12,9 @@ import java.util.logging.Logger;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingType, ResultType>
extends AbstractProcessor<InputType, ResultType> {
public abstract class AbstractParallelProcessor<InputType, ResultType> extends AbstractProcessor<InputType, ResultType> {
private static final Logger LOGGER = Logger.getLogger(AbstractPartitioningParallelProcessor.class.getName());
private static final Logger LOGGER = Logger.getLogger(AbstractParallelProcessor.class.getName());
private static final int SLEEP_TIME_DURING_FINISHING = 100;
private final Set<Processor<ResultType, ?>> resultReceivers;
@@ -25,27 +24,30 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
private boolean isFinished = false;
private boolean isAborted = false;
public AbstractPartitioningParallelProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
public AbstractParallelProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
super(inputType, resultType);
this.executor = executor;
this.resultReceivers = new HashSet<Processor<ResultType, ?>>(resultReceivers);
unfinishedInstructionsCounter = new AtomicInteger();
}
@Override
public boolean canProcessElements() {
return !isFinished() && !isAborted();
}
@Override
public void processElement(InputType element) {
if (!isFinished && !isAborted) {
for (WorkingType partialElement : partitionElement(element)) {
final AbstractProcessorInstruction<ResultType> instruction = createInstruction(partialElement);
if (isInstructionValid(instruction)) {
unfinishedInstructionsCounter.getAndIncrement();
try {
executor.execute(instruction);
} catch (RejectedExecutionException exc){
LOGGER.log(Level.WARNING, "A " + RejectedExecutionException.class.getSimpleName() +
" appeared during the processing.");
instruction.run();
}
if (canProcessElements()) {
final AbstractProcessorInstruction<ResultType> instruction = createInstruction(element);
if (isInstructionValid(instruction)) {
unfinishedInstructionsCounter.getAndIncrement();
try {
executor.execute(instruction);
} catch (RejectedExecutionException exc) {
LOGGER.log(Level.WARNING, "A " + RejectedExecutionException.class.getSimpleName()
+ " appeared during the processing.");
instruction.run();
}
}
}
@@ -54,31 +56,44 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
private boolean isInstructionValid(AbstractProcessorInstruction<ResultType> instruction) {
return instruction != null;
}
boolean isResultValid(ResultType result) {
return result != null;
}
AtomicInteger getUnfinishedInstructionsCounter() {
return unfinishedInstructionsCounter;
}
/**
* Forwards the given <code>result</code> to the result receivers, if it's {@link #isResultValid(Object) valid}
* and if the processor hasn't been {@link #abort() aborted}.
*
* @param result the element to forward to the result receivers
*/
protected void forwardResultToReceivers(ResultType result) {
if (isResultValid(result) && !isAborted()) {
for (Processor<ResultType, ?> resultReceiver : resultReceivers) {
resultReceiver.processElement(result);
}
}
}
/**
* Checks if the given <code>result</code> valid. For example a valid element has to be not <code>null</code>.
* If the element is valid, it can be forwarded to the result receivers.
*
* @param result the element to check
* @return <code>true</code>, if the result is valid, so it can be forwarded to the result receivers
*/
private boolean isResultValid(ResultType result) {
return result != null;
}
/**
* @return An invalid result, that won't be forwarded to the result receivers.
*/
protected ResultType createInvalidResult() {
return null;
}
protected void forwardResultToReceivers(ResultType result) {
for (Processor<ResultType, ?> resultReceiver : resultReceivers) {
resultReceiver.processElement(result);
}
}
protected abstract AbstractProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement);
protected abstract Iterable<WorkingType> partitionElement(InputType element);
protected abstract AbstractProcessorInstruction<ResultType> createInstruction(final InputType element);
@Override
public void onFailure(Throwable failure) {
@@ -86,6 +101,10 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
resultReceiver.onFailure(failure);
}
}
protected boolean isFinished() {
return isFinished;
}
@Override
public void finish() throws InterruptedException {
@@ -122,7 +141,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
}
}
boolean isAborted() {
protected boolean isAborted() {
return isAborted;
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java
+7 -9
View File
@@ -2,7 +2,7 @@ package com.sap.sse.datamining.impl.components;
public abstract class AbstractProcessorInstruction<ResultType> implements Runnable, Comparable<AbstractProcessorInstruction<?>> {
private final AbstractPartitioningParallelProcessor<?, ?, ResultType> processor;
private final AbstractParallelProcessor<?, ResultType> processor;
private final int priority;
/**
@@ -10,12 +10,12 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Runnab
* Use only, if <b>no processor</b> in the chain creates instruction <b>with</b> priorities other than 0 or
* if the chain doesn't use priorities at all.
* If this isn't the case use
* {@link #ProcessorInstruction(AbstractPartitioningParallelProcessor, ProcessorInstructionPriority)}
* {@link #ProcessorInstruction(AbstractParallelProcessor, ProcessorInstructionPriority)}
* instead.
*
* @param processor The processor, that processes this instruction
*/
public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor) {
public AbstractProcessorInstruction(AbstractParallelProcessor<?, ResultType> processor) {
this(processor, 0);
}
@@ -25,7 +25,7 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Runnab
* @param processor The processor, that processes this instruction
* @param priority The priority of this instruction
*/
public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
public AbstractProcessorInstruction(AbstractParallelProcessor<?, ResultType> processor,
ProcessorInstructionPriority priority) {
this(processor, priority.asIntValue());
}
@@ -34,13 +34,13 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Runnab
* Creates an instruction with the given priority as <code>int</code>.<br />
* <b>Use only, if you need instructions with custom priorities and if you know what you're doing!</b>
* If this isn't the case use
* {@link #ProcessorInstruction(AbstractPartitioningParallelProcessor, ProcessorInstructionPriority)}
* {@link #ProcessorInstruction(AbstractParallelProcessor, ProcessorInstructionPriority)}
* instead.
*
* @param processor The processor, that processes this instruction
* @param priority The priority of this instruction. <code>0</code> is the highest priority.
*/
public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
public AbstractProcessorInstruction(AbstractParallelProcessor<?, ResultType> processor,
int priority) {
this.processor = processor;
this.priority = priority;
@@ -50,9 +50,7 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Runnab
public void run() {
try {
ResultType result = computeResult();
if (processor.isResultValid(result) && !processor.isAborted()) {
processor.forwardResultToReceivers(result);
}
processor.forwardResultToReceivers(result);
} catch (Exception e) {
if (!processor.isAborted() || !(e instanceof InterruptedException)) {
processor.onFailure(e);
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractRetrievalProcessor.java
+8 -14
View File
@@ -7,8 +7,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractRetrievalProcessor<InputType, WorkingType, ResultType> extends
AbstractPartitioningParallelProcessor<InputType, WorkingType, ResultType> {
public abstract class AbstractRetrievalProcessor<InputType, ResultType> extends AbstractParallelProcessor<InputType, ResultType> {
private final int retrievalLevel;
private final AtomicInteger retrievedDataAmount;
@@ -29,25 +28,20 @@ public abstract class AbstractRetrievalProcessor<InputType, WorkingType, ResultT
}
@Override
protected AbstractProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement) {
protected AbstractProcessorInstruction<ResultType> createInstruction(final InputType element) {
return new AbstractProcessorInstruction<ResultType>(this, ProcessorInstructionPriority.createRetrievalPriority(retrievalLevel)) {
@Override
public ResultType computeResult() {
retrievedDataAmount.incrementAndGet();
return convertWorkingToResultType(partialElement);
for (ResultType retrievedElement : retrieveData(element)) {
retrievedDataAmount.incrementAndGet();
forwardResultToReceivers(retrievedElement);
}
return createInvalidResult();
}
};
}
protected abstract ResultType convertWorkingToResultType(WorkingType partialElement);
// Override, to provide a better method name to the sub classes
@Override
protected Iterable<WorkingType> partitionElement(InputType element) {
return retrieveData(element);
}
protected abstract Iterable<WorkingType> retrieveData(InputType element);
protected abstract Iterable<ResultType> retrieveData(InputType element);
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractSimpleParallelProcessor.java
-26
View File
@@ -1,26 +0,0 @@
package com.sap.sse.datamining.impl.components;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractSimpleParallelProcessor<InputType, ResultType>
extends AbstractPartitioningParallelProcessor<InputType, InputType, ResultType> {
public AbstractSimpleParallelProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
super(inputType, resultType, executor, resultReceivers);
}
@Override
protected Iterable<InputType> partitionElement(InputType element) {
return Arrays.asList(element);
}
//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 AbstractProcessorInstruction<ResultType> createInstruction(InputType element);
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractSimpleRetrievalProcessor.java
-29
View File
@@ -1,29 +0,0 @@
package com.sap.sse.datamining.impl.components;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractSimpleRetrievalProcessor<InputType, ResultType> extends
AbstractRetrievalProcessor<InputType, ResultType, ResultType> {
/**
*
* @param inputType
* @param resultType
* @param executor
* @param resultReceivers
* @param retrievalLevel The position of this retriever in it's chain. <code>0</code> represents the first.
*/
public AbstractSimpleRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType,
ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers, int retrievalLevel) {
super(inputType, resultType, executor, resultReceivers, retrievalLevel);
}
@Override
protected ResultType convertWorkingToResultType(ResultType partialElement) {
return partialElement;
}
}
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelFilteringProcessor.java
+1 -1
View File
@@ -8,7 +8,7 @@ import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.FilterCriterion;
import com.sap.sse.datamining.components.Processor;
public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParallelProcessor<InputType, InputType> {
public class ParallelFilteringProcessor<InputType> extends AbstractParallelProcessor<InputType, InputType> {
private final FilterCriterion<InputType> filterCriterion;
private final AtomicInteger filteredDataAmount;
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java
+1 -1
View File
@@ -9,7 +9,7 @@ import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.ParameterProvider;
public class ParallelGroupedElementsValueExtractionProcessor<DataType, FunctionReturnType>
extends AbstractSimpleParallelProcessor<GroupedDataEntry<DataType>, GroupedDataEntry<FunctionReturnType>> {
extends AbstractParallelProcessor<GroupedDataEntry<DataType>, GroupedDataEntry<FunctionReturnType>> {
private final Function<FunctionReturnType> extractionFunction;
private final ParameterProvider parameterProvider;
java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java
+2 -2
View File
@@ -7,12 +7,12 @@ 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.AbstractParallelProcessor;
import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
public abstract class AbstractParallelStoringAggregationProcessor<InputType, AggregatedType>
extends AbstractSimpleParallelProcessor<InputType, AggregatedType> {
extends AbstractParallelProcessor<InputType, AggregatedType> {
private final Lock storeLock;
private final String aggregationNameMessageKey;
wiki/data-mining-architecture.md
+7 -8
View File
@@ -142,6 +142,7 @@ This is an example how a `DataRetrieverChainBuilder<DataSourceType>` can be conf
Processors are the main component of the framework to process the data and work with the [Pipes and Filter Architecture](http://de.wikipedia.org/wiki/Pipes_und_Filter). The interface `Processor<InputType, ResultType>` describes the functionality of processors, which combines pipes and filters. It has methods to process given elements or to react to a thrown failure and also methods to control the work-flow (for example to finish or abort the work), but it doesn't have methods to get the result of the data processing. The results of processors are forwarded to something, that depends on the concrete implementation (this will be other processors in most cases). Here are the most important methods of `Processor`:
* The method `canProcessElements` returns `true`, if the processor will accept new elements.
* The method `processElement` processes the given element and forwards the result.
* The method `onFailure` handles failures during the processing.
* The standard implementation is forwarding them to the last processor, that collects the failures, until the processing is finished. Than the failures will be handled.
@@ -153,33 +154,31 @@ Processors are the main component of the framework to process the data and work
* It won't accept new elements given with `processElement`.
* All result receivers will be told to abort.
How this functionality is implemented, depends on the concrete implementation, but it should stick to this description, except for some special cases (like the last processor in a chain). New processors should extend `AbstractPartitioningParallelProcessor` or `AbstractSimpleParallelProcessor`. The main difference between these two abstract processors is that the partitioning processor allows to convert one element of the `InputType` to an iterable of elements of a `WorkingType`, which can be useful in some cases. These two abstract implementations implement most of the functionality defined by `Processor`, where the processing of the input elements is parallel. The parallelization is done by the method `processElement`:
How this functionality is implemented, depends on the concrete implementation, but it should stick to this description, except for some special cases (like the last processor in a chain). New processors should extend `AbstractParallelProcessor`. This abstract implementation implements most of the functionality defined by `Processor`, with a parallel processing of the input elements. The parallelization is done in the method `processElement`:
* The given input element gets partitioned to an iterable of elements of the `WorkingType`.
* This step simplified by the `AbstractSimpleParallelProcessor` in the way, that the `InputType` and `WorkingType` are the same and the resulting iterable contains exactly the given input element.
* For each partial element in the resulting iterable is a `AbstractProcessorInstruction` created.
* An `AbstractProcessorInstruction` created if for the given input element.
* The instruction is given to an `ExecutorService`, if it is valid.
* For example are `null` instructions invalid.
* See [Processor Instructions](#Processor-Instructions) for detailed information about instructions.
* The instruction forwards its result, after it has been executed.
* The instruction forwards its result to the result receivers, after it has been executed.
* The execution of instructions is handled by the `ExecutorService` of the processor.
Abstract processors have the following abstract methods:
* The method `createInstruction` creates an `AbstractProcessorInstruction` for the given input element, in which the concrete functionality of the instruction is implemented.
* See [Processor Instructions](#Processor-Instructions) for detailed information about instructions.
* The method `partitionElement` converts the given input element to an iterable of elements of the `WorkingType`
* The `AbstractSimpleParallelProcessor` implements this method in the way, that the `InputType` and `WorkingType` are the same and the resulting iterable contains exactly the given input element.
* The method `setAdditionalData` sets the additional result data for the processor.
* For example the amount of filtered or retrieved data elements.
An useful method for concrete processors is the protected method `createInvalidResult`, that returns an instruction, that won't be passed to the `ExecutorService`. This can be used to stop the processing of a specific input element (for example to filter the data).
**TODO:** *Describe the usage of prioritized instructions, with the `DataMiningExecutorService`*
There are useful implementations of the abstract processors, that are more specialized. These special processors implement the creation of instructions and the setting of the additional data for their special case.
#### Retrieval Processor
Retrieval processors are used to get many result elements (as `ResultType`) from one input element (as `InputType`). The abstract class `AbstractSimpleRetrievalProcessor<InputType, ResultType>` should be used to implement concrete retrieval processors. It has the abstract method `retrieveData`, that has to implement the concrete algorithm to get the result elements from the input element. This method is used by the `partitionElement` method, so the retrieval process isn't parallel, but the processing of each retrieved element will be parallel.
Retrieval processors are used to get many result elements (as `ResultType`) from one input element (as `InputType`). The abstract class `AbstractRetrievalProcessor<InputType, ResultType>` should be used to implement concrete retrieval processors. It has the abstract method `retrieveData`, that has to implement the concrete algorithm to get the result elements from the input element. This method is used by the created instructions, which forwards each retrieved data element to the result receivers. The instruction itself returns an invalid result (created with the method `createInvalidResult`).
#### Filtering Processor
wiki/typical-data-mining-scenarios.md
+2 -2
View File
@@ -20,7 +20,7 @@
* The `ResourceBundleStringMessages` needs the resource base name, which should be `<string messages package name>/<string messages base name>`
* A reference to this `ResourceBundleStringMessages` should be returned by the method `getStringMessages()`
## Adding a completely new Data Type
## Adding an absolute new Data Type
A data type is an atomic unit on which the processing components operate. A data type is normally based on a domain element (like `GPSFixMoving` or `TrackedLegOfCompetitor`) and provides contextual information about it (like the name of the regatta or race) via [Functions](wiki/data-mining-architecture#Data-Mining-Functions).
@@ -50,7 +50,7 @@ A data type is an atomic unit on which the processing components operate. A data
* Add a `DataSourceProvider` to the `DataMiningBundleService`, if it doesn't contain one for the data source of the data type.
* Implement a retrieval processor for each data type.
* Retrieval processors should extend `AbstractSimpleRetrievalProcessor`.
* Retrieval processors should extend `AbstractRetrievalProcessor`.
* Retrieval processors used by retriever chains need a constructor with the signature `(ExecutorService, Collection<Processor<ResultType, ?>>, int)`.
* A retrieval processor describes how to get from the input data type (like `RacingEventService` or `TrackedRace`) to the result data type (like `TrackedLeg`) with the method `retrieveData`.
* Create a `DataRetrieverChainDefinition` and add it to the `DataMiningBundleService`.