Skip to content

Commit d547863

Browse files
yruslanyruslan
committed
#780 Fix PR suggestions. Fix RDW+BDW encoders from Reader parameters converters.
1 parent bc8c400 commit d547863

8 files changed

Lines changed: 34 additions & 28 deletions

File tree

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,8 @@ object CobolProcessor {
8282
* @return this builder instance for method chaining.
8383
*/
8484
def option(key: String, value: String): CobolProcessorBuilder = {
85-
caseInsensitiveOptions += (key.toLowerCase -> value)
85+
require(key.trim.nonEmpty, "Option key must not be empty or whitespace-only")
86+
caseInsensitiveOptions += (key.trim.toLowerCase -> value)
8687
this
8788
}
8889

@@ -117,7 +118,9 @@ object CobolProcessor {
117118
case Some(extractor) => extractor
118119
case None if readerParameters.recordFormat == FixedLength =>
119120
val dataStream = inputStream.copyStream()
120-
val ctx = RawRecordContext.builder(dataStream, getCobolSchema(readerParameters).copybook).build()
121+
val ctx = RawRecordContext.builder(dataStream, getCobolSchema(readerParameters).copybook)
122+
.withReaderParams(readerParameters)
123+
.build()
121124
new FixedRecordLengthRawRecordExtractor(ctx, readerParameters.recordLength)
122125
case None =>
123126
throw new IllegalArgumentException(s"Cannot create a record extractor for the given reader parameters. " +

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

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -27,22 +27,21 @@ object StreamProcessor {
2727
/**
2828
* Processes a stream of COBOL raw records and writes it back in the same format as the input data.
2929
*
30-
* @param copybook the COBOL copybook that describes the schema of the records.
31-
* @param options arbitrary options used for splitting input data into records. Same as options to 'spark-cobol'. Can contain custom options as well.
32-
* @param inputStream the input stream containing the raw COBOL records.
30+
* @param copybook the COBOL copybook that describes the schema of the records.
31+
* @param options arbitrary options used for splitting input data into records (same as 'spark-cobol' options).
32+
* Keys are lower-cased for case-insensitive handling. Can contain custom options as well.
33+
* @param inputStream the input stream containing the raw COBOL records.
3334
* @param recordExtractor the extractor that extracts raw records from the input stream.
3435
* @param recordProcessor the per-record processing logic implementation.
35-
* @param outputStream the output stream where the processed records will be written.
36+
* @param outputStream the output stream where the processed records will be written.
3637
*/
3738
def processStream(copybook: Copybook,
3839
options: Map[String, String],
3940
inputStream: SimpleStream,
4041
recordExtractor: RawRecordExtractor,
4142
recordProcessor: RawRecordProcessor,
4243
outputStream: OutputStream): Unit = {
43-
var i = 0
4444
while (recordExtractor.hasNext) {
45-
i += 1
4645
val record = recordExtractor.next()
4746
val recordSize = record.length
4847

‎cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/VarLenNestedReader.scala‎

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package za.co.absa.cobrix.cobol.reader
1818

1919
import za.co.absa.cobrix.cobol.internal.Logging
20-
import za.co.absa.cobrix.cobol.parser.Copybook
2120
import za.co.absa.cobrix.cobol.parser.common.Constants
2221
import za.co.absa.cobrix.cobol.parser.headerparsers.{RecordHeaderParser, RecordHeaderParserFactory}
2322
import za.co.absa.cobrix.cobol.parser.recordformats.RecordFormat.{FixedBlock, FixedLength, VariableBlock, VariableLength}
@@ -27,7 +26,6 @@ import za.co.absa.cobrix.cobol.reader.index.IndexGenerator
2726
import za.co.absa.cobrix.cobol.reader.index.entry.SparseIndexEntry
2827
import za.co.absa.cobrix.cobol.reader.iterator.{VarLenHierarchicalIterator, VarLenNestedIterator}
2928
import za.co.absa.cobrix.cobol.reader.parameters.ReaderParameters
30-
import za.co.absa.cobrix.cobol.reader.recordheader.{RecordHeaderDecoderBdw, RecordHeaderDecoderRdw, RecordHeaderParameters}
3129
import za.co.absa.cobrix.cobol.reader.schema.CobolSchema
3230
import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
3331
import za.co.absa.cobrix.cobol.reader.validator.ReaderParametersValidator
@@ -56,19 +54,9 @@ class VarLenNestedReader[T: ClassTag](copybookContents: Seq[String],
5654
def recordExtractor(startingRecordNumber: Long,
5755
dataStream: SimpleStream,
5856
headerStream: SimpleStream): Option[RawRecordExtractor] = {
59-
val rdwParams = RecordHeaderParameters(readerProperties.isRdwBigEndian, readerProperties.rdwAdjustment)
60-
61-
val rdwDecoder = new RecordHeaderDecoderRdw(rdwParams)
62-
6357
val bdwOpt = readerProperties.bdw
64-
val bdwParamsOpt = bdwOpt.map(bdw => RecordHeaderParameters(bdw.isBigEndian, bdw.adjustment))
65-
val bdwDecoderOpt = bdwParamsOpt.map(bdwParams => new RecordHeaderDecoderBdw(bdwParams))
66-
6758
val reParams = RawRecordContext.builder(startingRecordNumber, dataStream, headerStream, cobolSchema.copybook)
68-
.withRdwDecoder(rdwDecoder)
69-
.withBdwDecoder(bdwDecoderOpt.getOrElse(rdwDecoder))
70-
.withAdditionalInfo(readerProperties.reAdditionalInfo)
71-
.withOptions(readerProperties.options)
59+
.withReaderParams(readerProperties)
7260
.build()
7361

7462
readerProperties.recordExtractor match {

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

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package za.co.absa.cobrix.cobol.reader.extractors.raw
1818

1919
import za.co.absa.cobrix.cobol.parser.Copybook
20+
import za.co.absa.cobrix.cobol.reader.parameters.ReaderParameters
2021
import za.co.absa.cobrix.cobol.reader.recordheader.{RecordHeaderDecoder, RecordHeaderDecoderBdw, RecordHeaderDecoderRdw, RecordHeaderParameters}
2122
import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
2223

@@ -56,7 +57,23 @@ object RawRecordContext {
5657
Map.empty[String, String]
5758
)
5859

60+
def withReaderParams(readerParameters: ReaderParameters): RawRecordContextBuilder = {
61+
val rdwParams = RecordHeaderParameters(readerParameters.isRdwBigEndian, readerParameters.rdwAdjustment)
62+
63+
val rdwDecoder = new RecordHeaderDecoderRdw(rdwParams)
64+
65+
val bdwOpt = readerParameters.bdw
66+
val bdwParamsOpt = bdwOpt.map(bdw => RecordHeaderParameters(bdw.isBigEndian, bdw.adjustment))
67+
val bdwDecoderOpt = bdwParamsOpt.map(bdwParams => new RecordHeaderDecoderBdw(bdwParams))
68+
69+
withAdditionalInfo(readerParameters.reAdditionalInfo)
70+
.withRdwDecoder(rdwDecoder)
71+
.withBdwDecoder(bdwDecoderOpt.getOrElse(rdwDecoder))
72+
.withOptions(readerParameters.options)
73+
}
74+
5975
def withStartingRecordNumber(startingRecordNumber: Long): RawRecordContextBuilder = {
76+
require(startingRecordNumber >= 0, s"startingRecordNumber must be >= 0, got: $startingRecordNumber")
6077
context = context.copy(startingRecordNumber = startingRecordNumber)
6178
this
6279
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ class CobolProcessorBuilderSuite extends AnyWordSpec {
7979
val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
8080
val builder = CobolProcessor.builder(copybook)
8181

82-
val ext = builder.getRecordExtractor(ReaderParameters(recordLength = Some(2)), stream)
82+
val ext = builder.getRecordExtractor(ReaderParameters(recordLength = Some(2), options = Map("test" -> "option")), stream)
8383

8484
assert(ext.isInstanceOf[FixedRecordLengthRawRecordExtractor])
8585

‎cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/RecordExtractorDebugSpec.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ class RecordExtractorDebugSpec extends AnyWordSpec {
3232
val dataStream = new TestStringStream(data)
3333
val headerStream = new TestStringStream(data)
3434

35-
val ctx = RawRecordContext.builder(0, dataStream, headerStream, null).build()
35+
val ctx = RawRecordContext.builder(0L, dataStream, headerStream, null).build()
3636
val extractor = new CustomRecordExtractorMock(ctx)
3737

3838
var i = 0

‎cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VarOccursRecordExtractorSuite.scala‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ package za.co.absa.cobrix.cobol.reader.extractors.raw
1919
import org.scalatest.wordspec.AnyWordSpec
2020
import za.co.absa.cobrix.cobol.parser.CopybookParser
2121
import za.co.absa.cobrix.cobol.reader.memorystream.TestByteStream
22-
import za.co.absa.cobrix.cobol.reader.recordheader.{RecordHeaderDecoderBdw, RecordHeaderDecoderRdw, RecordHeaderParametersFactory}
2322

2423
class VarOccursRecordExtractorSuite extends AnyWordSpec {
2524
"variable occurs record extractor" should {
@@ -42,7 +41,7 @@ class VarOccursRecordExtractorSuite extends AnyWordSpec {
4241
val ibs = new TestByteStream(recordData)
4342
val hbs = new TestByteStream(recordData)
4443

45-
val rc = RawRecordContext.builder(0, ibs, hbs, copybook).build()
44+
val rc = RawRecordContext.builder(0L, ibs, hbs, copybook).build()
4645

4746
val extractor = new VarOccursRecordExtractor(rc)
4847

@@ -85,7 +84,7 @@ class VarOccursRecordExtractorSuite extends AnyWordSpec {
8584
val ibs = new TestByteStream(recordData)
8685
val hbs = new TestByteStream(recordData)
8786

88-
val rc = RawRecordContext.builder(0, ibs, hbs, copybook).build()
87+
val rc = RawRecordContext.builder(0L, ibs, hbs, copybook).build()
8988

9089

9190
val extractor = new VarOccursRecordExtractor(rc)
@@ -141,7 +140,7 @@ class VarOccursRecordExtractorSuite extends AnyWordSpec {
141140
val ibs = new TestByteStream(recordData)
142141
val hbs = new TestByteStream(recordData)
143142

144-
val rc = RawRecordContext.builder(0, ibs, hbs, copybook).build()
143+
val rc = RawRecordContext.builder(0L, ibs, hbs, copybook).build()
145144

146145
val extractor = new VarOccursRecordExtractor(rc)
147146

‎cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableBlockVariableRecordExtractorSuite.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,7 @@ class VariableBlockVariableRecordExtractorSuite extends AnyWordSpec {
151151
val bdwDecoder = new RecordHeaderDecoderBdw(RecordHeaderParametersFactory.getDummyRecordHeaderParameters(bdwBigEndian, bdwAdjustment))
152152
val rdwDecoder = new RecordHeaderDecoderRdw(RecordHeaderParametersFactory.getDummyRecordHeaderParameters(rdwBigEndian, rdwAdjustment))
153153

154-
RawRecordContext.builder(0, ibs, hbs, copybook)
154+
RawRecordContext.builder(0L, ibs, hbs, copybook)
155155
.withRdwDecoder(rdwDecoder)
156156
.withBdwDecoder(bdwDecoder)
157157
.build()

0 commit comments

Comments
 (0)