| /* |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you 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. |
| */ |
| package org.apache.edgent.test.svt; |
| |
| import java.io.PrintWriter; |
| |
| /* |
| * Basic org.apache.edgent.org.apache.edgent.topology test to exercise TStream methods. |
| * |
| * Run from the edgent top-level directory: |
| * java -cp "test/svt/classes;target/java8/lib/*;target/java8/ext/slf4j-1.7.12/*" |
| * org.apache.edgent.test.svt.TopologyTestBasic <option> |
| * |
| * Option: |
| * console - specify to enable the console for this application. |
| * The console URL will be written into file consoleurl.txt. |
| * * |
| * To-Do: windowing |
| */ |
| |
| import java.util.Arrays; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Random; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.edgent.console.server.HttpServer; |
| import org.apache.edgent.metrics.Metrics; |
| import org.apache.edgent.providers.development.DevelopmentProvider; |
| import org.apache.edgent.topology.TStream; |
| import org.apache.edgent.topology.Topology; |
| |
| public class TopologyTestBasic { |
| public static void main(String[] args) throws Exception { |
| |
| System.out.println("TopologyTestBasic start"); |
| boolean console = false; |
| if (args.length == 1 && args[0].toLowerCase().equals("console")) |
| console = true; |
| |
| DevelopmentProvider tp = new DevelopmentProvider(); |
| |
| Topology t = tp.newTopology("TopologyTestBasic"); |
| Topology t2 = tp.newTopology("TopologyTestBasic2"); |
| |
| Metrics.counter(t); |
| Metrics.counter(t2); |
| |
| //************************************************************** |
| //Source 1 |
| //************************************************************** |
| Random r1 = new Random(); |
| TStream<Double> gaussian = t.poll(() -> r1.nextGaussian(), 100, TimeUnit.MILLISECONDS); |
| gaussian = Metrics.counter(gaussian); |
| |
| // testing Peek! |
| gaussian = gaussian.peek(g -> System.out.println("R1:" + g)); |
| |
| // A filter |
| gaussian = gaussian.filter(g -> g < 10000000); |
| gaussian.print(); |
| |
| // A modify |
| gaussian = gaussian.modify(g-> g*3 + 1); |
| |
| //Add counter |
| gaussian = Metrics.counter(gaussian); |
| |
| //A map |
| TStream<String> s1 = gaussian.map(g -> "g1: " + g.toString()).tag("s1", "gaussian"); |
| s1.sink(tuple -> {}); |
| |
| //A split into 11 streams |
| List<TStream<Double>> splits1 = gaussian.split(11, tuple -> { |
| switch (tuple.toString().charAt(0)) { |
| case '-': //negative numbers |
| return 10; |
| case '0': |
| return 0; |
| case '1': |
| return 1; |
| case '2': |
| return 2; |
| case '3': |
| return 3; |
| case '4': |
| return 4; |
| case '5': |
| return 5; |
| case '6': |
| return 6; |
| case '7': |
| return 7; |
| case '8': |
| return 8; |
| case '9': |
| return 9; |
| default: |
| return 10; |
| } |
| }); |
| |
| TStream<Double> sp0 = splits1.get(0).tag("split","sp0"); |
| sp0 = Metrics.counter(sp0); |
| Metrics.rateMeter(sp0); |
| sp0.print(); |
| |
| //Add a second tag |
| TStream<Double> sp0_1 = sp0.tag("split","sp0_1"); |
| sp0_1.print(); |
| |
| |
| TStream<Double> sp1 =splits1.get(1).tag("split","sp1"); |
| sp1 = Metrics.counter(sp1); |
| sp1.print(); |
| |
| TStream<Double> sp2 =splits1.get(2).tag("split","sp2"); |
| sp2 = Metrics.counter(sp2); |
| sp2.print(); |
| |
| TStream<Double> sp3 =splits1.get(3).tag("split","sp3"); |
| sp3 = Metrics.counter(sp3); |
| sp3.print(); |
| |
| TStream<Double> sp4 =splits1.get(4).tag("split","sp4"); |
| sp4 = Metrics.counter(sp4); |
| sp4.print(); |
| |
| TStream<Double> sp5 =splits1.get(5).tag("split","sp5"); |
| sp5 = Metrics.counter(sp1); |
| sp5.print(); |
| |
| TStream<Double> sp6 =splits1.get(6).tag("split","sp6"); |
| sp6 = Metrics.counter(sp6); |
| sp6.print(); |
| |
| TStream<Double> sp7 =splits1.get(7).tag("split","sp7"); |
| sp7 = Metrics.counter(sp7); |
| sp7.print(); |
| |
| TStream<Double> sp8 =splits1.get(8).tag("split","sp8"); |
| sp8 = Metrics.counter(sp8); |
| sp8.print(); |
| |
| TStream<Double> sp9 =splits1.get(9).tag("split","sp9"); |
| sp9 = Metrics.counter(sp9); |
| sp9.print(); |
| |
| TStream<Double> sp10 =splits1.get(10).tag("split","sp10"); |
| sp10 = Metrics.counter(sp10); |
| sp10.print(); |
| |
| |
| // Alternative 'split' functionality using 10 filters, then compare the results if possible. |
| TStream<Double> filter10 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '-' ).tag("split-"); |
| TStream<Double> filter0 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '0' ).tag("split0"); |
| TStream<Double> filter1 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '1' ).tag("split1"); |
| TStream<Double> filter2 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '2' ).tag("split2"); |
| TStream<Double> filter3 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '3' ).tag("split3"); |
| TStream<Double> filter4 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '4' ).tag("split4"); |
| TStream<Double> filter5 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '5' ).tag("split5"); |
| TStream<Double> filter6 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '6' ).tag("split6"); |
| TStream<Double> filter7 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '7' ).tag("split7"); |
| TStream<Double> filter8 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '8' ).tag("split8"); |
| TStream<Double> filter9 = gaussian.filter(tuple -> tuple.toString().charAt(0) == '9' ).tag("split9"); |
| |
| filter0 = filter0.peek(g -> System.out.println("filter0 : " + g)); |
| filter1 = filter1.peek(g -> System.out.println("filter1 : " + g)); |
| filter2 = filter2.peek(g -> System.out.println("filter2 : " + g)); |
| filter3 = filter3.peek(g -> System.out.println("filter3 : " + g)); |
| filter4 = filter4.peek(g -> System.out.println("filter4 : " + g)); |
| filter5 = filter5.peek(g -> System.out.println("filter5 : " + g)); |
| filter6 = filter6.peek(g -> System.out.println("filter6 : " + g)); |
| filter7 = filter7.peek(g -> System.out.println("filter7 : " + g)); |
| filter8 = filter8.peek(g -> System.out.println("filter8 : " + g)); |
| filter9 = filter9.peek(g -> System.out.println("filter9 : " + g)); |
| filter10 = filter10.peek(g -> System.out.println("filter10 : " + g)); |
| |
| filter0.print(); |
| filter1.print(); |
| filter2.print(); |
| filter3.print(); |
| filter4.print(); |
| filter5.print(); |
| filter6.print(); |
| filter7.print(); |
| filter8.print(); |
| filter9.print(); |
| filter10.print(); |
| |
| //************************************************************** |
| //Source 2 using complex tuple type |
| //************************************************************** |
| |
| Random r2 = new Random(); |
| TStream<MyClass1> mc1 = t.poll( |
| () -> new MyClass1(Double.toString(r2.nextGaussian()), |
| Double.toString(r2.nextGaussian()),r1.nextGaussian() |
| ),100, TimeUnit.MILLISECONDS).tag("mc1"); |
| |
| mc1.peek(g -> System.out.print(g.toString())); |
| |
| mc1.modify(tuple -> new MyClass1(tuple.getS1() + "a1 b1 c1 d1 ", tuple.getS2() +" e1 f1 g1 h1 ", tuple.getD1() +1) ); |
| mc1.peek(tuple -> System.out.println("MyClass1: " + tuple.toString())); |
| |
| mc1.flatMap(tuple -> Arrays.asList(tuple.toString().split(" "))); |
| |
| //An asString |
| |
| TStream<String> mcs1 = mc1.asString().tag("mcs1"); |
| mcs1.peek(tuple -> System.out.println(" mcs1_source2: " + tuple.toString())); |
| |
| List<TStream<String>> splits2 = mcs1.split(2, tuple -> { |
| switch (tuple.toString().charAt(0)) { |
| case '-': //negative numbers |
| return 0; |
| default: //everything else |
| return 1; |
| } |
| }); |
| |
| TStream<String> s2_0 = splits2.get(0).tag("s2_0"); |
| TStream<String> s2_1 = splits2.get(1).tag("s2_1"); |
| s2_0.sink(tuple -> System.out.println("s2_0: " + tuple.toString())); |
| s2_1.sink(tuple -> System.out.println("s2_1: " + tuple.toString())); |
| |
| //************************************************************** |
| // Source 3 - nested complex type |
| //************************************************************** |
| Random r3 = new Random(); |
| TStream<MyClass2> mc2 = t.poll( |
| () -> new MyClass2( |
| new MyClass1( |
| Double.toString(r3.nextGaussian()), |
| Double.toString(r3.nextGaussian()), |
| r3.nextGaussian()), |
| new MyClass1( |
| Double.toString(r3.nextGaussian()), |
| Double.toString(r3.nextGaussian()), |
| r3.nextGaussian()), |
| r3.nextGaussian(), |
| Double.toString(r3.nextGaussian()) |
| ), 100, TimeUnit.MILLISECONDS).tag("mc2"); |
| |
| // testing Peek! |
| mc2 = mc2.peek(tuple -> System.out.println("MyClass2_source3:" + tuple.toString())); |
| |
| // A filter |
| mc2 = mc2.filter(tuple -> |
| ( tuple.getMc1().getD1() > .5 && tuple.getMc2().getD1() < -.5 ) |
| || |
| ! tuple.getS1().startsWith("abc")); |
| |
| // modify |
| mc2 = mc2.modify( |
| tuple -> new MyClass2( |
| new MyClass1(tuple.getMc1().getS1() + " c3 d3 e3 f3 ", |
| tuple.getMc2().getS2() + "g3 h3 i3 j3 ", |
| tuple.getMc1().getD1() -13.3333), |
| new MyClass1(tuple.getMc2().getS2() + " x31 x32 x33 x34 x35 ", |
| tuple.getMc1().getS2() + " y31 y32 y33 y34 y35 ", |
| tuple.getMc2().getD1() +13.33), |
| tuple.getD1() *2 -.04556, |
| tuple.getS1() + " a31 b32 c33 d34 e35 " |
| ) |
| ).tag("mc2.modify"); |
| |
| //************************************************************** |
| // Source 4: Clone of Source 3 for now to generate more vertices. |
| //************************************************************** |
| Random r4 = new Random(); |
| TStream<MyClass2> mc4 = t.poll( |
| () -> new MyClass2( |
| new MyClass1( |
| Double.toString(r4.nextGaussian()), |
| Double.toString(r4.nextGaussian()), |
| r4.nextGaussian()), |
| new MyClass1(Double.toString(r4.nextGaussian()), |
| Double.toString(r4.nextGaussian()), |
| r4.nextGaussian()), |
| r4.nextGaussian(), |
| Double.toString(r4.nextGaussian()) |
| ), 100, TimeUnit.MILLISECONDS).tag("mc4"); |
| |
| // testing Peek! |
| mc4 = mc4.peek(tuple -> System.out.println("MyClass2_source4:" + tuple.toString())); |
| |
| // A filter |
| mc4 = mc4.filter(tuple -> |
| ( tuple.getMc1().getD1() > .5 && tuple.getMc2().getD1() < -.5 ) |
| || |
| ! tuple.getS1().startsWith("abc")); |
| |
| // modify |
| mc4 = mc4.modify( |
| tuple -> new MyClass2( |
| new MyClass1(tuple.getMc1().getS1() + " c41 d42 d43 f44 ", |
| tuple.getMc2().getS2() + "g41 h42 i43 j44 ", |
| tuple.getMc1().getD1() -17.03), |
| new MyClass1(tuple.getMc2().getS2() + " x41 x42 x43 x44 x45 ", |
| tuple.getMc1().getS2() + " y41 y42 y43 y44 y45 ", |
| tuple.getMc2().getD1() +14.4444), |
| tuple.getD1() *2 -.04556, |
| tuple.getS1() + " a b c d e " |
| ) |
| ); |
| |
| //Try to union mc4 to mc4 |
| TStream<MyClass2> su1 = mc4.union(new HashSet<>(Arrays.asList(mc2, mc4))); |
| su1.print(); |
| |
| //************************************************************** |
| // Source 5: Clone of Source 3 for now to generate more vertices. |
| //************************************************************** |
| Random r5 = new Random(); |
| TStream<MyClass2> mc5 = t2.poll( |
| () -> new MyClass2( |
| new MyClass1( |
| Double.toString(r5.nextGaussian()), |
| Double.toString(r5.nextGaussian()), |
| r5.nextGaussian()), |
| new MyClass1(Double.toString(r5.nextGaussian()), |
| Double.toString(r5.nextGaussian()), |
| r5.nextGaussian()), |
| r5.nextGaussian(), |
| Double.toString(r5.nextGaussian()) |
| ), 100, TimeUnit.MILLISECONDS); |
| |
| // Add a counter |
| mc5 = Metrics.counter(mc5); |
| |
| // testing Peek! |
| mc5 = mc5.peek(tuple -> System.out.println("MyClass2_source5:" + tuple.toString())); |
| |
| // A filter |
| mc5 = mc5.filter(tuple -> |
| ( tuple.getMc1().getD1() > .5 && tuple.getMc2().getD1() < -.5 ) |
| || |
| ! tuple.getS1().startsWith("abc")); |
| |
| // modify |
| mc5 = mc5.modify( |
| tuple -> new MyClass2( |
| new MyClass1(tuple.getMc1().getS1() + " c51 d52 e53 f54 ", |
| tuple.getMc2().getS2() + "g51 h52 i53 j54 ", |
| tuple.getMc1().getD1() -17.03), |
| new MyClass1(tuple.getMc2().getS2() + " x51 x52 x53 x54 x55 ", |
| tuple.getMc1().getS2() + " y51 y52 y53 y54 y55 ", |
| tuple.getMc2().getD1() +15.555), |
| tuple.getD1() *2 -.04556, |
| tuple.getS1() + " a51 b52 c53 d54 e55 ") |
| ).tag("mc5"); |
| |
| |
| mc5.sink(tuple -> {}); |
| |
| // Submit the jobs |
| tp.submit(t); |
| tp.submit(t2); |
| |
| |
| // If the console option was specified, write the console URL into file consoleUrl.txt |
| if (console) { |
| try { |
| PrintWriter writer = new PrintWriter("consoleUrl.txt", "UTF-8"); |
| writer.println(tp.getServices().getService(HttpServer.class).getConsoleUrl()); |
| writer.close(); |
| } catch ( Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| System.out.println("TopologyTestBasic end"); |
| |
| } |
| |
| } |
