Add interface for pipelineExecutionFailed to ProgressListener.java

Author: Evgenii Grigorev

s-pipes-core/src/main/java/cz/cvut/spipes/engine/ExecutionEngineImpl.java

@@ -1,6 +1,5 @@
 package cz.cvut.spipes.engine;
 
-import cz.cvut.spipes.logging.AdvancedLoggingProgressListener;
 import cz.cvut.spipes.modules.Module;
 import org.apache.jena.rdf.model.Model;
 import org.apache.jena.rdf.model.ModelFactory;
@@ -37,9 +36,7 @@ public ExecutionContext executePipeline(final Module module, final ExecutionCont
         } catch (Exception e) {
             log.error("Pipeline execution failed", e);
             fire((l) -> {
-                if (l instanceof AdvancedLoggingProgressListener) {
-                    ((AdvancedLoggingProgressListener) l).pipelineExecutionFailed(pipelineExecutionId);
-                }
+                l.pipelineExecutionFailed(pipelineExecutionId);
                 return null;
             });
             throw e;

s-pipes-core/src/main/java/cz/cvut/spipes/engine/LoggingProgressListener.java

@@ -17,6 +17,11 @@ public void pipelineExecutionFinished(long pipelineExecutionId) {
         LOG.debug("pipelineExecutionFinished - pipelineExecutionId: {}", pipelineExecutionId);
     }
 
+    @Override
+    public void pipelineExecutionFailed(long pipelineExecutionId) {
+        LOG.debug("pipelineExecutionFailed - pipelineExecutionId: {}", pipelineExecutionId);
+    }
+
     @Override
     public void moduleExecutionStarted(long pipelineExecutionId, String moduleExecutionId, Module outputModule, ExecutionContext inputContext, String predecessorModuleExecutionId) {
         LOG.debug("moduleExecutionStarted - pipelineExecutionId: {}, moduleExecutionId: {}, inputContext: {}, predecessorModuleExecutionId: {}", pipelineExecutionId,

s-pipes-core/src/main/java/cz/cvut/spipes/engine/ProgressListener.java

@@ -25,6 +25,8 @@ public interface ProgressListener {
      */
     void pipelineExecutionFinished(long pipelineExecutionId);
 
+    void pipelineExecutionFailed(long pipelineExecutionId);
+
     /**
      * Triggers when execution of a module within a pipeline starts.
      *

s-pipes-core/src/main/java/cz/cvut/spipes/logging/SemanticLoggingProgressListener.java

@@ -108,6 +108,34 @@ public SemanticLoggingProgressListener(Resource configResource) {
         }
     }
 
+    @Override
+    public void pipelineExecutionFailed(final long pipelineExecutionId) {
+        final EntityManager em = entityManagerMap.get(getPipelineExecutionIri(pipelineExecutionId));
+        synchronized (em) {
+            if (em.isOpen()) {
+                final TurtleWriterFactory factory = new TurtleWriterFactory();
+                try (FileOutputStream fos = new FileOutputStream(
+                        Files.createFile(getDir(pipelineExecutionId).resolve("log.ttl")).toFile())) {
+                    final RDFWriter writer = factory.getWriter(fos);
+                    writer.startRDF();
+                    RepositoryConnection con = em.unwrap(SailRepository.class).getConnection();
+                    final ValueFactory f = con.getValueFactory();
+                    final RepositoryResult<Statement> res = con
+                            .getStatements(null, null, null, true, f.createIRI(getPipelineExecutionIri(pipelineExecutionId)));
+                    while (res.hasNext()) {
+                        writer.handleStatement(res.next());
+                    }
+                    writer.endRDF();
+                } catch (IOException e) {
+                    log.error("Error during failed pipeline execution logging.", e);
+                }
+                entityManagerMap.remove(em);
+                em.close();
+                logDir.remove(pipelineExecutionId);
+            }
+        }
+    }
+
     @Override public void moduleExecutionStarted(final long pipelineExecutionId, final String moduleExecutionId,
                                                  final Module outputModule,
                                                  final ExecutionContext inputContext,