mirror of
https://github.com/eclipse-sailing-analytics/sailing-analytics.git
synced 2026-11-06 19:12:23 +00:00
Added a priority to the ProcessorInstruction and made it Comparable
This commit is contained in:
1 parent
571d6cb93a
commit
bb65e01ed4
25 files changed
+156
-64
No files matched your search
+1
-1
@@ -46,7 +46,7 @@ public class TestLeaderboardGroupRetrievalProcessor {
|
||||
|
||||
Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers = new ArrayList<>();
|
||||
resultReceivers.add(receiver);
|
||||
retriever = new LeaderboardGroupRetrievalProcessor(ConcurrencyTestsUtil.getExecutor(), resultReceivers);
|
||||
retriever = new LeaderboardGroupRetrievalProcessor(ConcurrencyTestsUtil.getExecutor(), resultReceivers, 0);
|
||||
}
|
||||
|
||||
private Map<String, LeaderboardGroup> getGroupsInService() {
|
||||
|
||||
+2
-2
@@ -16,8 +16,8 @@ public class CompetitorOfRaceInLeaderboardRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<HasLeaderboardContext, HasRaceResultOfCompetitorContext> {
|
||||
|
||||
public CompetitorOfRaceInLeaderboardRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<HasRaceResultOfCompetitorContext, ?>> resultReceivers) {
|
||||
super(HasLeaderboardContext.class, HasRaceResultOfCompetitorContext.class, executor, resultReceivers);
|
||||
Collection<Processor<HasRaceResultOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(HasLeaderboardContext.class, HasRaceResultOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -16,8 +16,8 @@ import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
|
||||
|
||||
public class GPSFixRetrievalProcessor extends AbstractSimpleRetrievalProcessor<HasTrackedLegOfCompetitorContext, HasGPSFixContext> {
|
||||
|
||||
public GPSFixRetrievalProcessor(ExecutorService executor, Collection<Processor<HasGPSFixContext, ?>> resultReceivers) {
|
||||
super(HasTrackedLegOfCompetitorContext.class, HasGPSFixContext.class, executor, resultReceivers);
|
||||
public GPSFixRetrievalProcessor(ExecutorService executor, Collection<Processor<HasGPSFixContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(HasTrackedLegOfCompetitorContext.class, HasGPSFixContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+9
-10
@@ -14,19 +14,18 @@ public class LeaderboardGroupRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<RacingEventService, LeaderboardGroupWithContext> {
|
||||
|
||||
public LeaderboardGroupRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers) {
|
||||
super(RacingEventService.class, LeaderboardGroupWithContext.class, executor, resultReceivers);
|
||||
Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(RacingEventService.class, LeaderboardGroupWithContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Iterable<LeaderboardGroupWithContext> retrieveData(RacingEventService element) {
|
||||
final PolarDataService polarDataService = element.getPolarDataService();
|
||||
return element
|
||||
.getLeaderboardGroups()
|
||||
.values()
|
||||
.stream()
|
||||
.map(lg -> new LeaderboardGroupWithContext(lg, polarDataService))
|
||||
.collect(Collectors.toSet());
|
||||
}
|
||||
final PolarDataService polarDataService = element.getPolarDataService();
|
||||
return element.getLeaderboardGroups()
|
||||
.values()
|
||||
.stream()
|
||||
.map(lg -> new LeaderboardGroupWithContext(lg, polarDataService))
|
||||
.collect(Collectors.toSet());
|
||||
}
|
||||
|
||||
}
|
||||
+2
-2
@@ -15,8 +15,8 @@ public class LeaderboardRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<LeaderboardGroupWithContext, HasLeaderboardContext> {
|
||||
|
||||
public LeaderboardRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<HasLeaderboardContext, ?>> resultReceivers) {
|
||||
super(LeaderboardGroupWithContext.class, HasLeaderboardContext.class, executor, resultReceivers);
|
||||
Collection<Processor<HasLeaderboardContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(LeaderboardGroupWithContext.class, HasLeaderboardContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -19,8 +19,8 @@ public class MarkPassingRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<HasTrackedRaceContext, HasMarkPassingContext> {
|
||||
|
||||
public MarkPassingRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<HasMarkPassingContext, ?>> resultReceivers) {
|
||||
super(HasTrackedRaceContext.class, HasMarkPassingContext.class, executor, resultReceivers);
|
||||
Collection<Processor<HasMarkPassingContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(HasTrackedRaceContext.class, HasMarkPassingContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -15,8 +15,8 @@ public class TrackedLegOfCompetitorRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<HasTrackedLegContext, HasTrackedLegOfCompetitorContext> {
|
||||
|
||||
public TrackedLegOfCompetitorRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<HasTrackedLegOfCompetitorContext, ?>> resultReceivers) {
|
||||
super(HasTrackedLegContext.class, HasTrackedLegOfCompetitorContext.class, executor, resultReceivers);
|
||||
Collection<Processor<HasTrackedLegOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(HasTrackedLegContext.class, HasTrackedLegOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -15,8 +15,8 @@ public class TrackedLegRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<HasTrackedRaceContext, HasTrackedLegContext> {
|
||||
|
||||
public TrackedLegRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<HasTrackedLegContext, ?>> resultReceivers) {
|
||||
super(HasTrackedRaceContext.class, HasTrackedLegContext.class, executor, resultReceivers);
|
||||
Collection<Processor<HasTrackedLegContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(HasTrackedRaceContext.class, HasTrackedLegContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -18,8 +18,8 @@ public class TrackedRaceRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<HasLeaderboardContext, HasTrackedRaceContext> {
|
||||
|
||||
public TrackedRaceRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<HasTrackedRaceContext, ?>> resultReceivers) {
|
||||
super(HasLeaderboardContext.class, HasTrackedRaceContext.class, executor, resultReceivers);
|
||||
Collection<Processor<HasTrackedRaceContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(HasLeaderboardContext.class, HasTrackedRaceContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+9
-14
@@ -46,20 +46,15 @@ public class TestDataRetrieverChainCreation {
|
||||
@Before
|
||||
public void initializeRetrieverChain() {
|
||||
dataRetrieverChainDefinition = new SimpleDataRetrieverChainDefinition<>((Class<Collection<Test_Regatta>>)(Class<?>) Collection.class, Test_HasLegOfCompetitorContext.class, "TestRetrieverChain");
|
||||
Class<Processor<Collection<Test_Regatta>, Test_Regatta>> regattaRetrieverClass = (Class<Processor<Collection<Test_Regatta>, Test_Regatta>>)(Class<?>) TestRegattaRetrievalProcessor.class;
|
||||
dataRetrieverChainDefinition.startWith(regattaRetrieverClass, Test_Regatta.class, "regatta");
|
||||
dataRetrieverChainDefinition.startWith(TestRegattaRetrievalProcessor.class, Test_Regatta.class, "regatta");
|
||||
|
||||
Class<Processor<Test_Regatta, Test_HasRaceContext>> raceRetrieverClass =
|
||||
(Class<Processor<Test_Regatta, Test_HasRaceContext>>)(Class<?>) TestRaceWithContextRetrievalProcessor.class;
|
||||
dataRetrieverChainDefinition.addAfter(regattaRetrieverClass,
|
||||
raceRetrieverClass,
|
||||
Test_HasRaceContext.class, "race");
|
||||
dataRetrieverChainDefinition.addAfter(TestRegattaRetrievalProcessor.class,
|
||||
TestRaceWithContextRetrievalProcessor.class,
|
||||
Test_HasRaceContext.class, "race");
|
||||
|
||||
Class<Processor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>> legRetrieverClass =
|
||||
(Class<Processor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>>)(Class<?>) TestLegOfCompetitorWithContextRetrievalProcessor.class;
|
||||
dataRetrieverChainDefinition.endWith(raceRetrieverClass,
|
||||
legRetrieverClass,
|
||||
Test_HasLegOfCompetitorContext.class, "legOfCompetitor");
|
||||
dataRetrieverChainDefinition.endWith(TestRaceWithContextRetrievalProcessor.class,
|
||||
TestLegOfCompetitorWithContextRetrievalProcessor.class,
|
||||
Test_HasLegOfCompetitorContext.class, "legOfCompetitor");
|
||||
}
|
||||
|
||||
@Before
|
||||
@@ -179,8 +174,8 @@ public class TestDataRetrieverChainCreation {
|
||||
dataRetrieverChainDefinition.startWith(regattaRetrieverClass, Test_Regatta.class, "regatta");
|
||||
|
||||
dataRetrieverChainDefinition.addAfter(regattaRetrieverClass,
|
||||
(Class<Processor<Test_Regatta, Test_HasRaceContext>>)(Class<?>) Processor.class,
|
||||
Test_HasRaceContext.class, "race");
|
||||
(Class<Processor<Test_Regatta, Test_HasRaceContext>>)(Class<?>) Processor.class,
|
||||
Test_HasRaceContext.class, "race");
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
|
||||
+1
-1
@@ -136,7 +136,7 @@ public class TestProcessorQuery {
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
Processor<Iterable<Number>, Number> retrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Number>, Number>((Class<Iterable<Number>>)(Class<?>) Iterable.class, Number.class,
|
||||
ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers) {
|
||||
ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 0) {
|
||||
@Override
|
||||
protected Iterable<Number> retrieveData(Iterable<Number> element) {
|
||||
return element;
|
||||
|
||||
+2
-2
@@ -72,7 +72,7 @@ public class TestRetrieverFilterProcessorChain {
|
||||
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,
|
||||
ConcurrencyTestsUtil.getExecutor(), resultReceivers) {
|
||||
ConcurrencyTestsUtil.getExecutor(), resultReceivers, 0) {
|
||||
@Override
|
||||
protected Iterable<Iterable<Integer>> retrieveData(Iterable<Iterable<Integer>> element) {
|
||||
return 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) {
|
||||
retrievalProcessor = new AbstractSimpleRetrievalProcessor<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;
|
||||
|
||||
+2
-2
@@ -15,8 +15,8 @@ import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasRace
|
||||
public class TestLegOfCompetitorWithContextRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>{
|
||||
|
||||
public TestLegOfCompetitorWithContextRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_HasLegOfCompetitorContext, ?>> resultReceivers) {
|
||||
super(Test_HasRaceContext.class, Test_HasLegOfCompetitorContext.class, executor, resultReceivers);
|
||||
public TestLegOfCompetitorWithContextRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_HasLegOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(Test_HasRaceContext.class, Test_HasLegOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -15,8 +15,8 @@ public class TestRaceWithContextRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<Test_Regatta, Test_HasRaceContext> {
|
||||
|
||||
public TestRaceWithContextRetrievalProcessor(ExecutorService executor,
|
||||
Collection<Processor<Test_HasRaceContext, ?>> resultReceivers) {
|
||||
super(Test_Regatta.class, Test_HasRaceContext.class, executor, resultReceivers);
|
||||
Collection<Processor<Test_HasRaceContext, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(Test_Regatta.class, Test_HasRaceContext.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -11,8 +11,8 @@ public class TestRegattaRetrievalProcessor extends
|
||||
AbstractSimpleRetrievalProcessor<Collection<Test_Regatta>, Test_Regatta> {
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public TestRegattaRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_Regatta, ?>> resultReceivers) {
|
||||
super((Class<Collection<Test_Regatta>>)(Class<?>) Collection.class, Test_Regatta.class, executor, resultReceivers);
|
||||
public TestRegattaRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_Regatta, ?>> resultReceivers, int retrievalLevel) {
|
||||
super((Class<Collection<Test_Regatta>>)(Class<?>) Collection.class, Test_Regatta.class, executor, resultReceivers, retrievalLevel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+6
-6
@@ -146,24 +146,24 @@ public class SimpleDataRetrieverChainBuilder<DataSourceType> implements DataRetr
|
||||
FilterCriterion<ResultType> filter = (FilterCriterion<ResultType>) filters.get(retrieverTypeIndex);
|
||||
|
||||
Class<Processor<?, ResultType>> retrieverType = (Class<Processor<?, ResultType>>)(Class<?>) dataRetrieverTypeWithInformation.getRetrieverType();
|
||||
return createRetriever(retrieverType, retrievedDataType, resultReceivers, filter);
|
||||
return createRetriever(retrieverType, retrievedDataType, resultReceivers, filter, retrieverTypeIndex);
|
||||
}
|
||||
|
||||
private <ResultType> Processor<?, ResultType> createRetriever(Class<Processor<?, ResultType>> retrieverType, Class<ResultType> retrievedDataType,
|
||||
Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter) {
|
||||
Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter, int retrieverTypeIndex) {
|
||||
Constructor<Processor<?, ResultType>> retrieverConstructor = null;
|
||||
try {
|
||||
retrieverConstructor = retrieverType.getConstructor(ExecutorService.class, Collection.class);
|
||||
retrieverConstructor = retrieverType.getConstructor(ExecutorService.class, Collection.class, int.class);
|
||||
} catch (NoSuchMethodException | SecurityException e) {
|
||||
throw new IllegalArgumentException("Couldn't get an usable constructor from the given retrieverType '"
|
||||
+ retrieverType.getSimpleName() + "'", e);
|
||||
}
|
||||
|
||||
return constructRetriever(retrieverConstructor, retrievedDataType, resultReceivers, filter);
|
||||
return constructRetriever(retrieverConstructor, retrievedDataType, resultReceivers, filter, retrieverTypeIndex);
|
||||
}
|
||||
|
||||
private <ResultType> Processor<?, ResultType> constructRetriever(Constructor<Processor<?, ResultType>> retrieverConstructor, Class<ResultType> retrievedDataType,
|
||||
Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter) {
|
||||
Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter, int retrieverTypeIndex) {
|
||||
try {
|
||||
Collection<Processor<ResultType, ?>> retrievalResultReceivers = resultReceivers;
|
||||
if (filter != null) {
|
||||
@@ -173,7 +173,7 @@ public class SimpleDataRetrieverChainBuilder<DataSourceType> implements DataRetr
|
||||
retrievalResultReceivers.add(filteringProcessor);
|
||||
}
|
||||
|
||||
return retrieverConstructor.newInstance(executor, retrievalResultReceivers);
|
||||
return retrieverConstructor.newInstance(executor, retrievalResultReceivers, retrieverTypeIndex);
|
||||
} catch (InstantiationException | IllegalAccessException |
|
||||
IllegalArgumentException | InvocationTargetException e) {
|
||||
throw new UnsupportedOperationException("Couldn't create a data retriever instance with the constructor "
|
||||
|
||||
+1
-1
@@ -105,7 +105,7 @@ public class SimpleDataRetrieverChainDefinition<DataSourceType, DataType> implem
|
||||
private <InputType, ResultType> void checkThatRetrieverHasUsableConstructor(
|
||||
Class<? extends Processor<InputType, ResultType>> retrieverType) {
|
||||
try {
|
||||
retrieverType.getConstructor(ExecutorService.class, Collection.class);
|
||||
retrieverType.getConstructor(ExecutorService.class, Collection.class, int.class);
|
||||
} catch (NoSuchMethodException | SecurityException e) {
|
||||
throw new IllegalArgumentException("Couldn't get an usable constructor from the given nextRetrieverType '"
|
||||
+ retrieverType.getSimpleName() + "'", e);
|
||||
|
||||
+1
-1
@@ -47,7 +47,7 @@ public abstract class AbstractParallelMultiDimensionalNestingGroupingProcessor<D
|
||||
|
||||
@Override
|
||||
protected ProcessorInstruction<GroupedDataEntry<DataType>> createInstruction(final DataType element) {
|
||||
return new ProcessorInstruction<GroupedDataEntry<DataType>>(this) {
|
||||
return new ProcessorInstruction<GroupedDataEntry<DataType>>(this, ProcessorInstructionPriority.Grouping) {
|
||||
@Override
|
||||
public GroupedDataEntry<DataType> computeResult() {
|
||||
return new GroupedDataEntry<DataType>(createCompoundKeyFor(element,
|
||||
|
||||
+14
-2
@@ -11,17 +11,29 @@ import com.sap.sse.datamining.components.Processor;
|
||||
public abstract class AbstractRetrievalProcessor<InputType, WorkingType, ResultType> extends
|
||||
AbstractPartitioningParallelProcessor<InputType, WorkingType, ResultType> {
|
||||
|
||||
private final int retrievalLevel;
|
||||
|
||||
private final Lock retrievedDataAmountLock;
|
||||
private int retrievedDataAmount;
|
||||
|
||||
public AbstractRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
|
||||
/**
|
||||
*
|
||||
* @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 AbstractRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType,
|
||||
ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers, int retrievalLevel) {
|
||||
super(inputType, resultType, executor, resultReceivers);
|
||||
this.retrievalLevel = retrievalLevel;
|
||||
retrievedDataAmountLock = new ReentrantLock();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement) {
|
||||
return new ProcessorInstruction<ResultType>(this) {
|
||||
return new ProcessorInstruction<ResultType>(this, ProcessorInstructionPriority.createRetrievalPriority(retrievalLevel)) {
|
||||
@Override
|
||||
public ResultType computeResult() {
|
||||
incrementRetrievedDataAmount();
|
||||
|
||||
+11
-2
@@ -8,8 +8,17 @@ import com.sap.sse.datamining.components.Processor;
|
||||
public abstract class AbstractSimpleRetrievalProcessor<InputType, ResultType> extends
|
||||
AbstractRetrievalProcessor<InputType, ResultType, ResultType> {
|
||||
|
||||
public AbstractSimpleRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
|
||||
super(inputType, resultType, executor, resultReceivers);
|
||||
/**
|
||||
*
|
||||
* @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
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParalle
|
||||
|
||||
@Override
|
||||
protected ProcessorInstruction<InputType> createInstruction(final InputType element) {
|
||||
return new ProcessorInstruction<InputType>(this) {
|
||||
return new ProcessorInstruction<InputType>(this, ProcessorInstructionPriority.Filtration) {
|
||||
@Override
|
||||
public InputType computeResult() {
|
||||
if (filterCriterion.matches(element)) {
|
||||
|
||||
+1
-1
@@ -31,7 +31,7 @@ public class ParallelGroupedElementsValueExtractionProcessor<DataType, FunctionR
|
||||
|
||||
@Override
|
||||
protected ProcessorInstruction<GroupedDataEntry<FunctionReturnType>> createInstruction(final GroupedDataEntry<DataType> element) {
|
||||
return new ProcessorInstruction<GroupedDataEntry<FunctionReturnType>>(this) {
|
||||
return new ProcessorInstruction<GroupedDataEntry<FunctionReturnType>>(this, ProcessorInstructionPriority.Extraction) {
|
||||
@Override
|
||||
public GroupedDataEntry<FunctionReturnType> computeResult() {
|
||||
FunctionReturnType value = extractionFunction.tryToInvoke(element.getDataEntry(), parameterProvider);
|
||||
|
||||
+44
-1
@@ -1,11 +1,49 @@
|
||||
package com.sap.sse.datamining.impl.components;
|
||||
|
||||
public abstract class ProcessorInstruction<ResultType> implements Runnable {
|
||||
public abstract class ProcessorInstruction<ResultType> implements Runnable, Comparable<ProcessorInstruction<?>> {
|
||||
|
||||
private final AbstractPartitioningParallelProcessor<?, ?, ResultType> processor;
|
||||
private final int priority;
|
||||
|
||||
/**
|
||||
* Creates an instruction with a priority of 0, which is the highest priority.<br />
|
||||
* 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)}
|
||||
* instead.
|
||||
*
|
||||
* @param processor The processor, that processes this instruction
|
||||
*/
|
||||
public ProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor) {
|
||||
this(processor, 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an instruction with the given {@linkplain ProcessorInstructionPriority priority}.
|
||||
*
|
||||
* @param processor The processor, that processes this instruction
|
||||
* @param priority The priority of this instruction
|
||||
*/
|
||||
public ProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
|
||||
ProcessorInstructionPriority priority) {
|
||||
this(processor, priority.asIntValue());
|
||||
}
|
||||
|
||||
/**
|
||||
* 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)}
|
||||
* instead.
|
||||
*
|
||||
* @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,
|
||||
int priority) {
|
||||
this.processor = processor;
|
||||
this.priority = priority;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -25,5 +63,10 @@ public abstract class ProcessorInstruction<ResultType> implements Runnable {
|
||||
}
|
||||
|
||||
protected abstract ResultType computeResult() throws Exception;
|
||||
|
||||
@Override
|
||||
public int compareTo(ProcessorInstruction<?> instruction) {
|
||||
return Integer.compare(priority, instruction.priority);
|
||||
}
|
||||
|
||||
}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
package com.sap.sse.datamining.impl.components;
|
||||
|
||||
public class ProcessorInstructionPriority {
|
||||
|
||||
public static final ProcessorInstructionPriority Aggregation = new ProcessorInstructionPriority(0);
|
||||
public static final ProcessorInstructionPriority Extraction = new ProcessorInstructionPriority(1);
|
||||
public static final ProcessorInstructionPriority Filtration = new ProcessorInstructionPriority(1);
|
||||
public static final ProcessorInstructionPriority Grouping = new ProcessorInstructionPriority(2);
|
||||
|
||||
private static final int retrievalBasePriority = Integer.MAX_VALUE;
|
||||
|
||||
/**
|
||||
* Creates a priority for the given <code>retrieverLevel</code>, where the level <code>0</code>
|
||||
* represents the first retriever in the chain.
|
||||
*
|
||||
* @param retrievalLevel The position of the retriever in its chain
|
||||
* @return A priority for the given <code>retrieverLevel</code>
|
||||
*/
|
||||
public static ProcessorInstructionPriority createRetrievalPriority(int retrievalLevel) {
|
||||
return new ProcessorInstructionPriority(retrievalBasePriority - retrievalLevel);
|
||||
}
|
||||
|
||||
private final int intValue;
|
||||
|
||||
private ProcessorInstructionPriority(int intValue) {
|
||||
this.intValue = intValue;
|
||||
}
|
||||
|
||||
public int asIntValue() {
|
||||
return intValue;
|
||||
}
|
||||
|
||||
}
|
||||
+2
-1
@@ -9,6 +9,7 @@ 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.ProcessorInstructionPriority;
|
||||
|
||||
public abstract class AbstractParallelStoringAggregationProcessor<InputType, AggregatedType>
|
||||
extends AbstractSimpleParallelProcessor<InputType, AggregatedType> {
|
||||
@@ -28,7 +29,7 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
|
||||
|
||||
@Override
|
||||
protected ProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
|
||||
return new ProcessorInstruction<AggregatedType>(this) {
|
||||
return new ProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
|
||||
@Override
|
||||
public AggregatedType computeResult() {
|
||||
storeLock.lock();
|
||||
|
||||
Reference in new issue
Block a user