@@ -0,0 +1,2543 @@
+/*%0A * Copyright (c) 2013 DataTorrent, Inc. ALL Rights Reserved.%0A *%0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A *%0A *   http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage com.datatorrent.demos.visualdata;%0A%0Aimport java.net.URI;%0A%0Aimport org.apache.commons.lang.StringUtils;%0Aimport org.apache.hadoop.conf.Configuration;%0A%0Aimport com.datatorrent.api.DAG;%0Aimport com.datatorrent.api.DAG.Locality;%0Aimport com.datatorrent.api.StreamingApplication;%0Aimport com.datatorrent.demos.pi.PiCalculateOperator;%0Aimport com.datatorrent.lib.io.ConsoleOutputOperator;%0Aimport com.datatorrent.lib.io.PubSubWebSocketOutputOperator;%0Aimport com.datatorrent.lib.testbench.RandomEventGenerator;%0A%0A/**%0A * Visual data demo.%0A */%0Apublic class Application implements StreamingApplication %7B%0A    private final Locality locality = null;%0A%0A    @Override%0A    public void populateDAG(DAG dag, Configuration conf) %7B%0A        dag.setAttribute(DAG.APPLICATION_NAME, %22VisualDataDemo%22);%0A        int maxValue = 30000;%0A%0A        RandomEventGenerator rand = dag.addOperator(%22random%22, new RandomEventGenerator());%0A        rand.setMinvalue(0);%0A        rand.setMaxvalue(maxValue);%0A%0A        PiCalculateOperator calc = dag.addOperator(%22picalc%22, new PiCalculateOperator());%0A        calc.setBase(maxValue * maxValue);%0A        dag.addStream(%22rand_calc%22, rand.integer_data, calc.input).setLocality(locality);%0A%0A        String gatewayAddress = dag.getValue(DAG.GATEWAY_ADDRESS);%0A        if (!StringUtils.isEmpty(gatewayAddress)) %7B%0A            URI uri = URI.create(%22ws://%22 + gatewayAddress + %22/pubsub%22);%0A%0A            PubSubWebSocketOutputOperator%3CObject%3E wsOut = dag.addOperator(%22wsOut%22,%0A                    new PubSubWebSocketOutputOperator%3CObject%3E());%0A            wsOut.setUri(uri);%0A            wsOut.setTopic(%22demos.visualdata.randomValue%22);%0A%0A            dag.addStream(%22randomdata%22, calc.output, wsOut.input);%0A        %7D else %7B%0A            ConsoleOutputOperator console = dag.addOperator(%22consoleout%22, new ConsoleOutputOperator());%0A            dag.addStream(%22rand_console%22, calc.output, console.input).setLocality(locality);%0A        %7D%0A    %7D%0A%0A%7D%0A

fix #MLHR-669 Visual Data Demo App