LogicalPlanConfiguration { public void prepareDAG(LogicalPlan dag, StreamingApplication app, String name) { prepareDAGAttributes(dag); pluginManager.setup(dag); if (app != null) { pluginManager.dispatch(PRE_POPULATE_DAG.event); app.populateDAG(dag, conf); pluginManager.dispatch(POST_POPULATE_DAG.event); } pluginManager.dispatch(PRE_CONFIGURE_DAG.event); String appAlias = getAppAlias(name); String appName = appAlias == null ? name : appAlias; List<AppConf> appConfs = stramConf.getMatchingChildConf(appName, StramElement.APPLICATION); setApplicationConfiguration(dag, appConfs, app); if (dag.getAttributes().get(Context.DAGContext.APPLICATION_NAME) == null) { dag.setAttribute(Context.DAGContext.APPLICATION_NAME, appName); } setModuleProperties(dag, appName); flattenDAG(dag, conf); setOperatorConfiguration(dag, appConfs, appName); setStreamConfiguration(dag, appConfs, appName); pluginManager.dispatch(POST_CONFIGURE_DAG.event); pluginManager.teardown(); } LogicalPlanConfiguration(Configuration conf); final void addFromConfiguration(Configuration conf); String getAppAlias(String appPath); LogicalPlanConfiguration addFromJson(JSONObject json, Configuration conf); LogicalPlanConfiguration addFromProperties(Properties props, Configuration conf); Properties getProperties(); Map<String, String> getAppAliases(); LogicalPlan createFromProperties(Properties props, String appName); LogicalPlan createFromJson(JSONObject json, String appName); LogicalPlan createEmptyForRecovery(String appName); LogicalPlan createFromStreamingApplication(StreamingApplication app, String appName); void populateDAG(LogicalPlan dag); void prepareDAG(LogicalPlan dag, StreamingApplication app, String name); static Properties readProperties(String filePath); Map<String, String> getProperties(OperatorMeta ow, String appName); static GenericOperator setOperatorProperties(GenericOperator operator, Map<String, String> properties); static StreamingApplication setApplicationProperties(StreamingApplication application, Map<String, String> properties); static BeanMap getObjectProperties(Object obj); void setOperatorProperties(LogicalPlan dag, String applicationName); void setModuleProperties(LogicalPlan dag, String applicationName); void setApplicationConfiguration(final LogicalPlan dag, String appName, StreamingApplication app); static final String GATEWAY_PREFIX; static final String GATEWAY_LISTEN_ADDRESS; static final String STREAM_PREFIX; static final String STREAM_SOURCE; static final String STREAM_SINKS; static final String STREAM_TEMPLATE; static final String STREAM_LOCALITY; static final String STREAM_SCHEMA; static final String OPERATOR_PREFIX; static final String OPERATOR_CLASSNAME; static final String OPERATOR_TEMPLATE; static final String TEMPLATE_idRegExp; static final String TEMPLATE_appNameRegExp; static final String TEMPLATE_classNameRegExp; static final String CLASS; static final String KEY_SEPARATOR; static final String KEY_SEPARATOR_SPLIT_REGEX; static final String KEY_APPLICATION_NAME; static final String KEY_GATEWAY_CONNECT_ADDRESS; static final String KEY_GATEWAY_USE_SSL; static final String KEY_GATEWAY_USER_NAME; static final String KEY_GATEWAY_PASSWORD; final Configuration conf; }
@Test public void testPrepareDAG() { final MutableBoolean appInitialized = new MutableBoolean(false); StreamingApplication app = new StreamingApplication() { @Override public void populateDAG(DAG dag, Configuration conf) { Assert.assertEquals("", "hostname:9090", dag.getValue(DAG.GATEWAY_CONNECT_ADDRESS)); dag.setAttribute(DAG.GATEWAY_CONNECT_ADDRESS, "hostname:9091"); appInitialized.setValue(true); } }; Configuration conf = new Configuration(false); conf.addResource(StramClientUtils.DT_SITE_XML_FILE); LogicalPlanConfiguration pb = new LogicalPlanConfiguration(conf); LogicalPlan dag = new LogicalPlan(); pb.prepareDAG(dag, app, "testconfig"); Assert.assertTrue("populateDAG called", appInitialized.booleanValue()); Assert.assertEquals("populateDAG overrides attribute", "hostname:9091", dag.getValue(DAG.GATEWAY_CONNECT_ADDRESS)); }
@Test public void testAppAnnotationAlias() { StreamingApplication app = new AnnotatedApplication(); Configuration conf = new Configuration(false); conf.addResource(StramClientUtils.DT_SITE_XML_FILE); LogicalPlanConfiguration builder = new LogicalPlanConfiguration(conf); LogicalPlan dag = new LogicalPlan(); String appPath = app.getClass().getName().replace(".", "/") + ".class"; builder.prepareDAG(dag, app, appPath); Assert.assertEquals("Application name", "AnnotatedAlias", dag.getAttributes().get(com.datatorrent.api.Context.DAGContext.APPLICATION_NAME)); }