You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samoa.apache.org by gd...@apache.org on 2015/06/18 13:25:48 UTC
[07/41] incubator-samoa git commit: SAMOA-9: Change namespace to
apache.org. Fix #26
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleStream.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleStream.java b/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleStream.java
deleted file mode 100644
index 109e927..0000000
--- a/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleStream.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * To change this template, choose Tools | Templates
- * and open the template in the editor.
- */
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import java.util.LinkedList;
-import java.util.List;
-
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-import com.yahoo.labs.samoa.core.ContentEvent;
-import com.yahoo.labs.samoa.topology.AbstractStream;
-import com.yahoo.labs.samoa.topology.IProcessingItem;
-import com.yahoo.labs.samoa.utils.StreamDestination;
-
-/**
- *
- * @author abifet
- */
-class SimpleStream extends AbstractStream {
- private List<StreamDestination> destinations;
- private int maxCounter;
- private int eventCounter;
-
- SimpleStream(IProcessingItem sourcePi) {
- super(sourcePi);
- this.destinations = new LinkedList<>();
- this.eventCounter = 0;
- this.maxCounter = 1;
- }
-
- private int getNextCounter() {
- if (maxCounter > 0 && eventCounter >= maxCounter)
- eventCounter = 0;
- this.eventCounter++;
- return this.eventCounter;
- }
-
- @Override
- public void put(ContentEvent event) {
- this.put(event, this.getNextCounter());
- }
-
- private void put(ContentEvent event, int counter) {
- SimpleProcessingItem pi;
- int parallelism;
- for (StreamDestination destination : destinations) {
- pi = (SimpleProcessingItem) destination.getProcessingItem();
- parallelism = destination.getParallelism();
- switch (destination.getPartitioningScheme()) {
- case SHUFFLE:
- pi.processEvent(event, counter % parallelism);
- break;
- case GROUP_BY_KEY:
- HashCodeBuilder hb = new HashCodeBuilder();
- hb.append(event.getKey());
- int key = hb.build() % parallelism;
- pi.processEvent(event, key);
- break;
- case BROADCAST:
- for (int p = 0; p < parallelism; p++) {
- pi.processEvent(event, p);
- }
- break;
- }
- }
- }
-
- public void addDestination(StreamDestination destination) {
- this.destinations.add(destination);
- if (maxCounter <= 0)
- maxCounter = 1;
- maxCounter *= destination.getParallelism();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleTopology.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleTopology.java b/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleTopology.java
deleted file mode 100644
index 5ffa09e..0000000
--- a/samoa-local/src/main/java/com/yahoo/labs/samoa/topology/impl/SimpleTopology.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * To change this template, choose Tools | Templates
- * and open the template in the editor.
- */
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import com.yahoo.labs.samoa.topology.AbstractTopology;
-
-public class SimpleTopology extends AbstractTopology {
- SimpleTopology(String name) {
- super(name);
- }
-
- public void run() {
- if (this.getEntranceProcessingItems() == null)
- throw new IllegalStateException("You need to set entrance PI before running the topology.");
- if (this.getEntranceProcessingItems().size() != 1)
- throw new IllegalStateException("SimpleTopology supports 1 entrance PI only. Number of entrance PIs is "
- + this.getEntranceProcessingItems().size());
-
- SimpleEntranceProcessingItem entrancePi = (SimpleEntranceProcessingItem) this.getEntranceProcessingItems()
- .toArray()[0];
- entrancePi.getProcessor().onCreate(0); // id=0 as it is not used in simple mode
- entrancePi.startSendingEvents();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/LocalDoTask.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/LocalDoTask.java b/samoa-local/src/main/java/org/apache/samoa/LocalDoTask.java
new file mode 100644
index 0000000..0a8c3d0
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/LocalDoTask.java
@@ -0,0 +1,90 @@
+package org.apache.samoa;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.tasks.Task;
+import org.apache.samoa.topology.impl.SimpleComponentFactory;
+import org.apache.samoa.topology.impl.SimpleEngine;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.javacliparser.ClassOption;
+import com.github.javacliparser.FlagOption;
+import com.github.javacliparser.IntOption;
+import com.github.javacliparser.Option;
+
+/**
+ * The Class DoTask.
+ */
+public class LocalDoTask {
+
+ // TODO: clean up this class for helping ML Developer in SAMOA
+ // TODO: clean up code from storm-impl
+
+ // It seems that the 3 extra options are not used.
+ // Probably should remove them
+ private static final String SUPPRESS_STATUS_OUT_MSG = "Suppress the task status output. Normally it is sent to stderr.";
+ private static final String SUPPRESS_RESULT_OUT_MSG = "Suppress the task result output. Normally it is sent to stdout.";
+ private static final String STATUS_UPDATE_FREQ_MSG = "Wait time in milliseconds between status updates.";
+ private static final Logger logger = LoggerFactory.getLogger(LocalDoTask.class);
+
+ /**
+ * The main method.
+ *
+ * @param args
+ * the arguments
+ */
+ public static void main(String[] args) {
+
+ // ArrayList<String> tmpArgs = new ArrayList<String>(Arrays.asList(args));
+
+ // args = tmpArgs.toArray(new String[0]);
+
+ FlagOption suppressStatusOutOpt = new FlagOption("suppressStatusOut", 'S', SUPPRESS_STATUS_OUT_MSG);
+
+ FlagOption suppressResultOutOpt = new FlagOption("suppressResultOut", 'R', SUPPRESS_RESULT_OUT_MSG);
+
+ IntOption statusUpdateFreqOpt = new IntOption("statusUpdateFrequency", 'F', STATUS_UPDATE_FREQ_MSG, 1000, 0,
+ Integer.MAX_VALUE);
+
+ Option[] extraOptions = new Option[] { suppressStatusOutOpt, suppressResultOutOpt, statusUpdateFreqOpt };
+
+ StringBuilder cliString = new StringBuilder();
+ for (String arg : args) {
+ cliString.append(" ").append(arg);
+ }
+ logger.debug("Command line string = {}", cliString.toString());
+ System.out.println("Command line string = " + cliString.toString());
+
+ Task task;
+ try {
+ task = ClassOption.cliStringToObject(cliString.toString(), Task.class, extraOptions);
+ logger.info("Successfully instantiating {}", task.getClass().getCanonicalName());
+ } catch (Exception e) {
+ logger.error("Fail to initialize the task", e);
+ System.out.println("Fail to initialize the task" + e);
+ return;
+ }
+ task.setFactory(new SimpleComponentFactory());
+ task.init();
+ SimpleEngine.submitTopology(task.getTopology());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleComponentFactory.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleComponentFactory.java b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleComponentFactory.java
new file mode 100644
index 0000000..c18b72d
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleComponentFactory.java
@@ -0,0 +1,53 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.core.EntranceProcessor;
+import org.apache.samoa.core.Processor;
+import org.apache.samoa.topology.ComponentFactory;
+import org.apache.samoa.topology.EntranceProcessingItem;
+import org.apache.samoa.topology.IProcessingItem;
+import org.apache.samoa.topology.ProcessingItem;
+import org.apache.samoa.topology.Stream;
+import org.apache.samoa.topology.Topology;
+
+public class SimpleComponentFactory implements ComponentFactory {
+
+ public ProcessingItem createPi(Processor processor, int paralellism) {
+ return new SimpleProcessingItem(processor, paralellism);
+ }
+
+ public ProcessingItem createPi(Processor processor) {
+ return this.createPi(processor, 1);
+ }
+
+ public EntranceProcessingItem createEntrancePi(EntranceProcessor processor) {
+ return new SimpleEntranceProcessingItem(processor);
+ }
+
+ public Stream createStream(IProcessingItem sourcePi) {
+ return new SimpleStream(sourcePi);
+ }
+
+ public Topology createTopology(String topoName) {
+ return new SimpleTopology(topoName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEngine.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEngine.java b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEngine.java
new file mode 100644
index 0000000..06d03d3
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEngine.java
@@ -0,0 +1,37 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.topology.Topology;
+
+public class SimpleEngine {
+
+ public static void submitTopology(Topology topology) {
+ SimpleTopology simpleTopology = (SimpleTopology) topology;
+ simpleTopology.run();
+ // runs until completion
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItem.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItem.java b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItem.java
new file mode 100644
index 0000000..729ad31
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItem.java
@@ -0,0 +1,33 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.core.EntranceProcessor;
+import org.apache.samoa.topology.LocalEntranceProcessingItem;
+
+class SimpleEntranceProcessingItem extends LocalEntranceProcessingItem {
+ public SimpleEntranceProcessingItem(EntranceProcessor processor) {
+ super(processor);
+ }
+
+ // The default waiting time when there is no available events is 100ms
+ // Override waitForNewEvents() to change it
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleProcessingItem.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleProcessingItem.java b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleProcessingItem.java
new file mode 100644
index 0000000..1dd1562
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleProcessingItem.java
@@ -0,0 +1,87 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.core.ContentEvent;
+import org.apache.samoa.core.Processor;
+import org.apache.samoa.topology.AbstractProcessingItem;
+import org.apache.samoa.topology.IProcessingItem;
+import org.apache.samoa.topology.ProcessingItem;
+import org.apache.samoa.topology.Stream;
+import org.apache.samoa.utils.PartitioningScheme;
+import org.apache.samoa.utils.StreamDestination;
+
+/**
+ *
+ * @author abifet
+ */
+class SimpleProcessingItem extends AbstractProcessingItem {
+ private IProcessingItem[] arrayProcessingItem;
+
+ SimpleProcessingItem(Processor processor) {
+ super(processor);
+ }
+
+ SimpleProcessingItem(Processor processor, int parallelism) {
+ super(processor);
+ this.setParallelism(parallelism);
+ }
+
+ public IProcessingItem getProcessingItem(int i) {
+ return arrayProcessingItem[i];
+ }
+
+ @Override
+ protected ProcessingItem addInputStream(Stream inputStream, PartitioningScheme scheme) {
+ StreamDestination destination = new StreamDestination(this, this.getParallelism(), scheme);
+ ((SimpleStream) inputStream).addDestination(destination);
+ return this;
+ }
+
+ public SimpleProcessingItem copy() {
+ Processor processor = this.getProcessor();
+ return new SimpleProcessingItem(processor.newProcessor(processor));
+ }
+
+ public void processEvent(ContentEvent event, int counter) {
+
+ int parallelism = this.getParallelism();
+ // System.out.println("Process event "+event+" (isLast="+event.isLastEvent()+") with counter="+counter+" while parallelism="+parallelism);
+ if (this.arrayProcessingItem == null && parallelism > 0) {
+ // Init processing elements, the first time they are needed
+ this.arrayProcessingItem = new IProcessingItem[parallelism];
+ for (int j = 0; j < parallelism; j++) {
+ arrayProcessingItem[j] = this.copy();
+ arrayProcessingItem[j].getProcessor().onCreate(j);
+ }
+ }
+ if (this.arrayProcessingItem != null) {
+ IProcessingItem pi = this.getProcessingItem(counter);
+ Processor p = pi.getProcessor();
+ // System.out.println("PI="+pi+", p="+p);
+ this.getProcessingItem(counter).getProcessor().process(event);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleStream.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleStream.java b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleStream.java
new file mode 100644
index 0000000..3137c60
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleStream.java
@@ -0,0 +1,95 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.samoa.core.ContentEvent;
+import org.apache.samoa.topology.AbstractStream;
+import org.apache.samoa.topology.IProcessingItem;
+import org.apache.samoa.utils.StreamDestination;
+
+/**
+ *
+ * @author abifet
+ */
+class SimpleStream extends AbstractStream {
+ private List<StreamDestination> destinations;
+ private int maxCounter;
+ private int eventCounter;
+
+ SimpleStream(IProcessingItem sourcePi) {
+ super(sourcePi);
+ this.destinations = new LinkedList<>();
+ this.eventCounter = 0;
+ this.maxCounter = 1;
+ }
+
+ private int getNextCounter() {
+ if (maxCounter > 0 && eventCounter >= maxCounter)
+ eventCounter = 0;
+ this.eventCounter++;
+ return this.eventCounter;
+ }
+
+ @Override
+ public void put(ContentEvent event) {
+ this.put(event, this.getNextCounter());
+ }
+
+ private void put(ContentEvent event, int counter) {
+ SimpleProcessingItem pi;
+ int parallelism;
+ for (StreamDestination destination : destinations) {
+ pi = (SimpleProcessingItem) destination.getProcessingItem();
+ parallelism = destination.getParallelism();
+ switch (destination.getPartitioningScheme()) {
+ case SHUFFLE:
+ pi.processEvent(event, counter % parallelism);
+ break;
+ case GROUP_BY_KEY:
+ HashCodeBuilder hb = new HashCodeBuilder();
+ hb.append(event.getKey());
+ int key = hb.build() % parallelism;
+ pi.processEvent(event, key);
+ break;
+ case BROADCAST:
+ for (int p = 0; p < parallelism; p++) {
+ pi.processEvent(event, p);
+ }
+ break;
+ }
+ }
+ }
+
+ public void addDestination(StreamDestination destination) {
+ this.destinations.add(destination);
+ if (maxCounter <= 0)
+ maxCounter = 1;
+ maxCounter *= destination.getParallelism();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleTopology.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleTopology.java b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleTopology.java
new file mode 100644
index 0000000..660ebc0
--- /dev/null
+++ b/samoa-local/src/main/java/org/apache/samoa/topology/impl/SimpleTopology.java
@@ -0,0 +1,46 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.topology.AbstractTopology;
+
+public class SimpleTopology extends AbstractTopology {
+ SimpleTopology(String name) {
+ super(name);
+ }
+
+ public void run() {
+ if (this.getEntranceProcessingItems() == null)
+ throw new IllegalStateException("You need to set entrance PI before running the topology.");
+ if (this.getEntranceProcessingItems().size() != 1)
+ throw new IllegalStateException("SimpleTopology supports 1 entrance PI only. Number of entrance PIs is "
+ + this.getEntranceProcessingItems().size());
+
+ SimpleEntranceProcessingItem entrancePi = (SimpleEntranceProcessingItem) this.getEntranceProcessingItems()
+ .toArray()[0];
+ entrancePi.getProcessor().onCreate(0); // id=0 as it is not used in simple mode
+ entrancePi.startSendingEvents();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/AlgosTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/AlgosTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/AlgosTest.java
deleted file mode 100644
index 2f7c7e1..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/AlgosTest.java
+++ /dev/null
@@ -1,86 +0,0 @@
-package com.yahoo.labs.samoa;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import org.junit.Test;
-
-public class AlgosTest {
-
- @Test
- public void testVHTLocal() throws Exception {
-
- TestParams vhtConfig = new TestParams.Builder()
- .inputInstances(200_000)
- .samplingSize(20_000)
- .evaluationInstances(200_000)
- .classifiedInstances(200_000)
- .classificationsCorrect(75f)
- .kappaStat(0f)
- .kappaTempStat(0f)
- .cliStringTemplate(TestParams.Templates.PREQEVAL_VHT_RANDOMTREE)
- .resultFilePollTimeout(10)
- .prePollWait(10)
- .taskClassName(LocalDoTask.class.getName())
- .build();
- TestUtils.test(vhtConfig);
-
- }
-
- @Test
- public void testBaggingLocal() throws Exception {
- TestParams baggingConfig = new TestParams.Builder()
- .inputInstances(200_000)
- .samplingSize(20_000)
- .evaluationInstances(180_000)
- .classifiedInstances(210_000)
- .classificationsCorrect(60f)
- .kappaStat(0f)
- .kappaTempStat(0f)
- .cliStringTemplate(TestParams.Templates.PREQEVAL_BAGGING_RANDOMTREE)
- .prePollWait(10)
- .resultFilePollTimeout(10)
- .taskClassName(LocalDoTask.class.getName())
- .build();
- TestUtils.test(baggingConfig);
-
- }
-
- @Test
- public void testNaiveBayesLocal() throws Exception {
-
- TestParams vhtConfig = new TestParams.Builder()
- .inputInstances(200_000)
- .samplingSize(20_000)
- .evaluationInstances(200_000)
- .classifiedInstances(200_000)
- .classificationsCorrect(65f)
- .kappaStat(0f)
- .kappaTempStat(0f)
- .cliStringTemplate(TestParams.Templates.PREQEVAL_NAIVEBAYES_HYPERPLANE)
- .resultFilePollTimeout(10)
- .prePollWait(10)
- .taskClassName(LocalDoTask.class.getName())
- .build();
- TestUtils.test(vhtConfig);
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleComponentFactoryTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleComponentFactoryTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleComponentFactoryTest.java
deleted file mode 100644
index c43ef72..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleComponentFactoryTest.java
+++ /dev/null
@@ -1,99 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import static org.junit.Assert.*;
-import mockit.Mocked;
-import mockit.Tested;
-
-import org.junit.Before;
-import org.junit.Test;
-
-import com.yahoo.labs.samoa.core.EntranceProcessor;
-import com.yahoo.labs.samoa.core.Processor;
-import com.yahoo.labs.samoa.topology.EntranceProcessingItem;
-import com.yahoo.labs.samoa.topology.ProcessingItem;
-import com.yahoo.labs.samoa.topology.Stream;
-import com.yahoo.labs.samoa.topology.Topology;
-
-/**
- * @author Anh Thu Vu
- *
- */
-public class SimpleComponentFactoryTest {
-
- @Tested
- private SimpleComponentFactory factory;
- @Mocked
- private Processor processor, processorReplica;
- @Mocked
- private EntranceProcessor entranceProcessor;
-
- private final int parallelism = 3;
- private final String topoName = "TestTopology";
-
- @Before
- public void setUp() throws Exception {
- factory = new SimpleComponentFactory();
- }
-
- @Test
- public void testCreatePiNoParallelism() {
- ProcessingItem pi = factory.createPi(processor);
- assertNotNull("ProcessingItem created is null.", pi);
- assertEquals("ProcessingItem created is not a SimpleProcessingItem.", SimpleProcessingItem.class, pi.getClass());
- assertEquals("Parallelism of PI is not 1", 1, pi.getParallelism(), 0);
- }
-
- @Test
- public void testCreatePiWithParallelism() {
- ProcessingItem pi = factory.createPi(processor, parallelism);
- assertNotNull("ProcessingItem created is null.", pi);
- assertEquals("ProcessingItem created is not a SimpleProcessingItem.", SimpleProcessingItem.class, pi.getClass());
- assertEquals("Parallelism of PI is not ", parallelism, pi.getParallelism(), 0);
- }
-
- @Test
- public void testCreateStream() {
- ProcessingItem pi = factory.createPi(processor);
-
- Stream stream = factory.createStream(pi);
- assertNotNull("Stream created is null", stream);
- assertEquals("Stream created is not a SimpleStream.", SimpleStream.class, stream.getClass());
- }
-
- @Test
- public void testCreateTopology() {
- Topology topology = factory.createTopology(topoName);
- assertNotNull("Topology created is null.", topology);
- assertEquals("Topology created is not a SimpleTopology.", SimpleTopology.class, topology.getClass());
- }
-
- @Test
- public void testCreateEntrancePi() {
- EntranceProcessingItem entrancePi = factory.createEntrancePi(entranceProcessor);
- assertNotNull("EntranceProcessingItem created is null.", entrancePi);
- assertEquals("EntranceProcessingItem created is not a SimpleEntranceProcessingItem.",
- SimpleEntranceProcessingItem.class, entrancePi.getClass());
- assertSame("EntranceProcessor is not set correctly.", entranceProcessor, entrancePi.getProcessor());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEngineTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEngineTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEngineTest.java
deleted file mode 100644
index da4be1e..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEngineTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import mockit.Mocked;
-import mockit.NonStrictExpectations;
-import mockit.Tested;
-import mockit.Verifications;
-
-import org.junit.Test;
-
-/**
- * @author Anh Thu Vu
- *
- */
-public class SimpleEngineTest {
-
- @Tested
- private SimpleEngine unused;
- @Mocked
- private SimpleTopology topology;
- @Mocked
- private Runtime mockedRuntime;
-
- @Test
- public void testSubmitTopology() {
- new NonStrictExpectations() {
- {
- Runtime.getRuntime();
- result = mockedRuntime;
- mockedRuntime.exit(0);
- }
- };
- SimpleEngine.submitTopology(topology);
- new Verifications() {
- {
- topology.run();
- }
- };
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEntranceProcessingItemTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEntranceProcessingItemTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEntranceProcessingItemTest.java
deleted file mode 100644
index 8c1ccaf..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleEntranceProcessingItemTest.java
+++ /dev/null
@@ -1,172 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import static org.junit.Assert.*;
-import mockit.Mocked;
-import mockit.StrictExpectations;
-import mockit.Tested;
-import mockit.Verifications;
-
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import com.yahoo.labs.samoa.core.ContentEvent;
-import com.yahoo.labs.samoa.core.EntranceProcessor;
-import com.yahoo.labs.samoa.topology.Stream;
-
-/**
- * @author Anh Thu Vu
- *
- */
-public class SimpleEntranceProcessingItemTest {
-
- @Tested
- private SimpleEntranceProcessingItem entrancePi;
-
- @Mocked
- private EntranceProcessor entranceProcessor;
- @Mocked
- private Stream outputStream, anotherStream;
- @Mocked
- private ContentEvent event;
-
- @Mocked
- private Thread unused;
-
- /**
- * @throws java.lang.Exception
- */
- @Before
- public void setUp() throws Exception {
- entrancePi = new SimpleEntranceProcessingItem(entranceProcessor);
- }
-
- @Test
- public void testContructor() {
- assertSame("EntranceProcessor is not set correctly.", entranceProcessor, entrancePi.getProcessor());
- }
-
- @Test
- public void testSetOutputStream() {
- entrancePi.setOutputStream(outputStream);
- assertSame("OutputStream is not set correctly.", outputStream, entrancePi.getOutputStream());
- }
-
- @Test
- public void testSetOutputStreamRepeate() {
- entrancePi.setOutputStream(outputStream);
- entrancePi.setOutputStream(outputStream);
- assertSame("OutputStream is not set correctly.", outputStream, entrancePi.getOutputStream());
- }
-
- @Test(expected = IllegalStateException.class)
- public void testSetOutputStreamError() {
- entrancePi.setOutputStream(outputStream);
- entrancePi.setOutputStream(anotherStream);
- }
-
- @Test
- public void testInjectNextEventSuccess() {
- entrancePi.setOutputStream(outputStream);
- new StrictExpectations() {
- {
- entranceProcessor.hasNext();
- result = true;
-
- entranceProcessor.nextEvent();
- result = event;
- }
- };
- entrancePi.injectNextEvent();
- new Verifications() {
- {
- outputStream.put(event);
- }
- };
- }
-
- @Test
- public void testStartSendingEvents() {
- entrancePi.setOutputStream(outputStream);
- new StrictExpectations() {
- {
- for (int i = 0; i < 1; i++) {
- entranceProcessor.isFinished();
- result = false;
- entranceProcessor.hasNext();
- result = false;
- }
-
- for (int i = 0; i < 5; i++) {
- entranceProcessor.isFinished();
- result = false;
- entranceProcessor.hasNext();
- result = true;
- entranceProcessor.nextEvent();
- result = event;
- outputStream.put(event);
- }
-
- for (int i = 0; i < 2; i++) {
- entranceProcessor.isFinished();
- result = false;
- entranceProcessor.hasNext();
- result = false;
- }
-
- for (int i = 0; i < 5; i++) {
- entranceProcessor.isFinished();
- result = false;
- entranceProcessor.hasNext();
- result = true;
- entranceProcessor.nextEvent();
- result = event;
- outputStream.put(event);
- }
-
- entranceProcessor.isFinished();
- result = true;
- times = 1;
- entranceProcessor.hasNext();
- times = 0;
- }
- };
- entrancePi.startSendingEvents();
- new Verifications() {
- {
- try {
- Thread.sleep(anyInt);
- times = 3;
- } catch (InterruptedException e) {
-
- }
- }
- };
- }
-
- @Test(expected = IllegalStateException.class)
- public void testStartSendingEventsError() {
- entrancePi.startSendingEvents();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleProcessingItemTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleProcessingItemTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleProcessingItemTest.java
deleted file mode 100644
index cd076fe..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleProcessingItemTest.java
+++ /dev/null
@@ -1,124 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import static org.junit.Assert.*;
-
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-
-import mockit.Expectations;
-import mockit.Mocked;
-import mockit.NonStrictExpectations;
-import mockit.Tested;
-
-import org.junit.Before;
-import org.junit.Test;
-
-import com.yahoo.labs.samoa.core.ContentEvent;
-import com.yahoo.labs.samoa.core.Processor;
-import com.yahoo.labs.samoa.utils.PartitioningScheme;
-import com.yahoo.labs.samoa.utils.StreamDestination;
-
-/**
- * @author Anh Thu Vu
- *
- */
-public class SimpleProcessingItemTest {
-
- @Tested
- private SimpleProcessingItem pi;
-
- @Mocked
- private Processor processor;
- @Mocked
- private SimpleStream stream;
- @Mocked
- private StreamDestination destination;
- @Mocked
- private ContentEvent event;
-
- private final int parallelism = 4;
- private final int counter = 2;
-
- @Before
- public void setUp() throws Exception {
- pi = new SimpleProcessingItem(processor, parallelism);
- }
-
- @Test
- public void testConstructor() {
- assertSame("Processor was not set correctly.", processor, pi.getProcessor());
- assertEquals("Parallelism was not set correctly.", parallelism, pi.getParallelism(), 0);
- }
-
- @Test
- public void testConnectInputShuffleStream() {
- new Expectations() {
- {
- destination = new StreamDestination(pi, parallelism, PartitioningScheme.SHUFFLE);
- stream.addDestination(destination);
- }
- };
- pi.connectInputShuffleStream(stream);
- }
-
- @Test
- public void testConnectInputKeyStream() {
- new Expectations() {
- {
- destination = new StreamDestination(pi, parallelism, PartitioningScheme.GROUP_BY_KEY);
- stream.addDestination(destination);
- }
- };
- pi.connectInputKeyStream(stream);
- }
-
- @Test
- public void testConnectInputAllStream() {
- new Expectations() {
- {
- destination = new StreamDestination(pi, parallelism, PartitioningScheme.BROADCAST);
- stream.addDestination(destination);
- }
- };
- pi.connectInputAllStream(stream);
- }
-
- @Test
- public void testProcessEvent() {
- new Expectations() {
- {
- for (int i = 0; i < parallelism; i++) {
- processor.newProcessor(processor);
- result = processor;
-
- processor.onCreate(anyInt);
- }
-
- processor.process(event);
- }
- };
- pi.processEvent(event, counter);
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleStreamTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleStreamTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleStreamTest.java
deleted file mode 100644
index 25cb5eb..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleStreamTest.java
+++ /dev/null
@@ -1,121 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import java.util.Arrays;
-import java.util.Collection;
-
-import mockit.Expectations;
-import mockit.Mocked;
-import mockit.NonStrictExpectations;
-import mockit.Tested;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-
-import com.yahoo.labs.samoa.core.ContentEvent;
-import com.yahoo.labs.samoa.utils.PartitioningScheme;
-import com.yahoo.labs.samoa.utils.StreamDestination;
-
-/**
- * @author Anh Thu Vu
- *
- */
-@RunWith(Parameterized.class)
-public class SimpleStreamTest {
-
- @Tested
- private SimpleStream stream;
-
- @Mocked
- private SimpleProcessingItem sourcePi, destPi;
- @Mocked
- private ContentEvent event;
- @Mocked
- private StreamDestination destination;
-
- private final String eventKey = "eventkey";
- private final int parallelism;
- private final PartitioningScheme scheme;
-
- @Parameters
- public static Collection<Object[]> generateParameters() {
- return Arrays.asList(new Object[][] {
- { 2, PartitioningScheme.SHUFFLE },
- { 3, PartitioningScheme.GROUP_BY_KEY },
- { 4, PartitioningScheme.BROADCAST }
- });
- }
-
- public SimpleStreamTest(int parallelism, PartitioningScheme scheme) {
- this.parallelism = parallelism;
- this.scheme = scheme;
- }
-
- @Before
- public void setUp() throws Exception {
- stream = new SimpleStream(sourcePi);
- stream.addDestination(destination);
- }
-
- @Test
- public void testPut() {
- new NonStrictExpectations() {
- {
- event.getKey();
- result = eventKey;
- destination.getProcessingItem();
- result = destPi;
- destination.getPartitioningScheme();
- result = scheme;
- destination.getParallelism();
- result = parallelism;
-
- }
- };
- switch (this.scheme) {
- case SHUFFLE:
- case GROUP_BY_KEY:
- new Expectations() {
- {
- // TODO: restrict the range of counter value
- destPi.processEvent(event, anyInt);
- times = 1;
- }
- };
- break;
- case BROADCAST:
- new Expectations() {
- {
- // TODO: restrict the range of counter value
- destPi.processEvent(event, anyInt);
- times = parallelism;
- }
- };
- break;
- }
- stream.put(event);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleTopologyTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleTopologyTest.java b/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleTopologyTest.java
deleted file mode 100644
index 9aaaebd..0000000
--- a/samoa-local/src/test/java/com/yahoo/labs/samoa/topology/impl/SimpleTopologyTest.java
+++ /dev/null
@@ -1,94 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-import static org.junit.Assert.*;
-
-import java.util.Set;
-
-import mockit.NonStrictExpectations;
-import mockit.Expectations;
-import mockit.Mocked;
-import mockit.Tested;
-
-import org.junit.Before;
-import org.junit.Test;
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import com.yahoo.labs.samoa.core.EntranceProcessor;
-import com.yahoo.labs.samoa.topology.EntranceProcessingItem;
-
-/**
- * @author Anh Thu Vu
- *
- */
-public class SimpleTopologyTest {
-
- @Tested
- private SimpleTopology topology;
-
- @Mocked
- private SimpleEntranceProcessingItem entrancePi;
- @Mocked
- private EntranceProcessor entranceProcessor;
-
- @Before
- public void setUp() throws Exception {
- topology = new SimpleTopology("TestTopology");
- }
-
- @Test
- public void testAddEntrancePi() {
- topology.addEntranceProcessingItem(entrancePi);
-
- Set<EntranceProcessingItem> entrancePIs = topology.getEntranceProcessingItems();
- assertNotNull("Set of entrance PIs is null.", entrancePIs);
- assertEquals("Number of entrance PI in SimpleTopology must be 1", 1, entrancePIs.size());
- assertSame("Entrance PI was not set correctly.", entrancePi, entrancePIs.toArray()[0]);
- // TODO: verify that entrance PI is in the set of ProcessingItems
- // Need to access topology's set of PIs (getProcessingItems() method)
- }
-
- @Test
- public void testRun() {
- topology.addEntranceProcessingItem(entrancePi);
-
- new NonStrictExpectations() {
- {
- entrancePi.getProcessor();
- result = entranceProcessor;
-
- }
- };
-
- new Expectations() {
- {
- entranceProcessor.onCreate(anyInt);
- entrancePi.startSendingEvents();
- }
- };
- topology.run();
- }
-
- @Test(expected = IllegalStateException.class)
- public void testRunWithoutEntrancePI() {
- topology.run();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/AlgosTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/AlgosTest.java b/samoa-local/src/test/java/org/apache/samoa/AlgosTest.java
new file mode 100644
index 0000000..f35b92a
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/AlgosTest.java
@@ -0,0 +1,87 @@
+package org.apache.samoa;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.samoa.LocalDoTask;
+import org.junit.Test;
+
+public class AlgosTest {
+
+ @Test
+ public void testVHTLocal() throws Exception {
+
+ TestParams vhtConfig = new TestParams.Builder()
+ .inputInstances(200_000)
+ .samplingSize(20_000)
+ .evaluationInstances(200_000)
+ .classifiedInstances(200_000)
+ .classificationsCorrect(75f)
+ .kappaStat(0f)
+ .kappaTempStat(0f)
+ .cliStringTemplate(TestParams.Templates.PREQEVAL_VHT_RANDOMTREE)
+ .resultFilePollTimeout(10)
+ .prePollWait(10)
+ .taskClassName(LocalDoTask.class.getName())
+ .build();
+ TestUtils.test(vhtConfig);
+
+ }
+
+ @Test
+ public void testBaggingLocal() throws Exception {
+ TestParams baggingConfig = new TestParams.Builder()
+ .inputInstances(200_000)
+ .samplingSize(20_000)
+ .evaluationInstances(180_000)
+ .classifiedInstances(210_000)
+ .classificationsCorrect(60f)
+ .kappaStat(0f)
+ .kappaTempStat(0f)
+ .cliStringTemplate(TestParams.Templates.PREQEVAL_BAGGING_RANDOMTREE)
+ .prePollWait(10)
+ .resultFilePollTimeout(10)
+ .taskClassName(LocalDoTask.class.getName())
+ .build();
+ TestUtils.test(baggingConfig);
+
+ }
+
+ @Test
+ public void testNaiveBayesLocal() throws Exception {
+
+ TestParams vhtConfig = new TestParams.Builder()
+ .inputInstances(200_000)
+ .samplingSize(20_000)
+ .evaluationInstances(200_000)
+ .classifiedInstances(200_000)
+ .classificationsCorrect(65f)
+ .kappaStat(0f)
+ .kappaTempStat(0f)
+ .cliStringTemplate(TestParams.Templates.PREQEVAL_NAIVEBAYES_HYPERPLANE)
+ .resultFilePollTimeout(10)
+ .prePollWait(10)
+ .taskClassName(LocalDoTask.class.getName())
+ .build();
+ TestUtils.test(vhtConfig);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleComponentFactoryTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleComponentFactoryTest.java b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleComponentFactoryTest.java
new file mode 100644
index 0000000..3085d4c
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleComponentFactoryTest.java
@@ -0,0 +1,103 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import static org.junit.Assert.*;
+import mockit.Mocked;
+import mockit.Tested;
+
+import org.apache.samoa.core.EntranceProcessor;
+import org.apache.samoa.core.Processor;
+import org.apache.samoa.topology.EntranceProcessingItem;
+import org.apache.samoa.topology.ProcessingItem;
+import org.apache.samoa.topology.Stream;
+import org.apache.samoa.topology.Topology;
+import org.apache.samoa.topology.impl.SimpleComponentFactory;
+import org.apache.samoa.topology.impl.SimpleEntranceProcessingItem;
+import org.apache.samoa.topology.impl.SimpleProcessingItem;
+import org.apache.samoa.topology.impl.SimpleStream;
+import org.apache.samoa.topology.impl.SimpleTopology;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author Anh Thu Vu
+ *
+ */
+public class SimpleComponentFactoryTest {
+
+ @Tested
+ private SimpleComponentFactory factory;
+ @Mocked
+ private Processor processor, processorReplica;
+ @Mocked
+ private EntranceProcessor entranceProcessor;
+
+ private final int parallelism = 3;
+ private final String topoName = "TestTopology";
+
+ @Before
+ public void setUp() throws Exception {
+ factory = new SimpleComponentFactory();
+ }
+
+ @Test
+ public void testCreatePiNoParallelism() {
+ ProcessingItem pi = factory.createPi(processor);
+ assertNotNull("ProcessingItem created is null.", pi);
+ assertEquals("ProcessingItem created is not a SimpleProcessingItem.", SimpleProcessingItem.class, pi.getClass());
+ assertEquals("Parallelism of PI is not 1", 1, pi.getParallelism(), 0);
+ }
+
+ @Test
+ public void testCreatePiWithParallelism() {
+ ProcessingItem pi = factory.createPi(processor, parallelism);
+ assertNotNull("ProcessingItem created is null.", pi);
+ assertEquals("ProcessingItem created is not a SimpleProcessingItem.", SimpleProcessingItem.class, pi.getClass());
+ assertEquals("Parallelism of PI is not ", parallelism, pi.getParallelism(), 0);
+ }
+
+ @Test
+ public void testCreateStream() {
+ ProcessingItem pi = factory.createPi(processor);
+
+ Stream stream = factory.createStream(pi);
+ assertNotNull("Stream created is null", stream);
+ assertEquals("Stream created is not a SimpleStream.", SimpleStream.class, stream.getClass());
+ }
+
+ @Test
+ public void testCreateTopology() {
+ Topology topology = factory.createTopology(topoName);
+ assertNotNull("Topology created is null.", topology);
+ assertEquals("Topology created is not a SimpleTopology.", SimpleTopology.class, topology.getClass());
+ }
+
+ @Test
+ public void testCreateEntrancePi() {
+ EntranceProcessingItem entrancePi = factory.createEntrancePi(entranceProcessor);
+ assertNotNull("EntranceProcessingItem created is null.", entrancePi);
+ assertEquals("EntranceProcessingItem created is not a SimpleEntranceProcessingItem.",
+ SimpleEntranceProcessingItem.class, entrancePi.getClass());
+ assertSame("EntranceProcessor is not set correctly.", entranceProcessor, entrancePi.getProcessor());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEngineTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEngineTest.java b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEngineTest.java
new file mode 100644
index 0000000..59d5afe
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEngineTest.java
@@ -0,0 +1,62 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import mockit.Mocked;
+import mockit.NonStrictExpectations;
+import mockit.Tested;
+import mockit.Verifications;
+
+import org.apache.samoa.topology.impl.SimpleEngine;
+import org.apache.samoa.topology.impl.SimpleTopology;
+import org.junit.Test;
+
+/**
+ * @author Anh Thu Vu
+ *
+ */
+public class SimpleEngineTest {
+
+ @Tested
+ private SimpleEngine unused;
+ @Mocked
+ private SimpleTopology topology;
+ @Mocked
+ private Runtime mockedRuntime;
+
+ @Test
+ public void testSubmitTopology() {
+ new NonStrictExpectations() {
+ {
+ Runtime.getRuntime();
+ result = mockedRuntime;
+ mockedRuntime.exit(0);
+ }
+ };
+ SimpleEngine.submitTopology(topology);
+ new Verifications() {
+ {
+ topology.run();
+ }
+ };
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItemTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItemTest.java b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItemTest.java
new file mode 100644
index 0000000..41d1f46
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleEntranceProcessingItemTest.java
@@ -0,0 +1,172 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import static org.junit.Assert.*;
+import mockit.Mocked;
+import mockit.StrictExpectations;
+import mockit.Tested;
+import mockit.Verifications;
+
+import org.apache.samoa.core.ContentEvent;
+import org.apache.samoa.core.EntranceProcessor;
+import org.apache.samoa.topology.Stream;
+import org.apache.samoa.topology.impl.SimpleEntranceProcessingItem;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+/**
+ * @author Anh Thu Vu
+ *
+ */
+public class SimpleEntranceProcessingItemTest {
+
+ @Tested
+ private SimpleEntranceProcessingItem entrancePi;
+
+ @Mocked
+ private EntranceProcessor entranceProcessor;
+ @Mocked
+ private Stream outputStream, anotherStream;
+ @Mocked
+ private ContentEvent event;
+
+ @Mocked
+ private Thread unused;
+
+ /**
+ * @throws java.lang.Exception
+ */
+ @Before
+ public void setUp() throws Exception {
+ entrancePi = new SimpleEntranceProcessingItem(entranceProcessor);
+ }
+
+ @Test
+ public void testContructor() {
+ assertSame("EntranceProcessor is not set correctly.", entranceProcessor, entrancePi.getProcessor());
+ }
+
+ @Test
+ public void testSetOutputStream() {
+ entrancePi.setOutputStream(outputStream);
+ assertSame("OutputStream is not set correctly.", outputStream, entrancePi.getOutputStream());
+ }
+
+ @Test
+ public void testSetOutputStreamRepeate() {
+ entrancePi.setOutputStream(outputStream);
+ entrancePi.setOutputStream(outputStream);
+ assertSame("OutputStream is not set correctly.", outputStream, entrancePi.getOutputStream());
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testSetOutputStreamError() {
+ entrancePi.setOutputStream(outputStream);
+ entrancePi.setOutputStream(anotherStream);
+ }
+
+ @Test
+ public void testInjectNextEventSuccess() {
+ entrancePi.setOutputStream(outputStream);
+ new StrictExpectations() {
+ {
+ entranceProcessor.hasNext();
+ result = true;
+
+ entranceProcessor.nextEvent();
+ result = event;
+ }
+ };
+ entrancePi.injectNextEvent();
+ new Verifications() {
+ {
+ outputStream.put(event);
+ }
+ };
+ }
+
+ @Test
+ public void testStartSendingEvents() {
+ entrancePi.setOutputStream(outputStream);
+ new StrictExpectations() {
+ {
+ for (int i = 0; i < 1; i++) {
+ entranceProcessor.isFinished();
+ result = false;
+ entranceProcessor.hasNext();
+ result = false;
+ }
+
+ for (int i = 0; i < 5; i++) {
+ entranceProcessor.isFinished();
+ result = false;
+ entranceProcessor.hasNext();
+ result = true;
+ entranceProcessor.nextEvent();
+ result = event;
+ outputStream.put(event);
+ }
+
+ for (int i = 0; i < 2; i++) {
+ entranceProcessor.isFinished();
+ result = false;
+ entranceProcessor.hasNext();
+ result = false;
+ }
+
+ for (int i = 0; i < 5; i++) {
+ entranceProcessor.isFinished();
+ result = false;
+ entranceProcessor.hasNext();
+ result = true;
+ entranceProcessor.nextEvent();
+ result = event;
+ outputStream.put(event);
+ }
+
+ entranceProcessor.isFinished();
+ result = true;
+ times = 1;
+ entranceProcessor.hasNext();
+ times = 0;
+ }
+ };
+ entrancePi.startSendingEvents();
+ new Verifications() {
+ {
+ try {
+ Thread.sleep(anyInt);
+ times = 3;
+ } catch (InterruptedException e) {
+
+ }
+ }
+ };
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStartSendingEventsError() {
+ entrancePi.startSendingEvents();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleProcessingItemTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleProcessingItemTest.java b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleProcessingItemTest.java
new file mode 100644
index 0000000..42602ec
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleProcessingItemTest.java
@@ -0,0 +1,125 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import static org.junit.Assert.*;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import mockit.Expectations;
+import mockit.Mocked;
+import mockit.NonStrictExpectations;
+import mockit.Tested;
+
+import org.apache.samoa.core.ContentEvent;
+import org.apache.samoa.core.Processor;
+import org.apache.samoa.topology.impl.SimpleProcessingItem;
+import org.apache.samoa.topology.impl.SimpleStream;
+import org.apache.samoa.utils.PartitioningScheme;
+import org.apache.samoa.utils.StreamDestination;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author Anh Thu Vu
+ *
+ */
+public class SimpleProcessingItemTest {
+
+ @Tested
+ private SimpleProcessingItem pi;
+
+ @Mocked
+ private Processor processor;
+ @Mocked
+ private SimpleStream stream;
+ @Mocked
+ private StreamDestination destination;
+ @Mocked
+ private ContentEvent event;
+
+ private final int parallelism = 4;
+ private final int counter = 2;
+
+ @Before
+ public void setUp() throws Exception {
+ pi = new SimpleProcessingItem(processor, parallelism);
+ }
+
+ @Test
+ public void testConstructor() {
+ assertSame("Processor was not set correctly.", processor, pi.getProcessor());
+ assertEquals("Parallelism was not set correctly.", parallelism, pi.getParallelism(), 0);
+ }
+
+ @Test
+ public void testConnectInputShuffleStream() {
+ new Expectations() {
+ {
+ destination = new StreamDestination(pi, parallelism, PartitioningScheme.SHUFFLE);
+ stream.addDestination(destination);
+ }
+ };
+ pi.connectInputShuffleStream(stream);
+ }
+
+ @Test
+ public void testConnectInputKeyStream() {
+ new Expectations() {
+ {
+ destination = new StreamDestination(pi, parallelism, PartitioningScheme.GROUP_BY_KEY);
+ stream.addDestination(destination);
+ }
+ };
+ pi.connectInputKeyStream(stream);
+ }
+
+ @Test
+ public void testConnectInputAllStream() {
+ new Expectations() {
+ {
+ destination = new StreamDestination(pi, parallelism, PartitioningScheme.BROADCAST);
+ stream.addDestination(destination);
+ }
+ };
+ pi.connectInputAllStream(stream);
+ }
+
+ @Test
+ public void testProcessEvent() {
+ new Expectations() {
+ {
+ for (int i = 0; i < parallelism; i++) {
+ processor.newProcessor(processor);
+ result = processor;
+
+ processor.onCreate(anyInt);
+ }
+
+ processor.process(event);
+ }
+ };
+ pi.processEvent(event, counter);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleStreamTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleStreamTest.java b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleStreamTest.java
new file mode 100644
index 0000000..51e25bb
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleStreamTest.java
@@ -0,0 +1,122 @@
+package org.apache.samoa.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import mockit.Expectations;
+import mockit.Mocked;
+import mockit.NonStrictExpectations;
+import mockit.Tested;
+
+import org.apache.samoa.core.ContentEvent;
+import org.apache.samoa.topology.impl.SimpleProcessingItem;
+import org.apache.samoa.topology.impl.SimpleStream;
+import org.apache.samoa.utils.PartitioningScheme;
+import org.apache.samoa.utils.StreamDestination;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+/**
+ * @author Anh Thu Vu
+ *
+ */
+@RunWith(Parameterized.class)
+public class SimpleStreamTest {
+
+ @Tested
+ private SimpleStream stream;
+
+ @Mocked
+ private SimpleProcessingItem sourcePi, destPi;
+ @Mocked
+ private ContentEvent event;
+ @Mocked
+ private StreamDestination destination;
+
+ private final String eventKey = "eventkey";
+ private final int parallelism;
+ private final PartitioningScheme scheme;
+
+ @Parameters
+ public static Collection<Object[]> generateParameters() {
+ return Arrays.asList(new Object[][] {
+ { 2, PartitioningScheme.SHUFFLE },
+ { 3, PartitioningScheme.GROUP_BY_KEY },
+ { 4, PartitioningScheme.BROADCAST }
+ });
+ }
+
+ public SimpleStreamTest(int parallelism, PartitioningScheme scheme) {
+ this.parallelism = parallelism;
+ this.scheme = scheme;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ stream = new SimpleStream(sourcePi);
+ stream.addDestination(destination);
+ }
+
+ @Test
+ public void testPut() {
+ new NonStrictExpectations() {
+ {
+ event.getKey();
+ result = eventKey;
+ destination.getProcessingItem();
+ result = destPi;
+ destination.getPartitioningScheme();
+ result = scheme;
+ destination.getParallelism();
+ result = parallelism;
+
+ }
+ };
+ switch (this.scheme) {
+ case SHUFFLE:
+ case GROUP_BY_KEY:
+ new Expectations() {
+ {
+ // TODO: restrict the range of counter value
+ destPi.processEvent(event, anyInt);
+ times = 1;
+ }
+ };
+ break;
+ case BROADCAST:
+ new Expectations() {
+ {
+ // TODO: restrict the range of counter value
+ destPi.processEvent(event, anyInt);
+ times = parallelism;
+ }
+ };
+ break;
+ }
+ stream.put(event);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleTopologyTest.java
----------------------------------------------------------------------
diff --git a/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleTopologyTest.java b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleTopologyTest.java
new file mode 100644
index 0000000..6d8d728
--- /dev/null
+++ b/samoa-local/src/test/java/org/apache/samoa/topology/impl/SimpleTopologyTest.java
@@ -0,0 +1,97 @@
+package org.apache.samoa.topology.impl;
+
+import static org.junit.Assert.*;
+
+import java.util.Set;
+
+import mockit.NonStrictExpectations;
+import mockit.Expectations;
+import mockit.Mocked;
+import mockit.Tested;
+
+import org.apache.samoa.core.EntranceProcessor;
+import org.apache.samoa.topology.EntranceProcessingItem;
+import org.apache.samoa.topology.impl.SimpleEntranceProcessingItem;
+import org.apache.samoa.topology.impl.SimpleTopology;
+import org.junit.Before;
+import org.junit.Test;
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+
+
+/**
+ * @author Anh Thu Vu
+ *
+ */
+public class SimpleTopologyTest {
+
+ @Tested
+ private SimpleTopology topology;
+
+ @Mocked
+ private SimpleEntranceProcessingItem entrancePi;
+ @Mocked
+ private EntranceProcessor entranceProcessor;
+
+ @Before
+ public void setUp() throws Exception {
+ topology = new SimpleTopology("TestTopology");
+ }
+
+ @Test
+ public void testAddEntrancePi() {
+ topology.addEntranceProcessingItem(entrancePi);
+
+ Set<EntranceProcessingItem> entrancePIs = topology.getEntranceProcessingItems();
+ assertNotNull("Set of entrance PIs is null.", entrancePIs);
+ assertEquals("Number of entrance PI in SimpleTopology must be 1", 1, entrancePIs.size());
+ assertSame("Entrance PI was not set correctly.", entrancePi, entrancePIs.toArray()[0]);
+ // TODO: verify that entrance PI is in the set of ProcessingItems
+ // Need to access topology's set of PIs (getProcessingItems() method)
+ }
+
+ @Test
+ public void testRun() {
+ topology.addEntranceProcessingItem(entrancePi);
+
+ new NonStrictExpectations() {
+ {
+ entrancePi.getProcessor();
+ result = entranceProcessor;
+
+ }
+ };
+
+ new Expectations() {
+ {
+ entranceProcessor.onCreate(anyInt);
+ entrancePi.startSendingEvents();
+ }
+ };
+ topology.run();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testRunWithoutEntrancePI() {
+ topology.run();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-s4/pom.xml
----------------------------------------------------------------------
diff --git a/samoa-s4/pom.xml b/samoa-s4/pom.xml
index 8135be9..0f9b804 100644
--- a/samoa-s4/pom.xml
+++ b/samoa-s4/pom.xml
@@ -31,14 +31,14 @@
<artifactId>samoa-s4</artifactId>
<parent>
- <groupId>com.yahoo.labs.samoa</groupId>
+ <groupId>org.apache.samoa</groupId>
<artifactId>samoa</artifactId>
<version>0.3.0-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
- <groupId>com.yahoo.labs.samoa</groupId>
+ <groupId>org.apache.samoa</groupId>
<artifactId>samoa-api</artifactId>
<version>${project.version}</version>
</dependency>
@@ -117,7 +117,7 @@
<Implementation-Version>${project.version}</Implementation-Version>
<Implementation-Vendor>Yahoo Labs</Implementation-Vendor>
<Implementation-Vendor-Id>SAMOA</Implementation-Vendor-Id>
- <S4-App-Class>com.yahoo.labs.samoa.topology.impl.S4DoTask</S4-App-Class>
+ <S4-App-Class>org.apache.samoa.topology.impl.S4DoTask</S4-App-Class>
<S4-Version>${s4.version}</S4-Version>
</manifestEntries>
</archive>
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-s4/samoa-s4-adapter/pom.xml
----------------------------------------------------------------------
diff --git a/samoa-s4/samoa-s4-adapter/pom.xml b/samoa-s4/samoa-s4-adapter/pom.xml
index 27c8d51..5a66a1e 100644
--- a/samoa-s4/samoa-s4-adapter/pom.xml
+++ b/samoa-s4/samoa-s4-adapter/pom.xml
@@ -37,22 +37,16 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
- <!-- PARENT MODULE SAMOA-S4 <parent> <groupId>com.yahoo.labs.bcn.samoa</groupId> <artifactId>samoa-s4</artifactId> <version>0.1</version>
- </parent> -->
-
- <!-- SAMOA-S4-ADAPTER MODUEL -->
<artifactId>samoa-s4-adapter</artifactId>
- <groupId>com.yahoo.labs.bcn.samoa</groupId>
+ <groupId>org.apache.samoa</groupId>
<version>0.1</version>
<name>samoa-s4-adapter</name>
<description>Adapter module to connect to external stream and also to provide entrance processing items for SAMOA</description>
<dependencies>
- <!-- dependency> <artifactId>samoa-api</artifactId> <groupId>com.yahoo.labs.bcn.samoa</groupId> <version>0.1</version>
- </dependency> -->
<dependency>
<artifactId>samoa-s4</artifactId>
- <groupId>com.yahoo.labs.bcn.samoa</groupId>
+ <groupId>org.apache.samoa</groupId>
<version>0.1</version>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4ComponentFactory.java
----------------------------------------------------------------------
diff --git a/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4ComponentFactory.java b/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4ComponentFactory.java
deleted file mode 100644
index 24602a4..0000000
--- a/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4ComponentFactory.java
+++ /dev/null
@@ -1,98 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.yahoo.labs.samoa.core.EntranceProcessor;
-import com.yahoo.labs.samoa.core.Processor;
-import com.yahoo.labs.samoa.topology.ComponentFactory;
-import com.yahoo.labs.samoa.topology.EntranceProcessingItem;
-import com.yahoo.labs.samoa.topology.IProcessingItem;
-import com.yahoo.labs.samoa.topology.ProcessingItem;
-import com.yahoo.labs.samoa.topology.Stream;
-import com.yahoo.labs.samoa.topology.Topology;
-
-/**
- * S4 Platform Component Factory
- *
- * @author severien
- *
- */
-public class S4ComponentFactory implements ComponentFactory {
-
- public static final Logger logger = LoggerFactory.getLogger(S4ComponentFactory.class);
- protected S4DoTask app;
-
- @Override
- public ProcessingItem createPi(Processor processor, int paralellism) {
- S4ProcessingItem processingItem = new S4ProcessingItem(app);
- // TODO refactor how to set the paralellism level
- processingItem.setParalellismLevel(paralellism);
- processingItem.setProcessor(processor);
-
- return processingItem;
- }
-
- @Override
- public ProcessingItem createPi(Processor processor) {
- return this.createPi(processor, 1);
- }
-
- @Override
- public EntranceProcessingItem createEntrancePi(EntranceProcessor entranceProcessor) {
- // TODO Create source Entry processing item that connects to an external
- // stream
- S4EntranceProcessingItem entrancePi = new S4EntranceProcessingItem(entranceProcessor, app);
- entrancePi.setParallelism(1); // FIXME should not be set to 1 statically
- return entrancePi;
- }
-
- @Override
- public Stream createStream(IProcessingItem sourcePi) {
- S4Stream aStream = new S4Stream(app);
- return aStream;
- }
-
- @Override
- public Topology createTopology(String topoName) {
- return new S4Topology(topoName);
- }
-
- /**
- * Initialization method.
- *
- * @param evalTask
- */
- public void init(String evalTask) {
- // Task is initiated in the DoTaskApp
- }
-
- /**
- * Sets S4 application.
- *
- * @param app
- */
- public void setApp(S4DoTask app) {
- this.app = app;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-samoa/blob/9b178f63/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4DoTask.java
----------------------------------------------------------------------
diff --git a/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4DoTask.java b/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4DoTask.java
deleted file mode 100644
index cc4b18d..0000000
--- a/samoa-s4/src/main/java/com/yahoo/labs/samoa/topology/impl/S4DoTask.java
+++ /dev/null
@@ -1,268 +0,0 @@
-package com.yahoo.labs.samoa.topology.impl;
-
-/*
- * #%L
- * SAMOA
- * %%
- * Copyright (C) 2014 - 2015 Apache Software Foundation
- * %%
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * #L%
- */
-
-/**
- * License
- */
-
-import java.io.UnsupportedEncodingException;
-import java.net.URLDecoder;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.s4.base.Event;
-import org.apache.s4.base.KeyFinder;
-import org.apache.s4.core.App;
-import org.apache.s4.core.ProcessingElement;
-import org.apache.s4.core.Stream;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import com.github.javacliparser.Option;
-import com.github.javacliparser.ClassOption;
-import com.yahoo.labs.samoa.core.Globals;
-import com.yahoo.labs.samoa.tasks.Task;
-import com.yahoo.labs.samoa.topology.ComponentFactory;
-
-import com.google.inject.Inject;
-import com.google.inject.name.Named;
-
-/*
- * S4 App that runs samoa Tasks
- *
- * */
-
-/**
- * The Class DoTaskApp.
- */
-final public class S4DoTask extends App {
-
- private final Logger logger = LoggerFactory.getLogger(S4DoTask.class);
- Task task;
-
- @Inject
- @Named("evalTask")
- public String evalTask;
-
- public S4DoTask() {
- super();
- }
-
- /** The engine. */
- protected ComponentFactory componentFactory;
-
- /**
- * Gets the factory.
- *
- * @return the factory
- */
- public ComponentFactory getFactory() {
- return componentFactory;
- }
-
- /**
- * Sets the factory.
- *
- * @param factory
- * the new factory
- */
- public void setFactory(ComponentFactory factory) {
- this.componentFactory = factory;
- }
-
- /*
- * Build the application
- *
- * @see org.apache.s4.core.App#onInit()
- */
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.App#onInit()
- */
- @Override
- protected void onInit() {
- logger.info("DoTaskApp onInit");
- // ConsoleReporters prints S4 metrics
- // MetricsRegistry mr = new MetricsRegistry();
- //
- // CsvReporter.enable(new File(System.getProperty("user.home")
- // + "/monitor/"), 10, TimeUnit.SECONDS);
- // ConsoleReporter.enable(10, TimeUnit.SECONDS);
- try {
- System.err.println();
- System.err.println(Globals.getWorkbenchInfoString());
- System.err.println();
-
- } catch (Exception ex) {
- ex.printStackTrace();
- }
- S4ComponentFactory factory = new S4ComponentFactory();
- factory.setApp(this);
-
- // logger.debug("LC {}", lc);
-
- // task = TaskProvider.getTask(evalTask);
-
- // EXAMPLE OPTIONS
- // -l Clustream -g Clustream -i 100000 -s (RandomRBFGeneratorEvents -K
- // 5 -N 0.0)
- // String[] args = new String[] {evalTask,"-l", "Clustream","-g",
- // "Clustream", "-i", "100000", "-s", "(RamdomRBFGeneratorsEvents",
- // "-K", "5", "-N", "0.0)"};
- // String[] args = new String[] { evalTask, "-l", "clustream.Clustream",
- // "-g", "clustream.Clustream", "-i", "100000", "-s",
- // "(RandomRBFGeneratorEvents", "-K", "5", "-N", "0.0)" };
- logger.debug("PARAMETERS {}", evalTask);
- // params = params.replace(":", " ");
- List<String> parameters = new ArrayList<String>();
- // parameters.add(evalTask);
- try {
- parameters.addAll(Arrays.asList(URLDecoder.decode(evalTask, "UTF-8").split(" ")));
- } catch (UnsupportedEncodingException ex) {
- ex.printStackTrace();
- }
- String[] args = parameters.toArray(new String[0]);
- Option[] extraOptions = new Option[] {};
- // build a single string by concatenating cli options
- StringBuilder cliString = new StringBuilder();
- for (int i = 0; i < args.length; i++) {
- cliString.append(" ").append(args[i]);
- }
-
- // parse options
- try {
- task = (Task) ClassOption.cliStringToObject(cliString.toString(), Task.class, extraOptions);
- task.setFactory(factory);
- task.init();
- } catch (Exception e) {
- e.printStackTrace();
- }
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.App#onStart()
- */
- @Override
- protected void onStart() {
- logger.info("Starting DoTaskApp... App Partition [{}]", this.getPartitionId());
- // <<<<<<< HEAD Task doesn't have start in latest storm-impl
- // TODO change the way the app starts
- // if (this.getPartitionId() == 0)
- S4Topology s4topology = (S4Topology) getTask().getTopology();
- S4EntranceProcessingItem epi = (S4EntranceProcessingItem) s4topology.getEntranceProcessingItem();
- while (epi.injectNextEvent())
- // inject events from the EntrancePI
- ;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.App#onClose()
- */
- @Override
- protected void onClose() {
- System.out.println("Closing DoTaskApp...");
-
- }
-
- /**
- * Gets the task.
- *
- * @return the task
- */
- public Task getTask() {
- return task;
- }
-
- // These methods are protected in App and can not be accessed from outside.
- // They are
- // called from parallel classifiers and evaluations. Is there a better way
- // to do that?
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.App#createPE(java.lang.Class)
- */
- @Override
- public <T extends ProcessingElement> T createPE(Class<T> type) {
- return super.createPE(type);
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.App#createStream(java.lang.String,
- * org.apache.s4.base.KeyFinder, org.apache.s4.core.ProcessingElement[])
- */
- @Override
- public <T extends Event> Stream<T> createStream(String name, KeyFinder<T> finder,
- ProcessingElement... processingElements) {
- return super.createStream(name, finder, processingElements);
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.s4.core.App#createStream(java.lang.String,
- * org.apache.s4.core.ProcessingElement[])
- */
- @Override
- public <T extends Event> Stream<T> createStream(String name, ProcessingElement... processingElements) {
- return super.createStream(name, processingElements);
- }
-
- // @com.beust.jcommander.Parameters(separators = "=")
- // class Parameters {
- //
- // @Parameter(names={"-lc","-local"}, description="Local clustering method")
- // private String localClustering;
- //
- // @Parameter(names={"-gc","-global"},
- // description="Global clustering method")
- // private String globalClustering;
- //
- // }
- //
- // class ParametersConverter {// implements IStringConverter<String[]> {
- //
- //
- // public String[] convertToArgs(String value) {
- //
- // String[] params = value.split(",");
- // String[] args = new String[params.length*2];
- // for(int i=0; i<params.length ; i++) {
- // args[i] = params[i].split("=")[0];
- // args[i+1] = params[i].split("=")[1];
- // i++;
- // }
- // return args;
- // }
- //
- // }
-
-}