AccountBalancesFileLoader { @Transactional public void loadAccountBalances(@NonNull File balanceFile, DateRangeFilter dateRangeFilter) { log.info("Starting processing account balances file {}", balanceFile.getPath()); String fileName = balanceFile.getName(); Stopwatch stopwatch = Stopwatch.createStarted(); Instant startTime = Instant.now(); Connection connection = DataSourceUtils.getConnection(dataSource); try (PreparedStatement insertSetStatement = connection.prepareStatement(INSERT_SET_STATEMENT); PreparedStatement insertBalanceStatement = connection.prepareStatement(INSERT_BALANCE_STATEMENT); PreparedStatement insertTokenBalanceStatement = connection .prepareStatement(INSERT_TOKEN_BALANCE_STATEMENT); PreparedStatement updateSetStatement = connection.prepareStatement(UPDATE_SET_STATEMENT); PreparedStatement updateAccountBalanceFileStatement = connection .prepareStatement(UPDATE_ACCOUNT_BALANCE_FILE_STATEMENT); Stream<AccountBalance> stream = balanceFileReader.read(balanceFile)) { long consensusTimestamp = -1; long timestampFromFileName = Utility.getTimestampFromFilename(fileName); List<AccountBalance> accountBalanceList = new ArrayList<>(); List<TokenBalance> tokenBalanceList = new ArrayList<>(); boolean skip = false; int validCount = 0; var iter = stream.iterator(); while (iter.hasNext()) { AccountBalance accountBalance = iter.next(); if (consensusTimestamp == -1) { consensusTimestamp = accountBalance.getId().getConsensusTimestamp(); if (timestampFromFileName != consensusTimestamp) { log.error("Account balance dataset timestamp mismatch! Processing can continue, but this " + "must be investigated! Dataset {} internal timestamp {} filename timestamp {}.", fileName, consensusTimestamp, timestampFromFileName); } if (dateRangeFilter != null && !dateRangeFilter.filter(consensusTimestamp)) { log.warn("Account balances file {} not in configured date range {}, skip it", fileName, dateRangeFilter); skip = true; } else { insertAccountBalanceSet(insertSetStatement, consensusTimestamp); } } validCount++; if (!skip) { accountBalanceList.add(accountBalance); tokenBalanceList.addAll(accountBalance.getTokenBalances()); tryInsertBatchAccountBalance(insertBalanceStatement, accountBalanceList, insertBatchSize); tryInsertBatchTokenBalance(insertTokenBalanceStatement, tokenBalanceList, insertBatchSize); } } if (!skip) { tryInsertBatchAccountBalance(insertBalanceStatement, accountBalanceList, 1); tryInsertBatchTokenBalance(insertTokenBalanceStatement, tokenBalanceList, 1); updateAccountBalanceSet(updateSetStatement, consensusTimestamp); } updateAccountBalanceFile(updateAccountBalanceFileStatement, consensusTimestamp, validCount, startTime.getEpochSecond(), Instant.now().getEpochSecond(), fileName); if (!skip) { log.info("Successfully processed account balances file {}, {} records inserted in {}", fileName, validCount, stopwatch); } } catch (SQLException ex) { throw new ParserSQLException("Error processing account balance file " + fileName, ex); } finally { DataSourceUtils.releaseConnection(connection, dataSource); } } AccountBalancesFileLoader(BalanceParserProperties balanceParserProperties, DataSource dataSource,
                                     CompositeBalanceFileReader balanceFileReader); @Transactional void loadAccountBalances(@NonNull File balanceFile, DateRangeFilter dateRangeFilter);  }
@Test void loadEmptyFile() throws IOException { testFile = Files.createTempFile(null, null).toFile(); FileUtils.write(testFile, "", "utf-8"); assertThrows(Exception.class, () -> { loader.loadAccountBalances(testFile, null); }); assertThat(accountBalanceRepository.count()).isZero(); assertThat(accountBalanceSetRepository.count()).isZero(); }
@Test void rollBackWhenMissingFileInAccountBalanceFileRepo() { assertThrows(Exception.class, () -> { loader.loadAccountBalances(testFile, null); }); assertThat(accountBalanceFileRepository.count()).isZero(); assertThat(accountBalanceRepository.count()).isZero(); assertThat(accountBalanceSetRepository.count()).isZero(); }