You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dj...@apache.org on 2016/03/16 14:34:52 UTC
[5/6] incubator-quarks git commit: Using explicit imports instead of
Starred, Modify lambda, Added testcase
Using explicit imports instead of Starred, Modify lambda, Added testcase
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/c3308be9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/c3308be9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/c3308be9
Branch: refs/heads/master
Commit: c3308be9d69fecf2f2862d4b65a18a232eaf2d5e
Parents: dc23b7a
Author: cazen.lee <ca...@samsung.com>
Authored: Wed Mar 16 17:34:00 2016 +0900
Committer: Cazen <Ca...@korea.com>
Committed: Wed Mar 16 20:56:16 2016 +0900
----------------------------------------------------------------------
.../src/main/java/quarks/topology/TStream.java | 8 +++-
.../java/quarks/test/topology/TStreamTest.java | 48 ++++++++++++++++----
.../samples/topology/SplitWithEnumSample.java | 8 ++--
.../topology/spi/graph/ConnectorStream.java | 47 +++++++++----------
4 files changed, 75 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/c3308be9/api/topology/src/main/java/quarks/topology/TStream.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TStream.java b/api/topology/src/main/java/quarks/topology/TStream.java
index 355504c..e40c650 100644
--- a/api/topology/src/main/java/quarks/topology/TStream.java
+++ b/api/topology/src/main/java/quarks/topology/TStream.java
@@ -4,7 +4,11 @@
*/
package quarks.topology;
-import quarks.function.*;
+import quarks.function.Consumer;
+import quarks.function.Function;
+import quarks.function.Predicate;
+import quarks.function.ToIntFunction;
+import quarks.function.UnaryOperator;
import quarks.oplet.core.Pipe;
import quarks.oplet.core.Sink;
@@ -202,6 +206,8 @@ public interface TStream<T> extends TopologyElement {
* @param splitter
* the splitter function
* @return EnumMap<E,TStream<T>>
+ * @throws IllegalArgumentException
+ * if {@code enumclass.size <= 0}
*/
<E extends Enum<E>> EnumMap<E,TStream<T>> split(Class<E> enumClass, Function<T, E> splitter);
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/c3308be9/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
----------------------------------------------------------------------
diff --git a/api/topology/src/test/java/quarks/test/topology/TStreamTest.java b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
index 5f45525..9168a42 100644
--- a/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
+++ b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
@@ -4,20 +4,36 @@
*/
package quarks.test.topology;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
import org.junit.Ignore;
import org.junit.Test;
+
import quarks.topology.TSink;
import quarks.topology.TStream;
import quarks.topology.Topology;
import quarks.topology.tester.Condition;
-import java.io.Serializable;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static org.junit.Assert.*;
-
@Ignore
public abstract class TStreamTest extends TopologyAbstractTest {
@@ -274,7 +290,7 @@ public abstract class TStreamTest extends TopologyAbstractTest {
* Test enum data structure
*/
private enum LogSeverityEnum {
- EMERG(0), ALERT(1), CRITICAL(2), ERROR(3), WARNING(4), NOTICE(5), INFO(6), DEBUG(7);
+ ALERT(1), CRITICAL(2), ERROR(3), WARNING(4), NOTICE(5), INFO(6), DEBUG(7);
private final int code;
@@ -321,6 +337,22 @@ public abstract class TStreamTest extends TopologyAbstractTest {
assertTrue(contents4.toString(), contents4.valid());
}
+ private enum EnumClassWithZerosize {
+ ;
+ }
+
+ /**
+ * Test split(enum) with integer type enum.
+ */
+ @Test(expected = IllegalArgumentException.class)
+ public void testSplitWithEnumForZeroSizeClass() throws Exception {
+
+ Topology t = newTopology();
+
+ TStream<String> s = t.strings("Test");
+ s.split(EnumClassWithZerosize.class, e -> EnumClassWithZerosize.valueOf("Test"));
+ }
+
@Test
public void testFanout2() throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/c3308be9/samples/topology/src/main/java/quarks/samples/topology/SplitWithEnumSample.java
----------------------------------------------------------------------
diff --git a/samples/topology/src/main/java/quarks/samples/topology/SplitWithEnumSample.java b/samples/topology/src/main/java/quarks/samples/topology/SplitWithEnumSample.java
index 2545ecf..0c18302 100644
--- a/samples/topology/src/main/java/quarks/samples/topology/SplitWithEnumSample.java
+++ b/samples/topology/src/main/java/quarks/samples/topology/SplitWithEnumSample.java
@@ -52,9 +52,9 @@ public class SplitWithEnumSample {
EnumMap<LogSeverityEnum, TStream<String>> categories = d
.split(LogSeverityEnum.class, e -> LogSeverityEnum.valueOf(e.split("_")[0]));
- TStream<String> warnStream = categories.get(LogSeverityEnum.WARNING);
- TStream<String> errorStream = categories.get(LogSeverityEnum.ERROR);
- TStream<String> infoStream = categories.get(LogSeverityEnum.INFO);
+ TStream<String> warnStream = categories.get(LogSeverityEnum.WARNING).tag("WARN");
+ TStream<String> errorStream = categories.get(LogSeverityEnum.ERROR).tag("ERROR");
+ TStream<String> infoStream = categories.get(LogSeverityEnum.INFO).tag("INFO");
warnStream.sink(data -> System.out.println("warnStream = " + data));
errorStream.sink(data -> System.out.println("errorStream = " + data));
@@ -64,4 +64,4 @@ public class SplitWithEnumSample {
System.out.println(dtp.getServices().getService(HttpServer.class).getConsoleUrl());
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/c3308be9/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
----------------------------------------------------------------------
diff --git a/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java b/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
index 5dc2902..a97d2b0 100644
--- a/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
+++ b/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
@@ -4,7 +4,20 @@
*/
package quarks.topology.spi.graph;
-import quarks.function.*;
+import static quarks.function.Functions.synchronizedFunction;
+
+import java.util.ArrayList;
+import java.util.EnumMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import quarks.function.Consumer;
+import quarks.function.Function;
+import quarks.function.Functions;
+import quarks.function.Predicate;
+import quarks.function.ToIntFunction;
import quarks.graph.Connector;
import quarks.graph.Graph;
import quarks.graph.Vertex;
@@ -22,11 +35,6 @@ import quarks.topology.TWindow;
import quarks.topology.Topology;
import quarks.topology.spi.AbstractTStream;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
-import static quarks.function.Functions.synchronizedFunction;
-
/**
* A stream that directly adds oplets to the graph.
*
@@ -92,26 +100,19 @@ public class ConnectorStream<G extends Topology, T> extends AbstractTStream<G, T
public <E extends Enum<E>> EnumMap<E,TStream<T>> split(Class<E> enumClass, Function<T, E> splitter) {
E[] es = enumClass.getEnumConstants();
+ if(es == null) {
+ throw new IllegalArgumentException("Class object does not represent an enum type");
+ }
-/*
- List<TStream<T>> outputs = split(es.length, e -> IntStream.range(0, es.length).filter(i-> es[i].equals(splitter.apply(e)))
- .mapToObj(i -> es[i].ordinal()).findAny().orElse(-1));
-*/
-
- List<TStream<T>> outputs = split(es.length, new ToIntFunction<T>() {
- @Override
- public int applyAsInt(T input) {
- for(int i = 0; i < es.length; i++){
- if(es[i].equals(splitter.apply(input))){
- return es[i].ordinal();
- }
- }
- return -1;
- }});
+ List<TStream<T>> outputs = split(es.length, t -> {
+ E split = splitter.apply(t);
+ return split != null ? split.ordinal() : -1;
+ });
EnumMap<E,TStream<T>> returnMap = new EnumMap<>(enumClass);
- for (int i = 0; i < es.length ; i++){
- returnMap.put(es[i], outputs.get(es[i].ordinal()));
+
+ for (E e : es) {
+ returnMap.put(e, outputs.get(e.ordinal()));
}
return returnMap;