104 lines
3.2 KiB
Java
104 lines
3.2 KiB
Java
package org.gcube.application.perform.service.engine.dm;
|
|
|
|
import java.sql.Connection;
|
|
import java.sql.PreparedStatement;
|
|
import java.time.Instant;
|
|
import java.util.Map;
|
|
|
|
import org.gcube.application.perform.service.engine.DataBaseManager;
|
|
import org.gcube.application.perform.service.engine.impl.Queries;
|
|
import org.gcube.application.perform.service.engine.model.DBField;
|
|
import org.gcube.application.perform.service.engine.model.DBField.ImportRoutine;
|
|
import org.gcube.application.perform.service.engine.model.DBQueryDescriptor;
|
|
import org.gcube.application.perform.service.engine.model.ISQueryDescriptor;
|
|
import org.gcube.application.perform.service.engine.model.importer.ImportRoutineDescriptor;
|
|
import org.gcube.application.perform.service.engine.model.importer.ImportStatus;
|
|
import org.gcube.data.analysis.dataminermanagercl.server.monitor.DMMonitorListener;
|
|
import org.gcube.data.analysis.dataminermanagercl.shared.data.computations.ComputationId;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
|
|
public class ImporterMonitor implements DMMonitorListener {
|
|
|
|
private static final Logger log= LoggerFactory.getLogger(ImporterMonitor.class);
|
|
|
|
public ImporterMonitor(ImportRoutineDescriptor routine, ISQueryDescriptor isQuery) {
|
|
super();
|
|
this.routine = routine;
|
|
this.isQuery = isQuery;
|
|
}
|
|
|
|
|
|
private ImportRoutineDescriptor routine;
|
|
private ISQueryDescriptor isQuery;
|
|
|
|
@Override
|
|
public void accepted() {
|
|
updateStatus(ImportStatus.ACCEPTED,routine,isQuery);
|
|
}
|
|
|
|
@Override
|
|
public void cancelled() {
|
|
updateStatus(ImportStatus.CANCELLED,routine,isQuery);
|
|
}
|
|
|
|
@Override
|
|
public void complete(double percentage) {
|
|
updateStatus(ImportStatus.COMPLETE,routine,isQuery);
|
|
loadOutputData(routine,isQuery);
|
|
}
|
|
|
|
@Override
|
|
public void failed(String message, Exception exception) {
|
|
updateStatus(ImportStatus.FAILED,routine,isQuery);
|
|
}
|
|
|
|
@Override
|
|
public void running(double percentage) {
|
|
updateStatus(ImportStatus.RUNNING,routine,isQuery);
|
|
}
|
|
|
|
|
|
|
|
private static final void updateStatus(ImportStatus status,ImportRoutineDescriptor routine,ISQueryDescriptor is) {
|
|
try{
|
|
log.debug("Updateing status {} for {} ",status,routine);
|
|
DataBaseManager db=DataBaseManager.get(is);
|
|
Connection conn=db.getConnection();
|
|
try {
|
|
conn.setAutoCommit(true);
|
|
|
|
Instant endTime=null;
|
|
switch(status) {
|
|
case CANCELLED:
|
|
case COMPLETE :
|
|
case FAILED : endTime=Instant.now();
|
|
}
|
|
|
|
DBQueryDescriptor queryValues=new DBQueryDescriptor().
|
|
add(DBField.ImportRoutine.fields.get(ImportRoutine.ID), routine.getId()).
|
|
add(DBField.ImportRoutine.fields.get(ImportRoutine.STATUS), status.toString()).
|
|
add(DBField.ImportRoutine.fields.get(ImportRoutine.END), endTime);
|
|
|
|
PreparedStatement psUpdate=Queries.UPDATE_IMPORT_STATUS.get(conn, queryValues);
|
|
psUpdate.executeUpdate();
|
|
}finally {
|
|
conn.close();
|
|
}
|
|
}catch(Throwable t) {
|
|
log.warn("Unable to update status on database");
|
|
}
|
|
}
|
|
|
|
|
|
private static final void loadOutputData(ImportRoutineDescriptor routine, ISQueryDescriptor is) {
|
|
log.debug("Loading output data for {} ",routine);
|
|
ComputationId id=DMUtils.getComputation(routine);
|
|
Map<String,String> outputFiles=DMUtils.getOutputFiles(id);
|
|
|
|
// if(outputFiles.containsKey(Local))
|
|
|
|
}
|
|
}
|