OldFaithfulApplication implements StreamingApplication { @Override public void populateDAG(DAG dag, Configuration conf) { InputGenerator randomInputGenerator = dag.addOperator("rand", new InputGenerator()); FaithfulRScript rScriptOp = dag.addOperator("rScriptOp", new FaithfulRScript("com/datatorrent/examples/r/oldfaithful/eruptionModel.R", "eruptionModel", "retVal")); ConsoleOutputOperator consoles = dag.addOperator("consoles", new ConsoleOutputOperator()); Map<String, FaithfulRScript.REXP_TYPE> argTypeMap = new HashMap<String, FaithfulRScript.REXP_TYPE>(); argTypeMap.put("ELAPSEDTIME", FaithfulRScript.REXP_TYPE.REXP_INT); argTypeMap.put("ERUPTIONS", FaithfulRScript.REXP_TYPE.REXP_ARRAY_DOUBLE); argTypeMap.put("WAITING", FaithfulRScript.REXP_TYPE.REXP_ARRAY_INT); rScriptOp.setArgTypeMap(argTypeMap); dag.addStream("ingen_faithfulRscript", randomInputGenerator.outputPort, rScriptOp.faithfulInput).setLocality(locality); dag.addStream("ingen_faithfulRscript_eT", randomInputGenerator.elapsedTime, rScriptOp.inputElapsedTime).setLocality(locality); dag.addStream("faithfulRscript_console_s", rScriptOp.strOutput, consoles.input).setLocality(locality); }  @Override void populateDAG(DAG dag, Configuration conf);  }
@Test public void testSomeMethod() throws Exception { LocalMode lma = LocalMode.newInstance(); OldFaithfulApplication app = new OldFaithfulApplication(); app.populateDAG(lma.getDAG(), new Configuration(false)); try { LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(false); lc.run(5000); } catch (Exception e) { LOG.error("Exception: ", e); Assert.fail("Unexpected exception."); } }