#769 Fix the processor not processing the last record, add unit tests.

Author: yruslan

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/processor/RecordProcessorBuilder.scala

@@ -16,8 +16,9 @@
 
 package za.co.absa.cobrix.cobol.processor
 
+import za.co.absa.cobrix.cobol.parser.recordformats.RecordFormat.FixedLength
 import za.co.absa.cobrix.cobol.reader.VarLenNestedReader
-import za.co.absa.cobrix.cobol.reader.extractors.raw.RawRecordExtractor
+import za.co.absa.cobrix.cobol.reader.extractors.raw.{FixedRecordLengthRawRecordExtractor, RawRecordContext, RawRecordExtractor}
 import za.co.absa.cobrix.cobol.reader.parameters.{CobolParametersParser, Parameters, ReaderParameters}
 import za.co.absa.cobrix.cobol.reader.schema.CobolSchema
 import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
@@ -93,22 +94,35 @@ class RecordProcessorBuilder(copybookContents: String) {
     }
   }
 
-  private def getCobolSchema(readerParameters: ReaderParameters): CobolSchema = {
+  private[processor] def getCobolSchema(readerParameters: ReaderParameters): CobolSchema = {
     CobolSchema.fromReaderParameters(Seq(copybookContents), readerParameters)
   }
 
-  private def getReaderParameters: ReaderParameters = {
+  private[processor] def getReaderParameters: ReaderParameters = {
     val cobolParameters = CobolParametersParser.parse(new Parameters(caseInsensitiveOptions.toMap))
 
     CobolParametersParser.getReaderProperties(cobolParameters, None)
   }
 
-  private def getRecordExtractor[T: ClassTag](readerParameters: ReaderParameters, inputStream: SimpleStream): RawRecordExtractor = {
+  private[processor] def getRecordExtractor[T: ClassTag](readerParameters: ReaderParameters, inputStream: SimpleStream): RawRecordExtractor = {
     val dataStream = inputStream.copyStream()
     val headerStream = inputStream.copyStream()
 
     val reader = new VarLenNestedReader[Array[Any]](Seq(copybookContents), readerParameters, new ArrayOfAnyHandler)
 
-    reader.recordExtractor(0, dataStream, headerStream).get
+    reader.recordExtractor(0, dataStream, headerStream) match {
+      case Some(extractor) => extractor
+      case None if readerParameters.recordFormat == FixedLength =>
+        val dataStream = inputStream.copyStream()
+        val headerStream = inputStream.copyStream()
+        val ctx = RawRecordContext(0, dataStream, headerStream, getCobolSchema(readerParameters).copybook, null, null, "")
+        new FixedRecordLengthRawRecordExtractor(ctx, readerParameters.recordLength)
+      case None =>
+        throw new IllegalArgumentException(s"Cannot create a record extractor for the given reader parameters. " +
+            "Please check the copybook and the reader parameters."
+        )
+    }
   }
+
+  private[processor] def getOptions: Map[String, String] = caseInsensitiveOptions.toMap
 }

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/processor/StreamProcessor.scala

@@ -45,11 +45,7 @@ object StreamProcessor {
       val record = recordExtractor.next()
       val recordSize = record.length
 
-      val updatedRecord = if (recordExtractor.hasNext) {
-        recordProcessor.processRecord(copybook, options, record, recordExtractor.offset)
-      } else {
-        record
-      }
+      val updatedRecord = recordProcessor.processRecord(copybook, options, record, recordExtractor.offset)
 
       val headerSize = recordExtractor.offset - recordSize - inputStream.offset
       if (headerSize > 0) {

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedRecordLengthRawRecordExtractor.scala

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.cobrix.cobol.reader.extractors.raw
+
+class FixedRecordLengthRawRecordExtractor(ctx: RawRecordContext, fixedRecordLength: Option[Int]) extends Serializable with RawRecordExtractor {
+  private var byteOffset: Long = ctx.inputStream.offset
+  private val recordSize = fixedRecordLength.getOrElse(ctx.copybook.getRecordSize)
+  private var currentRecordOpt: Option[Array[Byte]] = None
+
+  ctx.headerStream.close()
+
+  override def offset: Long = byteOffset
+
+  override def hasNext: Boolean = {
+    if (currentRecordOpt.isEmpty) {
+      readNextRecord()
+    }
+    currentRecordOpt.nonEmpty
+  }
+
+  private def readNextRecord(): Unit = {
+    if (!ctx.inputStream.isEndOfStream) {
+      val nextRecord = ctx.inputStream.next(recordSize)
+
+      if (nextRecord.length > 0) {
+        currentRecordOpt = Some(nextRecord)
+      }
+    }
+  }
+
+
+  @throws[NoSuchElementException]
+  override def next(): Array[Byte] = {
+    if (!hasNext) {
+      throw new NoSuchElementException
+    }
+    val record = currentRecordOpt.get
+    byteOffset += record.length
+    currentRecordOpt = None
+    record
+  }
+}

cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/RecordProcessorBuilderSuite.scala

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.cobrix.cobol.processor
+
+import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.cobrix.cobol.mock.ByteStreamMock
+import za.co.absa.cobrix.cobol.parser.Copybook
+import za.co.absa.cobrix.cobol.parser.recordformats.RecordFormat
+import za.co.absa.cobrix.cobol.reader.extractors.raw.{FixedRecordLengthRawRecordExtractor, TextFullRecordExtractor}
+import za.co.absa.cobrix.cobol.reader.parameters.ReaderParameters
+
+import java.io.ByteArrayOutputStream
+
+class RecordProcessorBuilderSuite extends AnyWordSpec {
+  private val copybook =
+    """      01 RECORD.
+      |         05  T     PIC X.
+      |""".stripMargin
+  "process" should {
+    "process an input data stream into an output stream" in {
+      val is = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
+      val os = new ByteArrayOutputStream(10)
+      val builder = RecordProcessorBuilder.copybookContents(copybook)
+
+      val processor = new RawRecordProcessor {
+        override def processRecord(copybook: Copybook, options: Map[String, String], record: Array[Byte], offset: Long): Array[Byte] = {
+          record.map(v => (v - 1).toByte)
+        }
+      }
+
+      builder.process(is, os)(processor)
+
+      val outputArray = os.toByteArray
+
+      assert(outputArray.head == -16)
+      assert(outputArray(1) == -15)
+      assert(outputArray(2) == -14)
+      assert(outputArray(3) == -13)
+    }
+  }
+
+  "getCobolSchema" should {
+    "return the schema of the copybook provided" in {
+      val builder = RecordProcessorBuilder.copybookContents(copybook)
+
+      val cobolSchema = builder.getCobolSchema(ReaderParameters())
+
+      assert(cobolSchema.copybook.ast.children.length == 1)
+    }
+  }
+
+  "getReaderParameters" should {
+    "return a reader according to passed options" in {
+      val builder = RecordProcessorBuilder.copybookContents(copybook)
+        .option("record_format", "D")
+
+      assert(builder.getReaderParameters.recordFormat == RecordFormat.AsciiText)
+      assert(builder.getOptions.contains("record_format"))
+    }
+  }
+
+  "getRecordExtractor" should {
+    "work for an fixed-record-length files" in {
+      val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
+      val builder = RecordProcessorBuilder.copybookContents(copybook)
+
+      val ext = builder.getRecordExtractor(ReaderParameters(recordLength = Some(2)), stream)
+
+      assert(ext.isInstanceOf[FixedRecordLengthRawRecordExtractor])
+
+      assert(ext.hasNext)
+      assert(ext.next().sameElements(Array(0xF1, 0xF2).map(_.toByte)))
+      assert(ext.next().sameElements(Array(0xF3, 0xF4).map(_.toByte)))
+      assert(!ext.hasNext)
+    }
+
+    "work for an variable-record-length files" in {
+      val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
+      val builder = RecordProcessorBuilder.copybookContents(copybook)
+
+      val ext = builder.getRecordExtractor(ReaderParameters(
+        recordFormat = RecordFormat.VariableLength,
+        isText = true
+      ), stream)
+
+      assert(ext.isInstanceOf[TextFullRecordExtractor])
+    }
+
+    "throw an exception on a non-supported record format for processing" in {
+      val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
+      val builder = RecordProcessorBuilder.copybookContents(copybook)
+
+      val ex = intercept[IllegalArgumentException] {
+        builder.getRecordExtractor(ReaderParameters(
+          recordFormat = RecordFormat.VariableLength,
+          isRecordSequence = true
+        ), stream)
+      }
+
+      assert(ex.getMessage.contains("Cannot create a record extractor for the given reader parameters."))
+    }
+  }
+
+}