@Test public void testCreateFlowSpecForConfig() throws URISyntaxException { Map<String, String> flowProperties = Maps.newHashMap(); flowProperties.put("param1", "a:b:c*.d"); FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME)) .setTemplateUris(TEST_TEMPLATE_URI).setSchedule(new Schedule().setCronSchedule(TEST_SCHEDULE). setRunImmediately(true)) .setProperties(new StringMap(flowProperties)); FlowSpec flowSpec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(flowConfig); Assert.assertEquals(flowSpec.getConfig().getString(ConfigurationKeys.FLOW_GROUP_KEY), TEST_GROUP_NAME); Assert.assertEquals(flowSpec.getConfig().getString(ConfigurationKeys.FLOW_NAME_KEY), TEST_FLOW_NAME); Assert.assertEquals(flowSpec.getConfig().getString(ConfigurationKeys.JOB_SCHEDULE_KEY), TEST_SCHEDULE); Assert.assertEquals(flowSpec.getConfig().getBoolean(ConfigurationKeys.FLOW_RUN_IMMEDIATELY), true); Assert.assertEquals(flowSpec.getConfig().getString("param1"), "a:b:c*.d"); Assert.assertEquals(flowSpec.getTemplateURIs().get().size(), 1); Assert.assertTrue(flowSpec.getTemplateURIs().get().contains(new URI(TEST_TEMPLATE_URI))); flowProperties = Maps.newHashMap(); flowProperties.put("param1", "value1"); flowProperties.put("param2", "${param1}-123"); flowProperties.put("param3", "\"a:b:c*.d\""); flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME)) .setTemplateUris(TEST_TEMPLATE_URI).setSchedule(new Schedule().setCronSchedule(TEST_SCHEDULE). setRunImmediately(true)) .setProperties(new StringMap(flowProperties)); flowSpec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(flowConfig); Assert.assertEquals(flowSpec.getConfig().getString(ConfigurationKeys.FLOW_GROUP_KEY), TEST_GROUP_NAME); Assert.assertEquals(flowSpec.getConfig().getString(ConfigurationKeys.FLOW_NAME_KEY), TEST_FLOW_NAME); Assert.assertEquals(flowSpec.getConfig().getString(ConfigurationKeys.JOB_SCHEDULE_KEY), TEST_SCHEDULE); Assert.assertEquals(flowSpec.getConfig().getBoolean(ConfigurationKeys.FLOW_RUN_IMMEDIATELY), true); Assert.assertEquals(flowSpec.getConfig().getString("param1"),"value1"); Assert.assertEquals(flowSpec.getConfig().getString("param2"),"value1-123"); Assert.assertEquals(flowSpec.getConfig().getString("param3"), "a:b:c*.d"); Assert.assertEquals(flowSpec.getTemplateURIs().get().size(), 1); Assert.assertTrue(flowSpec.getTemplateURIs().get().contains(new URI(TEST_TEMPLATE_URI))); }
public static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig) { ConfigBuilder configBuilder = ConfigBuilder.create() .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, flowConfig.getId().getFlowGroup()) .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, flowConfig.getId().getFlowName()); if (flowConfig.hasSchedule()) { Schedule schedule = flowConfig.getSchedule(); configBuilder.addPrimitive(ConfigurationKeys.JOB_SCHEDULE_KEY, schedule.getCronSchedule()); configBuilder.addPrimitive(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, schedule.isRunImmediately()); } else { String flowExecutionId; if (flowConfig.getProperties().containsKey(ConfigurationKeys.FLOW_EXECUTION_ID_KEY)) { flowExecutionId = flowConfig.getProperties().get(ConfigurationKeys.FLOW_EXECUTION_ID_KEY); log.info("Using the existing flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } else { flowExecutionId = String.valueOf(System.currentTimeMillis()); log.info("Created a flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } flowConfig.getProperties().put(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); } if (flowConfig.hasExplain()) { configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXPLAIN_KEY, flowConfig.isExplain()); } Config config = configBuilder.build(); Config configWithFallback; try { configWithFallback = config.withFallback(ConfigFactory.parseString(flowConfig.getProperties().toString()).resolve()); } catch (Exception e) { configWithFallback = config.withFallback(ConfigFactory.parseMap(flowConfig.getProperties())); } try { URI templateURI = new URI(flowConfig.getTemplateUris()); return FlowSpec.builder().withConfig(configWithFallback).withTemplate(templateURI).build(); } catch (URISyntaxException e) { throw new FlowConfigLoggedException(HttpStatus.S_400_BAD_REQUEST, "bad URI " + flowConfig.getTemplateUris(), e); } }
FlowConfigResourceLocalHandler implements FlowConfigsResourceHandler { public static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig) { ConfigBuilder configBuilder = ConfigBuilder.create() .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, flowConfig.getId().getFlowGroup()) .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, flowConfig.getId().getFlowName()); if (flowConfig.hasSchedule()) { Schedule schedule = flowConfig.getSchedule(); configBuilder.addPrimitive(ConfigurationKeys.JOB_SCHEDULE_KEY, schedule.getCronSchedule()); configBuilder.addPrimitive(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, schedule.isRunImmediately()); } else { String flowExecutionId; if (flowConfig.getProperties().containsKey(ConfigurationKeys.FLOW_EXECUTION_ID_KEY)) { flowExecutionId = flowConfig.getProperties().get(ConfigurationKeys.FLOW_EXECUTION_ID_KEY); log.info("Using the existing flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } else { flowExecutionId = String.valueOf(System.currentTimeMillis()); log.info("Created a flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } flowConfig.getProperties().put(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); } if (flowConfig.hasExplain()) { configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXPLAIN_KEY, flowConfig.isExplain()); } Config config = configBuilder.build(); Config configWithFallback; try { configWithFallback = config.withFallback(ConfigFactory.parseString(flowConfig.getProperties().toString()).resolve()); } catch (Exception e) { configWithFallback = config.withFallback(ConfigFactory.parseMap(flowConfig.getProperties())); } try { URI templateURI = new URI(flowConfig.getTemplateUris()); return FlowSpec.builder().withConfig(configWithFallback).withTemplate(templateURI).build(); } catch (URISyntaxException e) { throw new FlowConfigLoggedException(HttpStatus.S_400_BAD_REQUEST, "bad URI " + flowConfig.getTemplateUris(), e); } } }
FlowConfigResourceLocalHandler implements FlowConfigsResourceHandler { public static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig) { ConfigBuilder configBuilder = ConfigBuilder.create() .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, flowConfig.getId().getFlowGroup()) .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, flowConfig.getId().getFlowName()); if (flowConfig.hasSchedule()) { Schedule schedule = flowConfig.getSchedule(); configBuilder.addPrimitive(ConfigurationKeys.JOB_SCHEDULE_KEY, schedule.getCronSchedule()); configBuilder.addPrimitive(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, schedule.isRunImmediately()); } else { String flowExecutionId; if (flowConfig.getProperties().containsKey(ConfigurationKeys.FLOW_EXECUTION_ID_KEY)) { flowExecutionId = flowConfig.getProperties().get(ConfigurationKeys.FLOW_EXECUTION_ID_KEY); log.info("Using the existing flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } else { flowExecutionId = String.valueOf(System.currentTimeMillis()); log.info("Created a flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } flowConfig.getProperties().put(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); } if (flowConfig.hasExplain()) { configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXPLAIN_KEY, flowConfig.isExplain()); } Config config = configBuilder.build(); Config configWithFallback; try { configWithFallback = config.withFallback(ConfigFactory.parseString(flowConfig.getProperties().toString()).resolve()); } catch (Exception e) { configWithFallback = config.withFallback(ConfigFactory.parseMap(flowConfig.getProperties())); } try { URI templateURI = new URI(flowConfig.getTemplateUris()); return FlowSpec.builder().withConfig(configWithFallback).withTemplate(templateURI).build(); } catch (URISyntaxException e) { throw new FlowConfigLoggedException(HttpStatus.S_400_BAD_REQUEST, "bad URI " + flowConfig.getTemplateUris(), e); } } FlowConfigResourceLocalHandler(FlowCatalog flowCatalog); }
FlowConfigResourceLocalHandler implements FlowConfigsResourceHandler { public static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig) { ConfigBuilder configBuilder = ConfigBuilder.create() .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, flowConfig.getId().getFlowGroup()) .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, flowConfig.getId().getFlowName()); if (flowConfig.hasSchedule()) { Schedule schedule = flowConfig.getSchedule(); configBuilder.addPrimitive(ConfigurationKeys.JOB_SCHEDULE_KEY, schedule.getCronSchedule()); configBuilder.addPrimitive(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, schedule.isRunImmediately()); } else { String flowExecutionId; if (flowConfig.getProperties().containsKey(ConfigurationKeys.FLOW_EXECUTION_ID_KEY)) { flowExecutionId = flowConfig.getProperties().get(ConfigurationKeys.FLOW_EXECUTION_ID_KEY); log.info("Using the existing flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } else { flowExecutionId = String.valueOf(System.currentTimeMillis()); log.info("Created a flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } flowConfig.getProperties().put(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); } if (flowConfig.hasExplain()) { configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXPLAIN_KEY, flowConfig.isExplain()); } Config config = configBuilder.build(); Config configWithFallback; try { configWithFallback = config.withFallback(ConfigFactory.parseString(flowConfig.getProperties().toString()).resolve()); } catch (Exception e) { configWithFallback = config.withFallback(ConfigFactory.parseMap(flowConfig.getProperties())); } try { URI templateURI = new URI(flowConfig.getTemplateUris()); return FlowSpec.builder().withConfig(configWithFallback).withTemplate(templateURI).build(); } catch (URISyntaxException e) { throw new FlowConfigLoggedException(HttpStatus.S_400_BAD_REQUEST, "bad URI " + flowConfig.getTemplateUris(), e); } } FlowConfigResourceLocalHandler(FlowCatalog flowCatalog); FlowConfig getFlowConfig(FlowId flowId); Collection<FlowConfig> getFlowConfig(FlowSpecSearchObject flowSpecSearchObject); Collection<FlowConfig> getAllFlowConfigs(); CreateResponse createFlowConfig(FlowConfig flowConfig, boolean triggerListener); CreateResponse createFlowConfig(FlowConfig flowConfig); UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener); UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig); @Override UpdateResponse partialUpdateFlowConfig(FlowId flowId, PatchRequest<FlowConfig> flowConfigPatch); UpdateResponse deleteFlowConfig(FlowId flowId, Properties header, boolean triggerListener); UpdateResponse deleteFlowConfig(FlowId flowId, Properties header); static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig); }
FlowConfigResourceLocalHandler implements FlowConfigsResourceHandler { public static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig) { ConfigBuilder configBuilder = ConfigBuilder.create() .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, flowConfig.getId().getFlowGroup()) .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, flowConfig.getId().getFlowName()); if (flowConfig.hasSchedule()) { Schedule schedule = flowConfig.getSchedule(); configBuilder.addPrimitive(ConfigurationKeys.JOB_SCHEDULE_KEY, schedule.getCronSchedule()); configBuilder.addPrimitive(ConfigurationKeys.FLOW_RUN_IMMEDIATELY, schedule.isRunImmediately()); } else { String flowExecutionId; if (flowConfig.getProperties().containsKey(ConfigurationKeys.FLOW_EXECUTION_ID_KEY)) { flowExecutionId = flowConfig.getProperties().get(ConfigurationKeys.FLOW_EXECUTION_ID_KEY); log.info("Using the existing flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } else { flowExecutionId = String.valueOf(System.currentTimeMillis()); log.info("Created a flowExecutionId {} for {},{}", flowExecutionId, flowConfig.getId().getFlowGroup(), flowConfig.getId().getFlowName()); } flowConfig.getProperties().put(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId); } if (flowConfig.hasExplain()) { configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXPLAIN_KEY, flowConfig.isExplain()); } Config config = configBuilder.build(); Config configWithFallback; try { configWithFallback = config.withFallback(ConfigFactory.parseString(flowConfig.getProperties().toString()).resolve()); } catch (Exception e) { configWithFallback = config.withFallback(ConfigFactory.parseMap(flowConfig.getProperties())); } try { URI templateURI = new URI(flowConfig.getTemplateUris()); return FlowSpec.builder().withConfig(configWithFallback).withTemplate(templateURI).build(); } catch (URISyntaxException e) { throw new FlowConfigLoggedException(HttpStatus.S_400_BAD_REQUEST, "bad URI " + flowConfig.getTemplateUris(), e); } } FlowConfigResourceLocalHandler(FlowCatalog flowCatalog); FlowConfig getFlowConfig(FlowId flowId); Collection<FlowConfig> getFlowConfig(FlowSpecSearchObject flowSpecSearchObject); Collection<FlowConfig> getAllFlowConfigs(); CreateResponse createFlowConfig(FlowConfig flowConfig, boolean triggerListener); CreateResponse createFlowConfig(FlowConfig flowConfig); UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener); UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig); @Override UpdateResponse partialUpdateFlowConfig(FlowId flowId, PatchRequest<FlowConfig> flowConfigPatch); UpdateResponse deleteFlowConfig(FlowId flowId, Properties header, boolean triggerListener); UpdateResponse deleteFlowConfig(FlowId flowId, Properties header); static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig); static final Schedule NEVER_RUN_CRON_SCHEDULE; }