Support WayToNode for entry node.

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Graph.java

@@ -25,7 +25,7 @@ import java.util.concurrent.ConcurrentHashMap;
  */
 public final class Graph<INPUT> {
     private int id;
-    private Node startNode;
+    private WayToNode entryWay;
     private ConcurrentHashMap<Integer, Node> nodeIndex = new ConcurrentHashMap<>();
 
     Graph(int id) {
@@ -33,13 +33,18 @@ public final class Graph<INPUT> {
     }
 
     public void start(INPUT INPUT) {
-        startNode.execute(INPUT);
+        entryWay.in(INPUT);
     }
 
     public <OUTPUT> Node<INPUT, OUTPUT> addNode(NodeProcessor<INPUT, OUTPUT> nodeProcessor) {
+        return addNode(new DirectWay(nodeProcessor));
+    }
+
+    public <OUTPUT> Node<INPUT, OUTPUT> addNode(WayToNode<INPUT, OUTPUT> entryWay) {
         synchronized (this) {
-            startNode = new Node(this, nodeProcessor);
-            return startNode;
+            this.entryWay = entryWay;
+            this.entryWay.buildDestination(this);
+            return entryWay.getDestination();
         }
     }
 

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/GraphManagerTest.java

@@ -130,6 +130,20 @@ public class GraphManagerTest {
         Assert.assertEquals(expected, output);
     }
 
+    @Test
+    public void testEntryWay() {
+        Graph<String> graph = GraphManager.INSTANCE.createIfAbsent(8, String.class);
+        graph.addNode(new WayToNode<String, String>(new Node1Processor()) {
+            @Override protected void in(String INPUT) {
+                //don't call `out(intput)`;
+            }
+        }).addNext(new Node2Processor());
+
+        graph.start("Input String");
+
+        Assert.assertEquals("", outputStream.toString());
+    }
+
     @After
     public void tearDown() {
         GraphManager.INSTANCE.reset();