@Test void readValid() throws IOException { Stream<AccountBalance> accountBalanceStream = balanceFileReader.read(balanceFile); verifySuccess(balanceFile, accountBalanceStream, sampleConsensusTimestamp, 2); }
@Override public Stream<AccountBalance> read(File file) { if (file == null) { throw new InvalidDatasetException("Null file provided to balance file reader"); } try { BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)), balanceParserProperties.getFileBufferSize()); long consensusTimestamp = parseHeaderForConsensusTimestamp(reader); long shard = balanceParserProperties.getMirrorProperties().getShard(); return reader.lines() .map(line -> { try { return parser.parse(line, consensusTimestamp, shard); } catch (InvalidDatasetException ex) { log.error(ex); return null; } }) .filter(Objects::nonNull) .onClose(() -> { try { reader.close(); } catch (Exception ex) { } }); } catch (IOException ex) { throw new InvalidDatasetException("Error reading account balance file", ex); } }
BalanceFileReaderImplV2 implements BalanceFileReader { @Override public Stream<AccountBalance> read(File file) { if (file == null) { throw new InvalidDatasetException("Null file provided to balance file reader"); } try { BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)), balanceParserProperties.getFileBufferSize()); long consensusTimestamp = parseHeaderForConsensusTimestamp(reader); long shard = balanceParserProperties.getMirrorProperties().getShard(); return reader.lines() .map(line -> { try { return parser.parse(line, consensusTimestamp, shard); } catch (InvalidDatasetException ex) { log.error(ex); return null; } }) .filter(Objects::nonNull) .onClose(() -> { try { reader.close(); } catch (Exception ex) { } }); } catch (IOException ex) { throw new InvalidDatasetException("Error reading account balance file", ex); } } }
BalanceFileReaderImplV2 implements BalanceFileReader { @Override public Stream<AccountBalance> read(File file) { if (file == null) { throw new InvalidDatasetException("Null file provided to balance file reader"); } try { BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)), balanceParserProperties.getFileBufferSize()); long consensusTimestamp = parseHeaderForConsensusTimestamp(reader); long shard = balanceParserProperties.getMirrorProperties().getShard(); return reader.lines() .map(line -> { try { return parser.parse(line, consensusTimestamp, shard); } catch (InvalidDatasetException ex) { log.error(ex); return null; } }) .filter(Objects::nonNull) .onClose(() -> { try { reader.close(); } catch (Exception ex) { } }); } catch (IOException ex) { throw new InvalidDatasetException("Error reading account balance file", ex); } }  }
BalanceFileReaderImplV2 implements BalanceFileReader { @Override public Stream<AccountBalance> read(File file) { if (file == null) { throw new InvalidDatasetException("Null file provided to balance file reader"); } try { BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)), balanceParserProperties.getFileBufferSize()); long consensusTimestamp = parseHeaderForConsensusTimestamp(reader); long shard = balanceParserProperties.getMirrorProperties().getShard(); return reader.lines() .map(line -> { try { return parser.parse(line, consensusTimestamp, shard); } catch (InvalidDatasetException ex) { log.error(ex); return null; } }) .filter(Objects::nonNull) .onClose(() -> { try { reader.close(); } catch (Exception ex) { } }); } catch (IOException ex) { throw new InvalidDatasetException("Error reading account balance file", ex); } }  @Override Stream<AccountBalance> read(File file); boolean isFirstLineFromFileVersion(String firstLine); }
BalanceFileReaderImplV2 implements BalanceFileReader { @Override public Stream<AccountBalance> read(File file) { if (file == null) { throw new InvalidDatasetException("Null file provided to balance file reader"); } try { BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)), balanceParserProperties.getFileBufferSize()); long consensusTimestamp = parseHeaderForConsensusTimestamp(reader); long shard = balanceParserProperties.getMirrorProperties().getShard(); return reader.lines() .map(line -> { try { return parser.parse(line, consensusTimestamp, shard); } catch (InvalidDatasetException ex) { log.error(ex); return null; } }) .filter(Objects::nonNull) .onClose(() -> { try { reader.close(); } catch (Exception ex) { } }); } catch (IOException ex) { throw new InvalidDatasetException("Error reading account balance file", ex); } }  @Override Stream<AccountBalance> read(File file); boolean isFirstLineFromFileVersion(String firstLine);  }