#769 Add library methods to implement ebcdic to ebcdic file processor.

Author: yruslan

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

@@ -0,0 +1,33 @@
+/*
+ * 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 za.co.absa.cobrix.cobol.parser.ast.Group
+import za.co.absa.cobrix.cobol.reader.extractors.record.RecordHandler
+
+/**
+  * A handler for processing COBOL records and mapping it to JVM data structures.
+  *
+  * This implementation uses an array to group data fields of struct fields.
+  */
+class ArrayOfAnyHandler extends RecordHandler[scala.Array[Any]] {
+  override def create(values: Array[Any], group: Group): Array[Any] = values
+
+  override def toSeq(record: Array[Any]): Seq[Any] = record.toSeq
+
+  override def foreach(record: Array[Any])(f: Any => Unit): Unit = record.foreach(f)
+}

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

@@ -0,0 +1,40 @@
+/*
+ * 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 za.co.absa.cobrix.cobol.parser.ast.Group
+import za.co.absa.cobrix.cobol.reader.extractors.record.RecordHandler
+
+/**
+  * A handler for processing COBOL records and mapping it to JVM data structures.
+  *
+  * This implementation uses a map from a string field name to value to represent struct fields from data records.
+  */
+class MapOfAnyHandler extends RecordHandler[Map[String, Any]] {
+  override def create(values: Array[Any], group: Group): Map[String, Any] = {
+    (group.children zip values).map(t => t._1.name -> (t._2 match {
+      case s: Array[Any] => s.toSeq
+      case s => s
+    })).toMap
+  }
+
+  override def toSeq(record: Map[String, Any]): Seq[Any] = {
+    record.values.toSeq
+  }
+
+  override def foreach(record: Map[String, Any])(f: Any => Unit): Unit = record.values.foreach(f)
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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 za.co.absa.cobrix.cobol.parser.Copybook
+
+/**
+ * A trait that defines a processor for raw COBOL records.
+ * It provides a method to process a single COBOL record based on the provided copybook and options.
+ */
+trait RawRecordProcessor {
+  def processRecord(copybook: Copybook,
+                    options: Map[String, String],
+                    record: Array[Byte],
+                    offset: Long): Array[Byte]
+
+}

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

@@ -0,0 +1,114 @@
+/*
+ * 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 za.co.absa.cobrix.cobol.reader.VarLenNestedReader
+import za.co.absa.cobrix.cobol.reader.extractors.raw.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
+
+import java.io.OutputStream
+import scala.collection.mutable
+import scala.reflect.ClassTag
+
+object RecordProcessorBuilder {
+  /**
+    * Creates a new instance of the RecordProcessorBuilder with the given copybook content.
+    * The instabce is used to create an COBOL data processor allowing applying changes to a mainrame file
+    * without changing the original format.
+    *
+    * @param copybookContent the COBOL copybook content as a string.
+    * @return a new RecordProcessorBuilder instance.
+    */
+  def copybookContents(copybookContent: String): RecordProcessorBuilder = {
+    new RecordProcessorBuilder(copybookContent)
+  }
+}
+
+class RecordProcessorBuilder(copybookContents: String) {
+  private val caseInsensitiveOptions = new mutable.HashMap[String, String]()
+
+  /**
+    * Adds a single option to the builder.
+    *
+    * @param key   the option key.
+    * @param value the option value.
+    * @return this builder instance for method chaining.
+    */
+  def option(key: String, value: String): RecordProcessorBuilder = {
+    caseInsensitiveOptions += (key.toLowerCase -> value)
+    this
+  }
+
+  /**
+    * Adds multiple options to the builder.
+    *
+    * @param options a map of option key-value pairs.
+    * @return this builder instance for method chaining.
+    */
+  def options(options: Map[String, String]): RecordProcessorBuilder = {
+    caseInsensitiveOptions ++= options.map(kv => (kv._1.toLowerCase(), kv._2))
+    this
+  }
+
+  /**
+    * Processes the input stream of COBOL records and writes the output to the specified output stream.
+    *
+    * @param inputStream        the input stream containing raw COBOL records.
+    * @param outputStream       the output stream where processed records will be written.
+    * @param rawRecordProcessor the processor that processes each raw record.
+    */
+  def process(inputStream: SimpleStream,
+              outputStream: OutputStream)
+             (rawRecordProcessor: RawRecordProcessor): Unit = {
+    val readerParameters = getReaderParameters
+    val cobolSchema = getCobolSchema(readerParameters)
+    val recordExtractor = getRecordExtractor(readerParameters, inputStream)
+
+    val dataStream = inputStream.copyStream()
+    try {
+      StreamProcessor.processStream(cobolSchema.copybook,
+        caseInsensitiveOptions.toMap,
+        dataStream,
+        recordExtractor,
+        rawRecordProcessor,
+        outputStream)
+    } finally {
+      dataStream.close()
+    }
+  }
+
+  private def getCobolSchema(readerParameters: ReaderParameters): CobolSchema = {
+    CobolSchema.fromReaderParameters(Seq(copybookContents), readerParameters)
+  }
+
+  private 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 = {
+    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
+  }
+}

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

