From 406c4f52d8251eb81aa868ec38f17789e18e0dc0 Mon Sep 17 00:00:00 2001 From: wusheng Date: Wed, 8 Nov 2017 19:52:17 +0800 Subject: [PATCH] Add concept: WayToNode to define how to ask node to process. --- .../apm/collector/core/graph/DirectWay.java | 32 +++++++++++++ .../apm/collector/core/graph/Next.java | 10 ++--- .../apm/collector/core/graph/Node.java | 10 +++-- .../apm/collector/core/graph/WayToNode.java | 45 +++++++++++++++++++ .../core/graph/GraphManagerTest.java | 16 +++++++ 5 files changed, 105 insertions(+), 8 deletions(-) create mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/DirectWay.java create mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/WayToNode.java diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/DirectWay.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/DirectWay.java new file mode 100644 index 000000000..bdf85745f --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/DirectWay.java @@ -0,0 +1,32 @@ +/* + * 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 DirectWay extends WayToNode { + public DirectWay(NodeProcessor destinationHandler) { + super(destinationHandler); + } + + @Override protected void in(INPUT o) { + out(o); + } +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Next.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Next.java index ba8e6c049..31ddba477 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Next.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Next.java @@ -29,14 +29,14 @@ import org.skywalking.apm.collector.core.framework.Executor; */ public class Next implements Executor { - private final List nextNodes; + private final List ways; public Next() { - this.nextNodes = new LinkedList<>(); + this.ways = new LinkedList<>(); } - public final void addNext(Node node) { - nextNodes.add(node); + final void addWay(WayToNode way) { + ways.add(way); } /** @@ -45,6 +45,6 @@ public class Next implements Executor { * @param INPUT */ @Override public void execute(INPUT INPUT) { - nextNodes.forEach(node -> node.execute(INPUT)); + ways.forEach(way -> way.in(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 index 39f2c7d7c..e5bda3038 100644 --- 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 @@ -36,10 +36,14 @@ public final class Node { } public final Node addNext(NodeProcessor nodeProcessor) { + return this.addNext(new DirectWay(nodeProcessor)); + } + + public final Node addNext(WayToNode way) { synchronized (graph) { - Node node = new Node<>(graph, nodeProcessor); - next.addNext(node); - return node; + way.buildDestination(graph); + next.addWay(way); + return way.getDestination(); } } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/WayToNode.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/WayToNode.java new file mode 100644 index 000000000..19e30df2c --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/WayToNode.java @@ -0,0 +1,45 @@ +/* + * 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 abstract class WayToNode { + private Node destination; + private NodeProcessor destinationHandler; + + public WayToNode(NodeProcessor destinationHandler) { + this.destinationHandler = destinationHandler; + } + + void buildDestination(Graph graph) { + destination = new Node(graph, destinationHandler); + } + + protected abstract void in(INPUT INPUT); + + protected void out(INPUT INPUT) { + destination.execute(INPUT); + } + + Node getDestination() { + return destination; + } +} 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 index 0da098837..42f8b334f 100644 --- 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 @@ -114,6 +114,22 @@ public class GraphManagerTest { foundNode.addNext(new Node4Processor()); } + @Test + public void testDeadEndWay() { + Graph graph = GraphManager.INSTANCE.createIfAbsent(7, String.class); + graph.addNode(new Node1Processor()).addNext(new WayToNode(new Node2Processor()) { + @Override protected void in(String INPUT) { + //don't call `out(intput)`; + } + }); + + graph.start("Input String"); + String output = outputStream.toString(); + String expected = "Node1 process: s=Input String" + lineSeparator; + + Assert.assertEquals(expected, output); + } + @After public void tearDown() { GraphManager.INSTANCE.reset();