yystopf
/
piflow
forked from opensci/piflow
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

monitor project

This commit is contained in:
judy0131 2019-05-15 12:27:13 +08:00
parent 0fff9ec29b
commit c8c01811b4
2 changed files with 34 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
piflow-core/src/main/scala/cn/piflow/project.scala
View File

@ -131,7 +131,7 @@ class ProjectExecutionImpl(project: Project, runnerContext: Context, runner: Run
completedProjectEntry(name)
}
private def startProcess(name: String, flow: Flow): Unit = {
private def startProcess(name: String, flow: Flow, projectId: String = ""): Unit = {
println(flow.getFlowJson())
@ -177,14 +177,30 @@ class ProjectExecutionImpl(project: Project, runnerContext: Context, runner: Run
Thread.sleep(100)
}
appId = handle.getAppId
//H2Util.updateFlowGroupId(appId, groupId)
//wait flow process started
while(H2Util.getFlowProcessId(appId).equals("")){
Thread.sleep(1000)
}
if(projectId != ""){
H2Util.updateFlowProjectId(appId, projectId)
}
startedProcesses(name) = handle;
startedProcessesAppID(name) = appId
}
private def startFlowGroup(name: String, flowGroup: FlowGroup): Unit = {
private def startFlowGroup(name: String, flowGroup: FlowGroup, projectId: String): Unit = {
val flowGroupExecution = runner.start(flowGroup);
startedFlowGroup(name) = flowGroupExecution;
val flowGroupId = flowGroupExecution.groupId()
while(H2Util.getFlowGroupState(flowGroupId).equals("")){
Thread.sleep(1000)
}
if(projectId != ""){
H2Util.updateFlowGroupProject(flowGroupId,projectId)
}
}
val pollingThread = new Thread(new Runnable() {
@ -211,10 +227,10 @@ class ProjectExecutionImpl(project: Project, runnerContext: Context, runner: Run
}
startedProcesses.synchronized {
todosFlow.foreach(en => startProcess(en._1, en._2.asInstanceOf[Flow]));
todosFlow.foreach(en => startProcess(en._1, en._2.asInstanceOf[Flow],id));
}
startedFlowGroup.synchronized{
todosFlowGroup.foreach(en => startFlowGroup(en._1, en._2.asInstanceOf[FlowGroup]))
todosFlowGroup.foreach(en => startFlowGroup(en._1, en._2.asInstanceOf[FlowGroup],id))
}
Thread.sleep(POLLING_INTERVAL);

13
piflow-core/src/main/scala/cn/piflow/util/H2Util.scala
View File

@ -376,6 +376,19 @@ object H2Util {
statement.close()
}
def getFlowGroupState(flowGroupId:String) : String = {
val statement = getConnectionInstance().createStatement()
statement.setQueryTimeout(QUERY_TIME)
var flowGroupState = ""
val flowGroupRS : ResultSet = statement.executeQuery("select state from flowGroup where id='" + flowGroupId +"'")
if (flowGroupRS.next()){
flowGroupState = flowGroupRS.getString("state")
}
return flowGroupState
}
//project related api
def addProject(projectId:String,name:String)={
val startTime = new Date().toString