RecordFileParser implements FileParser { @Override @Transactional public void parse(StreamFileData streamFileData) { Instant startTime = Instant.now(); String expectedPrevFileHash = applicationStatusRepository.findByStatusCode(ApplicationStatusCode.LAST_PROCESSED_RECORD_HASH); DateRangeFilter dateRangeFilter = mirrorDateRangePropertiesProcessor.getDateRangeFilter(parserProperties.getStreamType()); AtomicInteger counter = new AtomicInteger(0); boolean success = false; try { RecordFile recordFileDb = recordStreamFileListener.onStart(streamFileData); Stopwatch stopwatch = Stopwatch.createStarted(); RecordFile recordFile = Utility.parseRecordFile( streamFileData.getFilename(), recordItem -> { if (processRecordItem(recordItem, dateRangeFilter)) { counter.incrementAndGet(); } }); if (!Utility.verifyHashChain(recordFile.getPreviousHash(), expectedPrevFileHash, parserProperties.getMirrorProperties().getVerifyHashAfter(), recordFile.getName())) { throw new HashMismatchException(recordFile.getName(), expectedPrevFileHash, recordFile.getPreviousHash()); } log.info("Time to parse record file: {}ms", stopwatch.elapsed(TimeUnit.MILLISECONDS)); recordFileDb.setLoadStart(startTime.getEpochSecond()); recordFileDb.setLoadEnd(Instant.now().getEpochSecond()); recordStreamFileListener.onEnd(recordFileDb); recordParserLatencyMetric(recordFileDb); success = true; } catch (Exception ex) { recordStreamFileListener.onError(); throw ex; } finally { var elapsedTimeMillis = Duration.between(startTime, Instant.now()).toMillis(); var rate = elapsedTimeMillis > 0 ? (int) (1000.0 * counter.get() / elapsedTimeMillis) : 0; log.info("Finished parsing {} transactions from record file {} in {}ms ({}/s). Success: {}", counter, streamFileData.getFilename(), elapsedTimeMillis, rate, success); parseDurationMetrics.get(success).record(elapsedTimeMillis, TimeUnit.MILLISECONDS); } } RecordFileParser(ApplicationStatusRepository applicationStatusRepository,
                            RecordParserProperties parserProperties, MeterRegistry meterRegistry,
                            RecordItemListener recordItemListener,
                            RecordStreamFileListener recordStreamFileListener,
                            MirrorDateRangePropertiesProcessor mirrorDateRangePropertiesProcessor); @Override @Transactional void parse(StreamFileData streamFileData);  }
@Test void parse() throws Exception { recordFileParser.parse(streamFileData1); assertProcessedFile(streamFileData1, recordFile1, NUM_TXNS_FILE_1); recordFileParser.parse(streamFileData2); verify(recordStreamFileListener, never()).onError(); assertAllProcessed(); }
@Test void invalidFile() throws Exception { FileUtils.writeStringToFile(file1, "corrupt", "UTF-8"); streamFileData1 = new StreamFileData(file1.toString(), new FileInputStream(file1)); Assertions.assertThrows(IllegalArgumentException.class, () -> { recordFileParser.parse(streamFileData1); }); verify(recordStreamFileListener).onStart(streamFileData1); verify(recordStreamFileListener, never()).onEnd(recordFile1); verify(recordStreamFileListener).onError(); }
@Test void hashMismatch() { when(applicationStatusRepository.findByStatusCode(LAST_PROCESSED_RECORD_HASH)).thenReturn("123"); Assertions.assertThrows(HashMismatchException.class, () -> { recordFileParser.parse(streamFileData1); }); verify(recordStreamFileListener).onStart(streamFileData1); verify(recordStreamFileListener, never()).onEnd(any()); verify(recordStreamFileListener).onError(); }
@Test void bypassHashMismatch() throws Exception { parserProperties.getMirrorProperties().setVerifyHashAfter(Instant.parse("2019-09-01T00:00:00.000000Z")); when(applicationStatusRepository.findByStatusCode(LAST_PROCESSED_RECORD_HASH)).thenReturn("123"); recordFileParser.parse(streamFileData1); verify(recordStreamFileListener, never()).onError(); assertProcessedFile(streamFileData1, recordFile1, NUM_TXNS_FILE_1); }
@Test void failureProcessingItemShouldRollback() { doThrow(ParserSQLException.class).when(recordItemListener).onItem(any()); Assertions.assertThrows(IllegalArgumentException.class, () -> { recordFileParser.parse(streamFileData1); }); verify(recordStreamFileListener).onStart(streamFileData1); verify(recordStreamFileListener, never()).onEnd(recordFile1); verify(recordStreamFileListener).onError(); }