Integrated Pipeline: ParsingTaskGenerators now forward scanning errors to the...

Integrated Pipeline: ParsingTaskGenerators now forward scanning errors to the TreeParser and its EventListener

integrated-pipeline/src/main/scala/eu/nomad_lab/integrated_pipeline/EventLogger.scala

@@ -24,6 +24,10 @@ trait EventLogger extends EventListener with StrictLogging {
         s"[TreeParser ${reporter.name}] found candidate in file tree " +
           s"'${x.treeTask.treeBasePath}' at '${x.relativePath}' (parser '${x.parser}')"
       )
+      case x: TreeParserEventScanError => logger.warn(
+        s"[TreeParser ${reporter.name}] scanning failure in file tree " +
+          s"'${x.treeTask.treeBasePath}' at '${x.relativePath}' (${x.error.toString})"
+      )
     }
     super.processEvent(reporter, message)
   }

integrated-pipeline/src/main/scala/eu/nomad_lab/integrated_pipeline/TreeParser.scala

@@ -2,6 +2,8 @@ package eu.nomad_lab.integrated_pipeline
 
 import eu.nomad_lab.integrated_pipeline.messages._
 
+import scala.annotation.tailrec
+
 trait TreeParser extends MessageProcessor[FileTreeScanTask, FileParsingTaskSignal] {
 
   protected[this] def createProcessor(task: FileTreeScanTask): ParsingTaskGenerator
@@ -29,12 +31,14 @@ trait TreeParser extends MessageProcessor[FileTreeScanTask, FileParsingTaskSigna
 
   final override def hasSignalToEmit(): Boolean = !readyForInbound
 
-  final override def getNextSignalToEmit(): FileParsingTaskSignal = {
+  @tailrec final override def getNextSignalToEmit(): FileParsingTaskSignal = {
     if (generator.exists(_.hasNext)) {
-      taskCount += 1
       generator.get.next() match {
-        case Left(_) => ???
+        case Left(scanError) =>
+          eventListener.processEvent(myId, scanError)
+          getNextSignalToEmit()
         case Right(task) =>
+          taskCount += 1
           eventListener.processEvent(
             myId,
             TreeParserEventCandidate(request.get, task.relativePath, task.parserName)

integrated-pipeline/src/main/scala/eu/nomad_lab/integrated_pipeline/io_integrations/DirectoryTreeParsingTaskGenerator.scala

@@ -10,6 +10,7 @@ import eu.nomad_lab.parsers.{ CandidateParser, ParserCollection }
 
 import scala.annotation.tailrec
 import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
 
 class DirectoryTreeParsingTaskGenerator(request: FileTreeScanTask, parserCollection: ParserCollection)
     extends ParsingTaskGenerator {
@@ -27,7 +28,8 @@ class DirectoryTreeParsingTaskGenerator(request: FileTreeScanTask, parserCollect
         val in = new FileInputStream(file.toFile)
         try {
           scanInputStream(parserCollection, in, internalFilePath.toString)
-          //TODO: handle errors
+        } catch {
+          case NonFatal(e) => return Some(Left(TreeParserEventScanError(request, internalFilePath, e)))
         } finally {
           in.close()
         }

integrated-pipeline/src/main/scala/eu/nomad_lab/integrated_pipeline/io_integrations/ZipTreeParsingTaskGenerator.scala

@@ -9,6 +9,7 @@ import eu.nomad_lab.parsers.{ CandidateParser, ParserCollection }
 import org.apache.commons.compress.archivers.zip.{ ZipArchiveEntry, ZipFile }
 
 import scala.annotation.tailrec
+import scala.util.control.NonFatal
 
 class ZipTreeParsingTaskGenerator(request: FileTreeScanTask, parserCollection: ParserCollection)
     extends ParsingTaskGenerator {
@@ -30,7 +31,8 @@ class ZipTreeParsingTaskGenerator(request: FileTreeScanTask, parserCollection: P
         val zIn = zipFile.getInputStream(zipEntry)
         try {
           scanInputStream(parserCollection, zIn, internalFilePath.toString)
-          //TODO: handle errors
+        } catch {
+          case NonFatal(e) => return Some(Left(TreeParserEventScanError(request, internalFilePath, e)))
         } finally {
           zIn.close()
         }

integrated-pipeline/src/test/scala/eu/nomad_lab/integrated_pipeline_tests/EventMatchers.scala

@@ -35,6 +35,24 @@ object TreeParserEventMatchers {
       test = (x: TreeParserEventCandidate) => x.parser
     )
 
+  def errorMessage(first: String, other: String*): HavePropertyMatcher[TreeParserEventScanError, String] =
+    MatcherHelpers.subStringPropertyMatcher(
+      propertyName = "error message content",
+      expected = first +: other,
+      test = (x: TreeParserEventScanError) => x.error.getMessage,
+      display = (x: TreeParserEventScanError) => x.error.toString
+    )
+
+  def errorRelativePath(expectedValue: Path): HavePropertyMatcher[TreeParserEventScanError, Path] =
+    MatcherHelpers.propertyMatcher(
+      propertyName = "relative file path",
+      expected = expectedValue,
+      test = (x: TreeParserEventScanError) => x.relativePath
+    )
+
+  def errorRelativePath(expectedValue: String): HavePropertyMatcher[TreeParserEventScanError, Path] =
+    errorRelativePath(Paths.get(expectedValue))
+
   def numCandidates(expectedValue: Long): HavePropertyMatcher[TreeParserEventEnd, Long] =
     MatcherHelpers.propertyMatcher(
       propertyName = "number of candidate calculations",

integrated-pipeline/src/test/scala/eu/nomad_lab/integrated_pipeline_tests/TreeParserSpec.scala

@@ -49,6 +49,16 @@ class TreeParserSpec extends WordSpec with TestDataBuilders with MockitoSugar
     new Fixture(data)
   }
 
+  def taskReceivedWillCreateScanError(): Fixture = {
+    val data: Iterator[Either[TreeParserEventScanError, FileParsingTask]] = Iterator.single(
+      Left(aTreeParserEventScanError().withTreeTask(sampleTreeTask).withRelativePath("nullPointer").
+        withError(new UnsupportedOperationException("failing test case")) build ())
+    )
+    val f = new Fixture(data)
+    f.treeParser.processSignal(sampleTreeTask)
+    f
+  }
+
   def taskReceivedNoSignalsFetched(): Fixture = {
     val f = noTaskReceived()
     f.treeParser.processSignal(sampleTreeTask)
@@ -175,6 +185,21 @@ class TreeParserSpec extends WordSpec with TestDataBuilders with MockitoSugar
         Seq(aFileTreeScanTask())
       )
     }
+
+    "receiving a file scanning error from the task generator" should {
+      def createFixture = () => taskReceivedWillCreateScanError()
+
+      "send a notification to the event listener" in {
+        import TreeParserEventMatchers._
+        val f = createFixture()
+        f.treeParser.getNextSignalToEmit()
+        val captor = argsCaptor()
+        verify(f.events, Mockito.atLeast(1)).processEvent(any(), captor.capture())
+        val events = captor.getAllValues.asScala.collect { case x: TreeParserEventScanError => x }
+        events should have size 1
+        events.head should have(errorMessage("failing", "test case"), errorRelativePath("nullPointer"))
+      }
+    }
   }
 
 }