@Test public void readBatchFiltersOutNulls() throws IOException { Predicate<Long> isOdd = num -> num % 2 != 0; when(mockLog.readRound(anyLong())) .thenAnswer(invocation -> { long sequence = (long) invocation.getArguments()[0]; if (!isOdd.test(sequence)) { return null; } return valueForRound(sequence).persistToBytes(); }); try (PaxosStateLogBatchReader<PaxosValue> reader = createReader()) { assertThat(reader.readBatch(START_SEQUENCE, BATCH_SIZE)) .isEqualTo(EXPECTED_ROUNDS.stream() .filter(round -> isOdd.test(round.sequence())) .collect(Collectors.toList())); } }
public List<PaxosRound<V>> readBatch(long startSequence, int numEntries) { return AtlasFutures.getUnchecked( Futures.allAsList( LongStream.range(startSequence, startSequence + numEntries) .mapToObj(sequence -> executor.submit(() -> singleRead(sequence))) .collect(Collectors.toList()))) .stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList()); }
PaxosStateLogBatchReader implements AutoCloseable { public List<PaxosRound<V>> readBatch(long startSequence, int numEntries) { return AtlasFutures.getUnchecked( Futures.allAsList( LongStream.range(startSequence, startSequence + numEntries) .mapToObj(sequence -> executor.submit(() -> singleRead(sequence))) .collect(Collectors.toList()))) .stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList()); } }
PaxosStateLogBatchReader implements AutoCloseable { public List<PaxosRound<V>> readBatch(long startSequence, int numEntries) { return AtlasFutures.getUnchecked( Futures.allAsList( LongStream.range(startSequence, startSequence + numEntries) .mapToObj(sequence -> executor.submit(() -> singleRead(sequence))) .collect(Collectors.toList()))) .stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList()); } PaxosStateLogBatchReader(PaxosStateLog<V> delegate, Persistable.Hydrator<V> hydrator, int numThreads); }
PaxosStateLogBatchReader implements AutoCloseable { public List<PaxosRound<V>> readBatch(long startSequence, int numEntries) { return AtlasFutures.getUnchecked( Futures.allAsList( LongStream.range(startSequence, startSequence + numEntries) .mapToObj(sequence -> executor.submit(() -> singleRead(sequence))) .collect(Collectors.toList()))) .stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList()); } PaxosStateLogBatchReader(PaxosStateLog<V> delegate, Persistable.Hydrator<V> hydrator, int numThreads); List<PaxosRound<V>> readBatch(long startSequence, int numEntries); @Override void close(); }
PaxosStateLogBatchReader implements AutoCloseable { public List<PaxosRound<V>> readBatch(long startSequence, int numEntries) { return AtlasFutures.getUnchecked( Futures.allAsList( LongStream.range(startSequence, startSequence + numEntries) .mapToObj(sequence -> executor.submit(() -> singleRead(sequence))) .collect(Collectors.toList()))) .stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList()); } PaxosStateLogBatchReader(PaxosStateLog<V> delegate, Persistable.Hydrator<V> hydrator, int numThreads); List<PaxosRound<V>> readBatch(long startSequence, int numEntries); @Override void close();  }