Removed the deprecated function retrievers and adjusted the

DataMiningActivator and the tests
This commit is contained in:
Lennart Hensler committed 2014-03-27 18:27:14 +01:00
1 parent e31a0d6c95
commit df43564aaa
13 files changed
+104 -363

No files matched your search

@@ -17,10 +17,7 @@ import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.FunctionProvider;
import com.sap.sse.datamining.functions.FunctionRegistry;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.impl.functions.PartitionParallelExternalFunctionRetriever;
import com.sap.sse.datamining.impl.functions.PartitioningParallelMarkedFunctionRetriever;
import com.sap.sse.datamining.impl.functions.RegistryFunctionsProvider;
import com.sap.sse.datamining.impl.functions.SimpleFunctionRegistry;
import com.sap.sse.datamining.shared.dto.FunctionDTO;
@@ -32,6 +29,7 @@ import com.sap.sse.datamining.test.function.test_classes.ExtendingInterface;
import com.sap.sse.datamining.test.function.test_classes.ExternalLibraryClass;
import com.sap.sse.datamining.test.function.test_classes.MarkedContainer;
import com.sap.sse.datamining.test.function.test_classes.SimpleClassWithMarkedMethods;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
import com.sap.sse.datamining.test.util.TestsUtil;
@@ -43,22 +41,20 @@ public class TestFunctionProvider {
@Before
public void initializeFunctionRegistry() {
functionRegistry = new SimpleFunctionRegistry();
functionRegistry = new SimpleFunctionRegistry(ConcurrencyTestsUtil.getExecutor());
Collection<Class<?>> classesToScan = new HashSet<>();
classesToScan.add(SimpleClassWithMarkedMethods.class);
classesToScan.add(DataTypeWithContext.class);
classesToScan.add(DataTypeWithContextProcessor.class);
classesToScan.add(ExtendingInterface.class);
classesToScan.add(MarkedContainer.class);
classesToScan.add(ContainerElement.class);
ParallelFunctionRetriever markedFunctionRetriever = new PartitioningParallelMarkedFunctionRetriever(classesToScan, FunctionTestsUtil.getExecutor());
functionRegistry.registerFunctionsRetrievedBy(markedFunctionRetriever);
Collection<Class<?>> internalClassesToScan = new HashSet<>();
internalClassesToScan.add(SimpleClassWithMarkedMethods.class);
internalClassesToScan.add(DataTypeWithContext.class);
internalClassesToScan.add(DataTypeWithContextProcessor.class);
internalClassesToScan.add(ExtendingInterface.class);
internalClassesToScan.add(MarkedContainer.class);
internalClassesToScan.add(ContainerElement.class);
functionRegistry.registerAllWithInternalFunctionPolicy(internalClassesToScan);
Collection<Class<?>> externalClasses = new HashSet<>();
externalClasses.add(ExternalLibraryClass.class);
ParallelFunctionRetriever externalFunctionRetriever = new PartitionParallelExternalFunctionRetriever(externalClasses, FunctionTestsUtil.getExecutor());
functionRegistry.registerFunctionsRetrievedBy(externalFunctionRetriever);
Collection<Class<?>> externalClassesToScan = new HashSet<>();
externalClassesToScan.add(ExternalLibraryClass.class);
functionRegistry.registerAllWithExternalFunctionPolicy(externalClassesToScan);
}
@Test
@@ -1,84 +0,0 @@
package com.sap.sse.datamining.impl.function;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.concurrent.ExecutionException;
import org.junit.Test;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.impl.functions.FunctionRetrievalWorker;
import com.sap.sse.datamining.impl.functions.PartitioningParallelMarkedFunctionRetriever;
import com.sap.sse.datamining.test.function.test_classes.ExternalLibraryClass;
import com.sap.sse.datamining.test.function.test_classes.SimpleClassWithMarkedMethods;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
import com.sap.sse.datamining.test.util.OpenDataReceiver;
public class TestFunctionRetrievers {
@Test
public void testMarkedFunctionRetrievalWorker() {
Collection<Class<?>> classesToScan = new ArrayList<>();
classesToScan.add(SimpleClassWithMarkedMethods.class);
OpenDataReceiver<Collection<Function<?>>> receiver = new OpenDataReceiver<>();
FunctionRetrievalWorker worker = FunctionRetrievalWorker.Util.createMarkedFunctionRetrievalWorker(classesToScan, receiver);
Collection<Function<?>> expectedFunctions = FunctionTestsUtil.getMarkedMethodsOfSimpleClassWithMarkedMethod();
worker.run();
assertThat(receiver.result, is(expectedFunctions));
}
@Test
public void testExternalFunctionRetrievalWorker() {
Collection<Class<?>> classesToScan = new ArrayList<>();
classesToScan.add(ExternalLibraryClass.class);
OpenDataReceiver<Collection<Function<?>>> receiver = new OpenDataReceiver<>();
FunctionRetrievalWorker worker = FunctionRetrievalWorker.Util.createExternalFunctionRetrievalWorker(classesToScan, receiver);
Collection<Function<?>> expectedFunctions = FunctionTestsUtil.getMethodsOfExternalLibraryClass();
worker.run();
assertThat(receiver.result, is(expectedFunctions));
}
@Test
public void testFunctionRetrievalWithManyClassesToScan() {
Collection<Class<?>> classesToScan = getManyClassesToScan();
classesToScan.add(SimpleClassWithMarkedMethods.class);
assertThat("Not enough classes to scan for this test.", classesToScan.size() >= getMaximumWorkerAmount(), is(true));
ParallelFunctionRetriever functionRetriever = new PartitioningParallelMarkedFunctionRetriever(classesToScan , FunctionTestsUtil.getExecutor());
Collection<Function<?>> expectedFunctions = FunctionTestsUtil.getMarkedMethodsOfSimpleClassWithMarkedMethod();
try {
Collection<Function<?>> retrievedFunctions = new HashSet<>(functionRetriever.start(null).get());
assertThat(retrievedFunctions, is(expectedFunctions));
} catch (InterruptedException | ExecutionException e) {
fail("Error during the function retrieving");
}
}
private int getMaximumWorkerAmount() {
return (int) (FunctionTestsUtil.getExecutor().getCorePoolSize() * 0.5);
}
private Collection<Class<?>> getManyClassesToScan() {
Collection<Class<?>> classesToScan = new ArrayList<>();
classesToScan.add(Math.class);
classesToScan.add(Integer.class);
classesToScan.add(String.class);
classesToScan.add(Double.class);
classesToScan.add(Long.class);
classesToScan.add(Float.class);
classesToScan.add(ExternalLibraryClass.class);
return classesToScan;
}
}
@@ -11,14 +11,12 @@ import org.junit.Test;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.FunctionRegistry;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.impl.functions.PartitionParallelExternalFunctionRetriever;
import com.sap.sse.datamining.impl.functions.PartitioningParallelMarkedFunctionRetriever;
import com.sap.sse.datamining.impl.functions.SimpleFunctionRegistry;
import com.sap.sse.datamining.test.function.test_classes.DataTypeWithContext;
import com.sap.sse.datamining.test.function.test_classes.DataTypeWithContextProcessor;
import com.sap.sse.datamining.test.function.test_classes.ExternalLibraryClass;
import com.sap.sse.datamining.test.function.test_classes.SimpleClassWithMarkedMethods;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
public class TestSimpleFunctionRegistryAccess {
@@ -27,19 +25,17 @@ public class TestSimpleFunctionRegistryAccess {
@Before
public void initializeTheFunctionRegistry() {
functionRegistry = new SimpleFunctionRegistry();
functionRegistry = new SimpleFunctionRegistry(ConcurrencyTestsUtil.getExecutor());
Collection<Class<?>> classesToScan = new HashSet<>();
classesToScan.add(SimpleClassWithMarkedMethods.class);
classesToScan.add(DataTypeWithContext.class);
classesToScan.add(DataTypeWithContextProcessor.class);
ParallelFunctionRetriever markedFunctionRetriever = new PartitioningParallelMarkedFunctionRetriever(classesToScan, FunctionTestsUtil.getExecutor());
functionRegistry.registerFunctionsRetrievedBy(markedFunctionRetriever);
Collection<Class<?>> internalClassesToScan = new HashSet<>();
internalClassesToScan.add(SimpleClassWithMarkedMethods.class);
internalClassesToScan.add(DataTypeWithContext.class);
internalClassesToScan.add(DataTypeWithContextProcessor.class);
functionRegistry.registerAllWithInternalFunctionPolicy(internalClassesToScan);
Collection<Class<?>> externalClasses = new HashSet<>();
externalClasses.add(ExternalLibraryClass.class);
ParallelFunctionRetriever externalFunctionRetriever = new PartitionParallelExternalFunctionRetriever(externalClasses, FunctionTestsUtil.getExecutor());
functionRegistry.registerFunctionsRetrievedBy(externalFunctionRetriever);
Collection<Class<?>> externalClassesToScan = new HashSet<>();
externalClassesToScan.add(ExternalLibraryClass.class);
functionRegistry.registerAllWithExternalFunctionPolicy(externalClassesToScan);
}
@Test
@@ -13,12 +13,10 @@ import org.junit.Test;
import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.FunctionRegistry;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.impl.functions.PartitionParallelExternalFunctionRetriever;
import com.sap.sse.datamining.impl.functions.PartitioningParallelMarkedFunctionRetriever;
import com.sap.sse.datamining.impl.functions.SimpleFunctionRegistry;
import com.sap.sse.datamining.test.function.test_classes.ExternalLibraryClass;
import com.sap.sse.datamining.test.function.test_classes.SimpleClassWithMarkedMethods;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
public class TestSimpleFunctionRegistryRegistrations {
@@ -27,7 +25,7 @@ public class TestSimpleFunctionRegistryRegistrations {
public void testSimpleRegistration() {
Method dimension = FunctionTestsUtil.getMethodFromSimpleClassWithMarkedMethod("dimension");
FunctionRegistry registry = new SimpleFunctionRegistry();
FunctionRegistry registry = new SimpleFunctionRegistry(ConcurrencyTestsUtil.getExecutor());
registry.register(dimension);
Set<Function<?>> expectedRegisteredFunctionsAsSet = new HashSet<>();
@@ -39,8 +37,8 @@ public class TestSimpleFunctionRegistryRegistrations {
@Test
public void testRegistrationByFunctionRetrievers() {
FunctionRegistry registry = new SimpleFunctionRegistry();
registerMethodsOfTestClassesViaFunctionRetrieversTo(registry);
FunctionRegistry registry = new SimpleFunctionRegistry(ConcurrencyTestsUtil.getExecutor());
registerMethodsOfTestClasses(registry);
Set<Function<?>> expectedRegisteredFunctionsAsSet = new HashSet<>();
expectedRegisteredFunctionsAsSet.addAll(FunctionTestsUtil.getMarkedMethodsOfSimpleClassWithMarkedMethod());
@@ -49,23 +47,21 @@ public class TestSimpleFunctionRegistryRegistrations {
assertThat(registry.getAllFunctions(), is(expectedRegisteredFunctions));
}
private void registerMethodsOfTestClassesViaFunctionRetrieversTo(FunctionRegistry registry) {
Collection<Class<?>> classesToScan = new HashSet<>();
classesToScan.add(SimpleClassWithMarkedMethods.class);
ParallelFunctionRetriever markedFunctionRetriever = new PartitioningParallelMarkedFunctionRetriever(classesToScan, FunctionTestsUtil.getExecutor());
registry.registerFunctionsRetrievedBy(markedFunctionRetriever);
private void registerMethodsOfTestClasses(FunctionRegistry registry) {
Collection<Class<?>> internalClassesToScan = new HashSet<>();
internalClassesToScan.add(SimpleClassWithMarkedMethods.class);
registry.registerAllWithInternalFunctionPolicy(internalClassesToScan);
Collection<Class<?>> externalClasses = new HashSet<>();
externalClasses.add(ExternalLibraryClass.class);
ParallelFunctionRetriever externalFunctionRetriever = new PartitionParallelExternalFunctionRetriever(externalClasses, FunctionTestsUtil.getExecutor());
registry.registerFunctionsRetrievedBy(externalFunctionRetriever);
Collection<Class<?>> externalClassesToScan = new HashSet<>();
externalClassesToScan.add(ExternalLibraryClass.class);
registry.registerAllWithExternalFunctionPolicy(externalClassesToScan);
}
@Test
public void testTheMultipleRegistrationOfTheSameFunction() {
Method dimension = FunctionTestsUtil.getMethodFromSimpleClassWithMarkedMethod("dimension");
FunctionRegistry registry = new SimpleFunctionRegistry();
FunctionRegistry registry = new SimpleFunctionRegistry(ConcurrencyTestsUtil.getExecutor());
registry.register(dimension);
registry.register(dimension);
registry.register(dimension);
@@ -7,8 +7,9 @@ public interface FunctionRegistry {
public void register(Method method);
public void register(Function<?> function);
public void registerFunctionsRetrievedBy(ParallelFunctionRetriever functionRetriever);
public void registerAll(Iterable<Function<?>> functions);
public void registerAllWithInternalFunctionPolicy(Collection<Class<?>> classesToScan);
public void registerAllWithExternalFunctionPolicy(Collection<Class<?>> externalClassesToScan);
public Collection<Function<?>> getAllFunctions();
public Iterable<Function<?>> getFunctionsOf(Class<?> declaringClass);
@@ -1,7 +0,0 @@
package com.sap.sse.datamining.functions;
import com.sap.sse.datamining.components.ParallelDataRetriever;
public interface ParallelFunctionRetriever extends ParallelDataRetriever<Function<?>> {
}
@@ -19,10 +19,7 @@ import com.sap.sse.datamining.DataMiningServer;
import com.sap.sse.datamining.functions.ClassesWithFunctionsService;
import com.sap.sse.datamining.functions.FunctionProvider;
import com.sap.sse.datamining.functions.FunctionRegistry;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.impl.functions.PartitionParallelExternalFunctionRetriever;
import com.sap.sse.datamining.impl.functions.PartitioningParallelMarkedFunctionRetriever;
import com.sap.sse.datamining.impl.functions.RegistryFunctionsProvider;
import com.sap.sse.datamining.impl.functions.SimpleFunctionRegistry;
@@ -42,7 +39,6 @@ public class DataMiningActivator implements BundleActivator {
public void start(BundleContext context) throws Exception {
DataMiningActivator.context = context;
serviceReferences = getAllClassesWithMarkedMethodsServices();
stringMessages = DataMiningStringMessages.Util.getDefaultStringMessages();
executor = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 60, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>());
@@ -59,7 +55,7 @@ public class DataMiningActivator implements BundleActivator {
}
private FunctionRegistry createAndBuildFunctionRegistry() {
FunctionRegistry functionRegistry = new SimpleFunctionRegistry();
FunctionRegistry functionRegistry = new SimpleFunctionRegistry(getExecutor());
Set<Class<?>> internalClassesWithMarkedMethods = new HashSet<>();
Set<Class<?>> externalLibraryClasses = new HashSet<>();
@@ -77,7 +73,8 @@ public class DataMiningActivator implements BundleActivator {
}
}
registerFunctions(internalClassesWithMarkedMethods, externalLibraryClasses, functionRegistry);
functionRegistry.registerAllWithInternalFunctionPolicy(internalClassesWithMarkedMethods);
functionRegistry.registerAllWithExternalFunctionPolicy(externalLibraryClasses);
return functionRegistry;
}
@@ -110,17 +107,6 @@ public class DataMiningActivator implements BundleActivator {
return specificServiceReferences;
}
private void registerFunctions(Set<Class<?>> internalClassesWithMarkedMethods,
Set<Class<?>> externalLibraryClasses, FunctionRegistry functionRegistry) {
ParallelFunctionRetriever internalMarkedFunctionsRetriever = new PartitioningParallelMarkedFunctionRetriever(
internalClassesWithMarkedMethods, getExecutor());
functionRegistry.registerFunctionsRetrievedBy(internalMarkedFunctionsRetriever);
ParallelFunctionRetriever externalLibraryFunctionsRetriever = new PartitionParallelExternalFunctionRetriever(
externalLibraryClasses, getExecutor());
functionRegistry.registerFunctionsRetrievedBy(externalLibraryFunctionsRetriever);
}
@Override
public void stop(BundleContext context) throws Exception {
for (ServiceReference<?> serviceReference : serviceReferences) {
@@ -1,76 +0,0 @@
package com.sap.sse.datamining.impl.functions;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadPoolExecutor;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.impl.components.deprecated.AbstractParallelComponent;
import com.sap.sse.datamining.workers.ComponentWorker;
public abstract class AbstractPartitioningParallelFunctionRetriever
extends AbstractParallelComponent<Void, Collection<Function<?>>>
implements ParallelFunctionRetriever {
private List<Class<?>> classesToScan;
public AbstractPartitioningParallelFunctionRetriever(Collection<Class<?>> classesToScan, ThreadPoolExecutor executor) {
super(executor);
this.classesToScan = new ArrayList<>(classesToScan);
}
@Override
protected void setUpWorkersFor(Void data) {
final int workerAmount = calculateMaximumAmountOfWorkers();
if (workerAmount >= classesToScan.size()) {
simpleWorkerSetUp();
} else {
final int partitionSize = calculatePartitionSize(workerAmount);
partitioningWorkerSetUp(partitionSize);
}
}
public int calculatePartitionSize(final int workerAmount) {
return (int) Math.ceil((double) classesToScan.size() / workerAmount);
}
public int calculateMaximumAmountOfWorkers() {
return (int) (getExecutor().getCorePoolSize() * 0.5);
}
private void simpleWorkerSetUp() {
for (Class<?> classToScan : classesToScan) {
addWorker(createWorker(classToScan));
}
}
private ComponentWorker<Collection<Function<?>>> createWorker(Class<?> classToScan) {
Collection<Class<?>> classToScanAsCollection = new HashSet<>();
classToScanAsCollection.add(classToScan);
return createWorker(classToScanAsCollection);
}
private void partitioningWorkerSetUp(int partitionSize) {
for (int i = 0; i < classesToScan.size(); i += partitionSize) {
List<Class<?>> partition = classesToScan.subList(i, i + Math.min(partitionSize, classesToScan.size() - i));
addWorker(createWorker(partition));
}
}
protected abstract FunctionRetrievalWorker createWorker(Iterable<Class<?>> classesToScan);
@Override
protected Collection<Function<?>> finalizeData() {
Collection<Function<?>> data = new HashSet<>();
for (Collection<Function<?>> results : getResults()) {
data.addAll(results);
}
return Collections.unmodifiableCollection(new CopyOnWriteArrayList<Function<?>>(data));
}
}
@@ -1,46 +0,0 @@
package com.sap.sse.datamining.impl.functions;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.HashSet;
import com.sap.sse.datamining.components.FilterCriteria;
import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.impl.workers.AbstractComponentWorker;
public class FilteringFunctionRetrievalWorker extends AbstractComponentWorker<Collection<Function<?>>> implements
FunctionRetrievalWorker {
private Iterable<Class<?>> classesToScan;
private FilterCriteria<Method> filter;
@Override
protected Collection<Function<?>> doWork() {
Collection<Function<?>> functions = new HashSet<>();
for (Class<?> classToScan : classesToScan) {
functions.addAll(getFilteredMethodsFrom(classToScan));
}
return functions;
}
private Collection<Function<?>> getFilteredMethodsFrom(Class<?> classToScan) {
Collection<Function<?>> functions = new HashSet<>();
for (Method method : classToScan.getMethods()) {
if (filter != null && filter.matches(method)) {
functions.add(FunctionFactory.createMethodWrappingFunction(method));
}
}
return functions;
}
@Override
public void setSource(Iterable<Class<?>> classesToScan) {
this.classesToScan = classesToScan;
}
public void setFilter(FilterCriteria<Method> filterCriteria) {
this.filter = filterCriteria;
}
}
@@ -1,43 +0,0 @@
package com.sap.sse.datamining.impl.functions;
import java.lang.reflect.Method;
import java.util.Collection;
import com.sap.sse.datamining.components.FilterCriteria;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.impl.criterias.CompoundFilterCriteria;
import com.sap.sse.datamining.impl.criterias.OrCompoundFilterCriteria;
import com.sap.sse.datamining.workers.DataRetrievalWorker;
import com.sap.sse.datamining.workers.WorkReceiver;
public interface FunctionRetrievalWorker extends DataRetrievalWorker<Iterable<Class<?>>, Function<?>> {
public static final class Util {
public static FunctionRetrievalWorker createMarkedFunctionRetrievalWorker(Iterable<Class<?>> classesToScan, WorkReceiver<Collection<Function<?>>> receiver) {
FilteringFunctionRetrievalWorker worker = new FilteringFunctionRetrievalWorker();
worker.setSource(classesToScan);
worker.setReceiver(receiver);
CompoundFilterCriteria<Method> filterCriteria = new OrCompoundFilterCriteria<>();
filterCriteria.addCriteria(new MethodIsCorrectDimensionFilterCriteria());
filterCriteria.addCriteria(new MethodIsCorrectSideEffectFreeValueFilterCriteria());
worker.setFilter(filterCriteria);
return worker;
}
public static FunctionRetrievalWorker createExternalFunctionRetrievalWorker(Iterable<Class<?>> classesToScan, WorkReceiver<Collection<Function<?>>> receiver) {
FilteringFunctionRetrievalWorker worker = new FilteringFunctionRetrievalWorker();
worker.setSource(classesToScan);
worker.setReceiver(receiver);
FilterCriteria<Method> filterCriteria = new MethodIsCorrectExternalFunctionFilterCriteria();
worker.setFilter(filterCriteria);
return worker;
}
}
}
@@ -1,17 +0,0 @@
package com.sap.sse.datamining.impl.functions;
import java.util.Collection;
import java.util.concurrent.ThreadPoolExecutor;
public class PartitionParallelExternalFunctionRetriever extends AbstractPartitioningParallelFunctionRetriever {
public PartitionParallelExternalFunctionRetriever(Collection<Class<?>> externalClasses, ThreadPoolExecutor executor) {
super(externalClasses, executor);
}
@Override
protected FunctionRetrievalWorker createWorker(Iterable<Class<?>> classesToScan) {
return FunctionRetrievalWorker.Util.createExternalFunctionRetrievalWorker(classesToScan, this);
}
}
@@ -1,17 +0,0 @@
package com.sap.sse.datamining.impl.functions;
import java.util.Collection;
import java.util.concurrent.ThreadPoolExecutor;
public class PartitioningParallelMarkedFunctionRetriever extends AbstractPartitioningParallelFunctionRetriever {
public PartitioningParallelMarkedFunctionRetriever(Collection<Class<?>> classesToScan, ThreadPoolExecutor executor) {
super(classesToScan, executor);
}
@Override
protected FunctionRetrievalWorker createWorker(Iterable<Class<?>> classesToScan) {
return FunctionRetrievalWorker.Util.createMarkedFunctionRetrievalWorker(classesToScan, this);
}
}
@@ -1,39 +1,62 @@
package com.sap.sse.datamining.impl.functions;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.functions.FunctionRegistry;
import com.sap.sse.datamining.functions.ParallelFunctionRetriever;
import com.sap.sse.datamining.functions.FunctionRetrievalProcessor;
public class SimpleFunctionRegistry implements FunctionRegistry {
private static final Logger LOGGER = Logger.getLogger(SimpleFunctionRegistry.class.getName());
private final ExecutorService executor;
private final Processor<Collection<Function<?>>> functionsReceiver;
private final Map<Class<?>, Collection<Function<?>>> functionsMappedByDeclaringClass;
private final Map<Class<?>, Collection<Function<?>>> dimensionsMappedByDeclaringClass;
public SimpleFunctionRegistry() {
public SimpleFunctionRegistry(ExecutorService executor) {
this.executor = executor;
functionsReceiver = new FunctionsReceiver();
functionsMappedByDeclaringClass = new HashMap<>();
dimensionsMappedByDeclaringClass = new HashMap<>();
}
@Override
public void registerFunctionsRetrievedBy(ParallelFunctionRetriever functionRetriever) {
public void registerAllWithInternalFunctionPolicy(Collection<Class<?>> internalClassesToScan) {
FunctionRetrievalProcessor internalFunctionsRetriever = new InternalFunctionRetrievalProcessor(executor, Arrays.asList(functionsReceiver));
internalFunctionsRetriever.onElement(internalClassesToScan);
try {
Collection<Function<?>> functions = functionRetriever.start(null).get();
registerAll(functions);
} catch (InterruptedException | ExecutionException exception) {
LOGGER.log(Level.SEVERE, "Error during the function retrieving. Data-Mining may not work.", exception);
internalFunctionsRetriever.finish();
} catch (InterruptedException exception) {
LOGGER.log(Level.SEVERE, "The functions retrieval got interrupted. Data-Mining won't work!", exception);
}
}
@Override
public void registerAllWithExternalFunctionPolicy(Collection<Class<?>> externalClassesToScan) {
FunctionRetrievalProcessor externalFunctionsRetriever = new ExternalFunctionRetrievalProcessor(executor, Arrays.asList(functionsReceiver));
externalFunctionsRetriever.onElement(externalClassesToScan);
try {
externalFunctionsRetriever.finish();
} catch (InterruptedException exception) {
LOGGER.log(Level.SEVERE, "The functions retrieval got interrupted. Data-Mining won't work!", exception);
}
}
@@ -101,4 +124,37 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
return registeredMethods;
}
private class FunctionsReceiver implements Processor<Collection<Function<?>>> {
private final Lock functionsRegistrationLock;
public FunctionsReceiver() {
functionsRegistrationLock = new ReentrantLock();
}
@Override
public void onElement(Collection<Function<?>> element) {
functionsRegistrationLock.lock();
try {
SimpleFunctionRegistry.this.registerAll(element);
} finally {
functionsRegistrationLock.unlock();
}
}
@Override
public void finish() throws InterruptedException {
}
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
}
}