diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Graph.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Graph.java
new file mode 100644
index 000000000..8c1f88251
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Graph.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author peng-yongsheng, wu-sheng
+ */
+public final class Graph nodeIndex = new ConcurrentHashMap<>();
+
+ Graph(int id) {
+ this.id = id;
+ }
+
+ public void start(Input input) {
+ startNode.execute(input);
+ }
+
+ public Node addNode(NodeHandler nodeHandler) {
+ synchronized (this) {
+ startNode = new Node(this, nodeHandler);
+ return startNode;
+ }
+ }
+
+ public Next findNext(int handlerId) {
+ Node node = nodeIndex.get(handlerId);
+ if (node == null) {
+ throw new NodeNotFoundException("Can't find node with handlerId="
+ + handlerId
+ + " in graph[" + id + "】");
+ }
+ return node.getNext();
+ }
+
+ void checkForNewNode(Node node) {
+ int nodeId = node.getHandler().id();
+ if (nodeIndex.containsKey(nodeId)) {
+ throw new PotentialAcyclicGraphException("handler="
+ + node.getHandler().getClass().getName()
+ + " already exists in graph[" + id + "】");
+ }
+ nodeIndex.put(nodeId, node);
+ }
+
+}
diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/GraphManager.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/GraphManager.java
new file mode 100644
index 000000000..a77d9de9b
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/GraphManager.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author wusheng
+ */
+public enum GraphManager {
+ INSTANCE;
+
+ private Map allGraphs = new HashMap<>();
+
+ /**
+ * Create a stream process graph.
+ *
+ * @param graphId represents a graph, which is used for finding it.
+ * @return
+ */
+ public synchronized Next is a delegate object for the following {@link Node}.
+ *
+ * @author peng-yongsheng, wu-sheng
*/
-public class Next nextNodes;
public Next() {
- this.nextNodes = new ArrayList<>();
+ this.nextNodes = new LinkedList<>();
}
public final void addNext(Node node) {
nextNodes.add(node);
}
+ /**
+ * Drive to the next nodes
+ *
+ * @param input
+ */
@Override public void execute(Input input) {
nextNodes.forEach(node -> node.execute(input));
}
diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Node.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Node.java
new file mode 100644
index 000000000..8a947bb34
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Node.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+/**
+ * The Node in the graph with explicit Input and Output types.
+ *
+ * @author peng-yongsheng, wu-sheng
+ */
+public final class Node {
+ private final NodeHandler nodeHandler;
+ private final Next next;
+ private final Graph graph;
+
+ Node(Graph graph, NodeHandler nodeHandler) {
+ this.graph = graph;
+ this.nodeHandler = nodeHandler;
+ this.next = new Next<>();
+ this.graph.checkForNewNode(this);
+ }
+
+ public final Node addNext(NodeHandler nodeHandler) {
+ synchronized (graph) {
+ Node node = new Node<>(graph, nodeHandler);
+ next.addNext(node);
+ return node;
+ }
+ }
+
+ final void execute(Input input) {
+ nodeHandler.process(input, next);
+ }
+
+ NodeHandler getHandler() {
+ return nodeHandler;
+ }
+
+ Next getNext() {
+ return next;
+ }
+}
diff --git a/apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/Aggregator.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeHandler.java
similarity index 77%
rename from apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/Aggregator.java
rename to apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeHandler.java
index ab10cb29d..10b1d63fb 100644
--- a/apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/Aggregator.java
+++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeHandler.java
@@ -16,13 +16,18 @@
* Project repository: https://github.com/OpenSkywalking/skywalking
*/
-package org.skywalking.apm.collector.stream.graph;
-
-import org.skywalking.apm.collector.core.data.Data;
+package org.skywalking.apm.collector.core.graph;
/**
- * @author peng-yongsheng
+ * @author peng-yongsheng, wu-sheng
*/
-public interface Aggregator {
+ /**
+ * The unique id in the certain graph.
+ *
+ * @return id
+ */
+ int id();
+
void process(Input input, Next next);
}
diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeNotFoundException.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeNotFoundException.java
new file mode 100644
index 000000000..dc2a27a4f
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeNotFoundException.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+/**
+ * @author wusheng
+ */
+public class NodeNotFoundException extends RuntimeException {
+ public NodeNotFoundException(String message) {
+ super(message);
+ }
+}
diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/PotentialAcyclicGraphException.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/PotentialAcyclicGraphException.java
new file mode 100644
index 000000000..91d06790f
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/PotentialAcyclicGraphException.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+/**
+ * @author wusheng
+ */
+public class PotentialAcyclicGraphException extends RuntimeException {
+ public PotentialAcyclicGraphException(String message) {
+ super(message);
+ }
+}
diff --git a/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/GraphManagerTest.java b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/GraphManagerTest.java
new file mode 100644
index 000000000..11a13490e
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/GraphManagerTest.java
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author wusheng
+ */
+public class GraphManagerTest {
+ private static PrintStream OUT_REF;
+ private ByteArrayOutputStream outputStream;
+
+ @Before
+ public void initAndHoldOut() {
+ OUT_REF = System.out;
+ outputStream = new ByteArrayOutputStream();
+ PrintStream testStream = new PrintStream(outputStream);
+ System.setOut(testStream);
+ }
+
+ @After
+ public void reset() {
+ System.setOut(OUT_REF);
+ }
+
+ @Test
+ public void testGraph() {
+ Graph testGraph = GraphManager.INSTANCE.createIfAbsent(1, String.class);
+ Node node = testGraph.addNode(new Node1Handler());
+ Node node1 = node.addNext(new Node2Handler());
+ testGraph.start("Input String");
+
+ String output = outputStream.toString();
+ String expected = "Node1 process: s=Input String\n" +
+ "Node2 process: s=Input String\n";
+
+ Assert.assertEquals(expected, output);
+ }
+
+ @Test
+ public void testGraphWithChainStyle() {
+ Graph graph = GraphManager.INSTANCE.createIfAbsent(2, String.class);
+ graph.addNode(new Node1Handler()).addNext(new Node2Handler()).addNext(new Node4Handler());
+
+ graph.start("Input String");
+
+ String output = outputStream.toString();
+ String expected = "Node1 process: s=Input String\n" +
+ "Node2 process: s=Input String\n" +
+ "Node4 process: int=123\n";
+
+ Assert.assertEquals(expected, output);
+ }
+
+ @Test(expected = PotentialAcyclicGraphException.class)
+ public void testPotentialAcyclicGraph() {
+ Graph testGraph = GraphManager.INSTANCE.createIfAbsent(3, String.class);
+ Node node = testGraph.addNode(new Node1Handler());
+ node.addNext(new Node1Handler());
+ }
+
+ @Test
+ public void testContinueStream() {
+ Graph graph = GraphManager.INSTANCE.createIfAbsent(4, String.class);
+ graph.addNode(new Node1Handler()).addNext(new Node2Handler()).addNext(new Node4Handler());
+
+ Next next = GraphManager.INSTANCE.findGraph(4).findNext(2);
+
+ next.execute(123);
+ String output = outputStream.toString();
+ String expected =
+ "Node4 process: int=123\n";
+
+ Assert.assertEquals(expected, output);
+ }
+
+ @Test(expected = NodeNotFoundException.class)
+ public void handlerNotFound() {
+ Graph graph = GraphManager.INSTANCE.createIfAbsent(5, String.class);
+ graph.addNode(new Node1Handler()).addNext(new Node2Handler()).addNext(new Node4Handler());
+
+ Next next = GraphManager.INSTANCE.findGraph(5).findNext(3);
+ }
+}
diff --git a/apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/StreamGraph.java b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node1Handler.java
similarity index 66%
rename from apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/StreamGraph.java
rename to apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node1Handler.java
index a9c3209c6..0a00d5a55 100644
--- a/apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/StreamGraph.java
+++ b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node1Handler.java
@@ -16,23 +16,18 @@
* Project repository: https://github.com/OpenSkywalking/skywalking
*/
-package org.skywalking.apm.collector.stream.graph;
-
-import org.skywalking.apm.collector.core.data.Data;
+package org.skywalking.apm.collector.core.graph;
/**
- * @author peng-yongsheng
+ * @author wusheng
*/
-public class StreamGraph {
-
- private Node startNode;
-
- public void start(Data input) {
- startNode.execute(input);
+public class Node1Handler implements NodeHandler {
+ @Override public int id() {
+ return 1;
}
- public Node addNode(Aggregator aggregator) {
- startNode = new Node(aggregator);
- return startNode;
+ @Override public void process(String s, Next next) {
+ System.out.println("Node1 process: s=" + s);
+ next.execute(s);
}
}
diff --git a/apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/Node.java b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node2Handler.java
similarity index 55%
rename from apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/Node.java
rename to apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node2Handler.java
index 148d20a05..75adafae2 100644
--- a/apm-collector/apm-collector-stream/collector-stream-define/src/main/java/org/skywalking/apm/collector/stream/graph/Node.java
+++ b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node2Handler.java
@@ -16,30 +16,18 @@
* Project repository: https://github.com/OpenSkywalking/skywalking
*/
-package org.skywalking.apm.collector.stream.graph;
-
-import org.skywalking.apm.collector.core.data.Data;
+package org.skywalking.apm.collector.core.graph;
/**
- * @author peng-yongsheng
+ * @author wusheng
*/
-public class Node {
+ @Override public int id() {
+ return 2;
}
- public final Node addNext(Aggregator aggregator) {
- Node node = new Node(aggregator);
- next.addNext(node);
- return node;
- }
-
- final void execute(Input input) {
- aggregator.process(input, next);
+ @Override public void process(String s, Next next) {
+ System.out.println("Node2 process: s=" + s);
+ next.execute(123);
}
}
diff --git a/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node3Handler.java b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node3Handler.java
new file mode 100644
index 000000000..eb063a610
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node3Handler.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+/**
+ * @author wusheng
+ */
+public class Node3Handler implements NodeHandler {
+ @Override public int id() {
+ return 3;
+ }
+
+ @Override
+ public void process(Long aLong, Next next) {
+ System.out.println("Node3 process: long=" + aLong);
+ next.execute(aLong);
+ }
+}
diff --git a/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node4Handler.java b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node4Handler.java
new file mode 100644
index 000000000..5bf7d4e0e
--- /dev/null
+++ b/apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node4Handler.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * 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.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+/**
+ * @author wusheng
+ */
+public class Node4Handler implements NodeHandler {
+ @Override public int id() {
+ return 4;
+ }
+
+ @Override
+ public void process(Integer in, Next next) {
+ System.out.println("Node4 process: int=" + in);
+ next.execute(new Long(in.intValue()));
+ }
+}