@@ -0,0 +1,69 @@
+/*
+ * 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 za.co.absa.cobrix.cobol.parser.Copybook
+import za.co.absa.cobrix.cobol.reader.extractors.raw.RawRecordExtractor
+import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
+
+import java.io.OutputStream
+
+object StreamProcessor {
+  /**
+    * Processes a stream of COBOL raw records and writes it back in the same format as the input data.
+    *
+    * @param copybook the COBOL copybook that describes the schema of the records.
+    * @param options arbitrary options used for splitting input data into records. Same as options to 'spark-cobol'. Can contain custom options as well.
+    * @param inputStream the input stream containing the raw COBOL records.
+    * @param recordExtractor the extractor that extracts raw records from the input stream.
+    * @param recordProcessor the per-record processing logic implementation.
+    * @param outputStream the output stream where the processed records will be written.
+    */
+  def processStream(copybook: Copybook,
+                    options: Map[String, String],
+                    inputStream: SimpleStream,
+                    recordExtractor: RawRecordExtractor,
+                    recordProcessor: RawRecordProcessor,
+                    outputStream: OutputStream): Unit = {
+    var i = 0
+    while (recordExtractor.hasNext) {
+      i += 1
+      val record = recordExtractor.next()
+      val recordSize = record.length
+
+      val updatedRecord = if (recordExtractor.hasNext) {
+        recordProcessor.processRecord(copybook, options, record, recordExtractor.offset)
+      } else {
+        record
+      }
+
+      val headerSize = recordExtractor.offset - recordSize - inputStream.offset
+      if (headerSize > 0) {
+        val header = inputStream.next(headerSize.toInt)
+        outputStream.write(header)
+      }
+      inputStream.next(recordSize)
+      outputStream.write(updatedRecord)
+    }
+
+    val footerSize = inputStream.size - inputStream.offset
+    if (footerSize > 0) {
+      val footer = inputStream.next(footerSize.toInt)
+      outputStream.write(footer)
+    }
+  }
+}

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/stream/FSStream.scala

@@ -16,7 +16,7 @@
 
 package za.co.absa.cobrix.cobol.reader.stream
 
-import java.io.{BufferedInputStream, File, FileInputStream, IOException}
+import java.io.{BufferedInputStream, File, FileInputStream, FileNotFoundException, IOException}
 
 class FSStream (fileName: String) extends SimpleStream {
   val bytesStream = new BufferedInputStream(new FileInputStream(fileName))
@@ -33,7 +33,6 @@ class FSStream (fileName: String) extends SimpleStream {
 
   override def inputFileName: String = fileName
 
-  @throws(classOf[IllegalArgumentException])
   @throws(classOf[IOException])
   override def next(numberOfBytes: Int): Array[Byte] = {
     if (numberOfBytes <= 0) throw new IllegalArgumentException("Value of numberOfBytes should be greater than zero.")
@@ -55,4 +54,9 @@ class FSStream (fileName: String) extends SimpleStream {
       isClosed = true
     }
   }
+
+  @throws(classOf[FileNotFoundException])
+  override def copyStream(): SimpleStream = {
+    new FSStream(fileName)
+  }
 }

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/stream/SimpleStream.scala

@@ -29,6 +29,9 @@ trait SimpleStream {
 
   def isEndOfStream: Boolean = offset >= size
 
+  @throws(classOf[Exception])
+  def copyStream(): SimpleStream
+
   @throws(classOf[Exception]) def next(numberOfBytes: Int): Array[Byte]
 
   @throws(classOf[Exception]) def close(): Unit

cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/mock/ByteStreamMock.scala

@@ -16,7 +16,9 @@
 
 package za.co.absa.cobrix.cobol.mock
 
-import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
+import za.co.absa.cobrix.cobol.reader.stream.{FSStream, SimpleStream}
+
+import java.io.FileNotFoundException
 
 class ByteStreamMock(bytes: Array[Byte]) extends SimpleStream{
 
@@ -48,4 +50,8 @@ class ByteStreamMock(bytes: Array[Byte]) extends SimpleStream{
   }
 
   override def close(): Unit = position = sz
+
+  override def copyStream(): SimpleStream = {
+    new ByteStreamMock(bytes)
+  }
 }

cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/memorystream/TestByteStream.scala

@@ -16,6 +16,7 @@
 
 package za.co.absa.cobrix.cobol.reader.memorystream
 
+import za.co.absa.cobrix.cobol.mock.ByteStreamMock
 import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
 
 class TestByteStream(bytes: Array[Byte]) extends SimpleStream{
@@ -48,4 +49,8 @@ class TestByteStream(bytes: Array[Byte]) extends SimpleStream{
   }
 
   override def close(): Unit = position = sz
+
+  override def copyStream(): SimpleStream = {
+    new TestByteStream(bytes)
+  }
 }

cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/memorystream/TestStringStream.scala

@@ -48,4 +48,8 @@ class TestStringStream(str: String) extends SimpleStream{
   }
 
   override def close(): Unit = position = sz
+
+  override def copyStream(): SimpleStream = {
+    new TestStringStream(str)
+  }
 }