public void schedule ( final Project project , final String submitUser ) throws ProjectManagerException , IOException , SchedulerException { for ( final Flow flow : project . getFlows ( ) ) { //todo chengren311: we should validate embedded flow shouldn't have flow trigger defined. if ( flow . isEmbeddedFlow ( ) ) { // skip scheduling embedded flow since embedded flow are not allowed to have flow trigger continue ; } final String flowFileName = flow . getId ( ) + ".flow" ; final int latestFlowVersion = this . projectLoader . getLatestFlowVersion ( flow . getProjectId ( ) , flow . getVersion ( ) , flowFileName ) ; if ( latestFlowVersion > 0 ) { final File tempDir = Files . createTempDir ( ) ; final File flowFile ; try { flowFile = this . projectLoader . getUploadedFlowFile ( project . getId ( ) , project . getVersion ( ) , flowFileName , latestFlowVersion , tempDir ) ; final FlowTrigger flowTrigger = FlowLoaderUtils . getFlowTriggerFromYamlFile ( flowFile ) ; if ( flowTrigger != null ) { final Map < String , Object > contextMap = ImmutableMap . of ( FlowTriggerQuartzJob . SUBMIT_USER , submitUser , FlowTriggerQuartzJob . FLOW_TRIGGER , flowTrigger , FlowTriggerQuartzJob . FLOW_ID , flow . getId ( ) , FlowTriggerQuartzJob . FLOW_VERSION , latestFlowVersion , FlowTriggerQuartzJob . PROJECT_ID , project . getId ( ) ) ; final boolean scheduleSuccess = this . scheduler . scheduleJobIfAbsent ( flowTrigger . getSchedule ( ) . getCronExpression ( ) , new QuartzJobDescription ( FlowTriggerQuartzJob . class , FlowTriggerQuartzJob . JOB_NAME , generateGroupName ( flow ) , contextMap ) ) ; if ( scheduleSuccess ) { logger . info ( "Successfully registered flow {}.{} to scheduler" , project . getName ( ) , flow . getId ( ) ) ; } else { logger . info ( "Fail to register a duplicate flow {}.{} to scheduler" , project . getName ( ) , flow . getId ( ) ) ; } } } catch ( final SchedulerException | IOException ex ) { logger . error ( "Error in registering flow {}.{}" , project . getName ( ) , flow . getId ( ) , ex ) ; throw ex ; } finally { FlowLoaderUtils . cleanUpDir ( tempDir ) ; } } } }
Schedule flows containing flow triggers for this project .