@Test void stream() throws SQLException { CompositeAutoCloseable closeable = new CompositeAutoCloseable(); when(preparedStatement.executeQuery()).thenReturn(resultSet); when(resultSet.next()).thenReturn(true).thenReturn(true).thenReturn(false); when(rowMapper.mapRow(resultSet)).thenReturn("Fred").thenReturn("Barney").thenThrow(new AssertionError("Unexpected call to row mapper")); JdbcSqlExecutor sut = JdbcSqlExecutor.of(dataSource, 100); String sql = "select name from foo where bar = ?"; List<String> result; try (Stream<String> stream = sut.stream(connection, sql, toArray("Bob"), rowMapper, closeable)) { result = stream.collect(Collectors.toList()); } verify(connection).prepareStatement(sql); verify(preparedStatement).executeQuery(); verify(preparedStatement).setObject(1, "Bob"); verify(preparedStatement).setFetchSize(100); verify(preparedStatement).close(); verify(resultSet, times(3)).next(); verify(resultSet).close(); verify(rowMapper, times(2)).mapRow(resultSet); verifyNoMoreInteractions(connection, preparedStatement, resultSet, rowMapper); assertThat(result, contains("Fred", "Barney")); }
@Override public <T> Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper) { CompositeAutoCloseable closeable = new CompositeAutoCloseable(); Connection connection = closeable.add(connect()); return stream(connection, sql, args, rowMapper, closeable); }
JdbcSqlExecutor implements SqlExecutor { @Override public <T> Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper) { CompositeAutoCloseable closeable = new CompositeAutoCloseable(); Connection connection = closeable.add(connect()); return stream(connection, sql, args, rowMapper, closeable); } }
JdbcSqlExecutor implements SqlExecutor { @Override public <T> Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper) { CompositeAutoCloseable closeable = new CompositeAutoCloseable(); Connection connection = closeable.add(connect()); return stream(connection, sql, args, rowMapper, closeable); } private  JdbcSqlExecutor(DataSource dataSource, int fetchSize, Executor executor); }
JdbcSqlExecutor implements SqlExecutor { @Override public <T> Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper) { CompositeAutoCloseable closeable = new CompositeAutoCloseable(); Connection connection = closeable.add(connect()); return stream(connection, sql, args, rowMapper, closeable); } private  JdbcSqlExecutor(DataSource dataSource, int fetchSize, Executor executor); @Override Dialect dialect(); @Override JdbcTransaction beginTransaction(); @Override DatabaseMetaData metadata(CompositeAutoCloseable autoCloseable); @Override List<T> query(String sql, Object[] args, RowMapper<T> rowMapper); @Override Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper); @Override int update(String sql, Object[] args); CompletableFuture<Integer> updateAsync(Connection connection, String sql, Object[] args); static JdbcSqlExecutor of(DataSource dataSource); static JdbcSqlExecutor of(DataSource dataSource, int fetchSize); static JdbcSqlExecutor of(DataSource dataSource, Executor executor); static JdbcSqlExecutor of(DataSource dataSource, int fetchSize, Executor executor); }
JdbcSqlExecutor implements SqlExecutor { @Override public <T> Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper) { CompositeAutoCloseable closeable = new CompositeAutoCloseable(); Connection connection = closeable.add(connect()); return stream(connection, sql, args, rowMapper, closeable); } private  JdbcSqlExecutor(DataSource dataSource, int fetchSize, Executor executor); @Override Dialect dialect(); @Override JdbcTransaction beginTransaction(); @Override DatabaseMetaData metadata(CompositeAutoCloseable autoCloseable); @Override List<T> query(String sql, Object[] args, RowMapper<T> rowMapper); @Override Stream<T> stream(String sql, Object[] args, RowMapper<T> rowMapper); @Override int update(String sql, Object[] args); CompletableFuture<Integer> updateAsync(Connection connection, String sql, Object[] args); static JdbcSqlExecutor of(DataSource dataSource); static JdbcSqlExecutor of(DataSource dataSource, int fetchSize); static JdbcSqlExecutor of(DataSource dataSource, Executor executor); static JdbcSqlExecutor of(DataSource dataSource, int fetchSize, Executor executor);  }