@Test void should_find_stocks_when_find_all_by_symbol_given_valid_input() throws ExecutionException, InterruptedException { given(findBySymbol.bind("ABC", i1, i2)).willReturn(bound); given(session.executeAsync(bound)).willReturn(completedStage(resultSet)); List<Row> rows = List.of(row1, row2); given(resultSet.currentPage()).willReturn(rows); given(resultSet.hasMorePages()).willReturn(false); given(row1.getString(SYMBOL)).willReturn(stock1.getSymbol()); given(row1.getInstant(DATE)).willReturn(stock1.getDate()); given(row1.getBigDecimal(VALUE)).willReturn(stock1.getValue()); given(row2.getString(SYMBOL)).willReturn(stock2.getSymbol()); given(row2.getInstant(DATE)).willReturn(stock2.getDate()); given(row2.getBigDecimal(VALUE)).willReturn(stock2.getValue()); var stockRepository = new AsyncStockRepository(session, insert, delete, findById, findBySymbol, rowMapper); var future = stockRepository.findAllBySymbol("ABC", i1, i2, 0, 10); assertThat(future).isNotNull().isCompleted(); var result = future.toCompletableFuture().get(); assertThat(result).containsExactly(stock1, stock2); verify(session).executeAsync(bound); }
@NonNull public CompletionStage<Stream<Stock>> findAllBySymbol( @NonNull String symbol, @NonNull Instant start, @NonNull Instant end, long offset, long limit) { BoundStatement bound = findBySymbol.bind(symbol, start, end); CompletionStage<AsyncResultSet> stage = session.executeAsync(bound); return stage .thenCompose(first -> new RowCollector(first, offset, limit)) .thenApply(rows -> rows.stream().map(rowMapper)); }
AsyncStockRepository { @NonNull public CompletionStage<Stream<Stock>> findAllBySymbol( @NonNull String symbol, @NonNull Instant start, @NonNull Instant end, long offset, long limit) { BoundStatement bound = findBySymbol.bind(symbol, start, end); CompletionStage<AsyncResultSet> stage = session.executeAsync(bound); return stage .thenCompose(first -> new RowCollector(first, offset, limit)) .thenApply(rows -> rows.stream().map(rowMapper)); } }
AsyncStockRepository { @NonNull public CompletionStage<Stream<Stock>> findAllBySymbol( @NonNull String symbol, @NonNull Instant start, @NonNull Instant end, long offset, long limit) { BoundStatement bound = findBySymbol.bind(symbol, start, end); CompletionStage<AsyncResultSet> stage = session.executeAsync(bound); return stage .thenCompose(first -> new RowCollector(first, offset, limit)) .thenApply(rows -> rows.stream().map(rowMapper)); } AsyncStockRepository(
      CqlSession session,
      @Qualifier("stocks.prepared.insert") PreparedStatement insert,
      @Qualifier("stocks.prepared.deleteById") PreparedStatement deleteById,
      @Qualifier("stocks.prepared.findById") PreparedStatement findById,
      @Qualifier("stocks.prepared.findBySymbol") PreparedStatement findBySymbol,
      Function<Row, Stock> rowMapper); }
AsyncStockRepository { @NonNull public CompletionStage<Stream<Stock>> findAllBySymbol( @NonNull String symbol, @NonNull Instant start, @NonNull Instant end, long offset, long limit) { BoundStatement bound = findBySymbol.bind(symbol, start, end); CompletionStage<AsyncResultSet> stage = session.executeAsync(bound); return stage .thenCompose(first -> new RowCollector(first, offset, limit)) .thenApply(rows -> rows.stream().map(rowMapper)); } AsyncStockRepository(
      CqlSession session,
      @Qualifier("stocks.prepared.insert") PreparedStatement insert,
      @Qualifier("stocks.prepared.deleteById") PreparedStatement deleteById,
      @Qualifier("stocks.prepared.findById") PreparedStatement findById,
      @Qualifier("stocks.prepared.findBySymbol") PreparedStatement findBySymbol,
      Function<Row, Stock> rowMapper); @NonNull CompletionStage<Stock> save(@NonNull Stock stock); @NonNull CompletionStage<Void> deleteById(@NonNull String symbol, @NonNull Instant date); @NonNull CompletionStage<Optional<Stock>> findById(@NonNull String symbol, @NonNull Instant date); @NonNull CompletionStage<Stream<Stock>> findAllBySymbol(
      @NonNull String symbol,
      @NonNull Instant start,
      @NonNull Instant end,
      long offset,
      long limit); }
AsyncStockRepository { @NonNull public CompletionStage<Stream<Stock>> findAllBySymbol( @NonNull String symbol, @NonNull Instant start, @NonNull Instant end, long offset, long limit) { BoundStatement bound = findBySymbol.bind(symbol, start, end); CompletionStage<AsyncResultSet> stage = session.executeAsync(bound); return stage .thenCompose(first -> new RowCollector(first, offset, limit)) .thenApply(rows -> rows.stream().map(rowMapper)); } AsyncStockRepository(
      CqlSession session,
      @Qualifier("stocks.prepared.insert") PreparedStatement insert,
      @Qualifier("stocks.prepared.deleteById") PreparedStatement deleteById,
      @Qualifier("stocks.prepared.findById") PreparedStatement findById,
      @Qualifier("stocks.prepared.findBySymbol") PreparedStatement findBySymbol,
      Function<Row, Stock> rowMapper); @NonNull CompletionStage<Stock> save(@NonNull Stock stock); @NonNull CompletionStage<Void> deleteById(@NonNull String symbol, @NonNull Instant date); @NonNull CompletionStage<Optional<Stock>> findById(@NonNull String symbol, @NonNull Instant date); @NonNull CompletionStage<Stream<Stock>> findAllBySymbol(
      @NonNull String symbol,
      @NonNull Instant start,
      @NonNull Instant end,
      long offset,
      long limit);  }