Add Stream support to JdbcAggregateOperations

Author: serezakorotaev

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java

@@ -17,6 +17,7 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import org.springframework.dao.IncorrectUpdateSemanticsDataAccessException;
 import org.springframework.data.domain.Example;
@@ -35,6 +36,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Myeonghyeon Lee
+ * @author Sergey Korotaev
  */
 public interface JdbcAggregateOperations {
 
@@ -165,6 +167,17 @@ public interface JdbcAggregateOperations {
  */
 <T> List<T> findAllById(Iterable<?> ids, Class<T> domainType);
 
+/**
+ * Loads all entities that match one of the ids passed as an argument to a {@link Stream}.
+ * It is not guaranteed that the number of ids passed in matches the number of entities returned.
+ *
+ * @param ids the Ids of the entities to load. Must not be {@code null}.
+ * @param domainType the type of entities to load. Must not be {@code null}.
+ * @param <T> type of entities to load.
+ * @return the loaded entities. Guaranteed to be not {@code null}.
+ */
+<T> Stream<T> streamAllByIds(Iterable<?> ids, Class<T> domainType);
+
 /**
  * Load all aggregates of a given type.
  *
@@ -174,6 +187,15 @@ public interface JdbcAggregateOperations {
  */
 <T> List<T> findAll(Class<T> domainType);
 
+/**
+ * Load all aggregates of a given type to a {@link Stream}.
+ *
+ * @param domainType the type of the aggregate roots. Must not be {@code null}.
+ * @param <T> the type of the aggregate roots. Must not be {@code null}.
+ * @return Guaranteed to be not {@code null}.
+ */
+<T> Stream<T> streamAll(Class<T> domainType);
+
 /**
  * Load all aggregates of a given type, sorted.
  *
@@ -185,6 +207,17 @@ public interface JdbcAggregateOperations {
  */
 <T> List<T> findAll(Class<T> domainType, Sort sort);
 
+/**
+ * Loads all entities of the given type to a {@link Stream}, sorted.
+ *
+ * @param domainType the type of entities to load. Must not be {@code null}.
+ * @param <T> the type of entities to load.
+ * @param sort the sorting information. Must not be {@code null}.
+ * @return Guaranteed to be not {@code null}.
+ * @since 2.0
+ */
+<T> Stream<T> streamAll(Class<T> domainType, Sort sort);
+
 /**
  * Load a page of (potentially sorted) aggregates of a given type.
  *
@@ -218,6 +251,17 @@ public interface JdbcAggregateOperations {
  */
 <T> List<T> findAll(Query query, Class<T> domainType);
 
+/**
+ * Execute a {@code SELECT} query and convert the resulting items to a {@link Stream}.
+ *
+ * @param query must not be {@literal null}.
+ * @param domainType the type of entities. Must not be {@code null}.
+ * @return a non-null list with all the matching results.
+ * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+ * @since 3.0
+ */
+<T> Stream<T> streamAll(Query query, Class<T> domainType);
+
 /**
  * Returns a {@link Page} of entities matching the given {@link Query}. In case no match could be found, an empty
  * {@link Page} is returned.

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java

@@ -25,6 +25,7 @@
 import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import org.springframework.context.ApplicationContext;
@@ -68,6 +69,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Diego Krupitza
+ * @author Sergey Korotaev
  */
 public class JdbcAggregateTemplate implements JdbcAggregateOperations {
 
@@ -283,6 +285,16 @@ public <T> List<T> findAll(Class<T> domainType, Sort sort) {
 return triggerAfterConvert(all);
 }
 
+@Override
+public <T> Stream<T> streamAll(Class<T> domainType, Sort sort) {
+
+Assert.notNull(domainType, "Domain type must not be null");
+
+Stream<T> allStreamable = accessStrategy.streamAll(domainType, sort);
+
+return allStreamable.map(this::triggerAfterConvert);
+}
+
 @Override
 public <T> Page<T> findAll(Class<T> domainType, Pageable pageable) {
 
@@ -309,6 +321,11 @@ public <T> List<T> findAll(Query query, Class<T> domainType) {
 return Streamable.of(all).toList();
 }
 
+@Override
+public <T> Stream<T> streamAll(Query query, Class<T> domainType) {
+return accessStrategy.streamAll(query, domainType).map(this::triggerAfterConvert);
+}
+
 @Override
 public <T> Page<T> findAll(Query query, Class<T> domainType, Pageable pageable) {
 
@@ -327,6 +344,12 @@ public <T> List<T> findAll(Class<T> domainType) {
 return triggerAfterConvert(all);
 }
 
+@Override
+public <T> Stream<T> streamAll(Class<T> domainType) {
+Iterable<T> items = triggerAfterConvert(accessStrategy.findAll(domainType));
+return StreamSupport.stream(items.spliterator(), false).map(this::triggerAfterConvert);
+}
+
 @Override
 public <T> List<T> findAllById(Iterable<?> ids, Class<T> domainType) {
 
@@ -337,6 +360,17 @@ public <T> List<T> findAllById(Iterable<?> ids, Class<T> domainType) {
 return triggerAfterConvert(allById);
 }
 
+@Override
+public <T> Stream<T> streamAllByIds(Iterable<?> ids, Class<T> domainType) {
+
+Assert.notNull(ids, "Ids must not be null");
+Assert.notNull(domainType, "Domain type must not be null");
+
+Stream<T> allByIdStreamable = accessStrategy.streamAllByIds(ids, domainType);
+
+return allByIdStreamable.map(this::triggerAfterConvert);
+}
+
 @Override
 public <S> void delete(S aggregateRoot) {
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java

@@ -22,6 +22,7 @@
 import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.stream.Stream;
 
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -42,6 +43,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Diego Krupitza
+ * @author Sergey Korotaev
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -132,11 +134,21 @@ public <T> Iterable<T> findAll(Class<T> domainType) {
 return collect(das -> das.findAll(domainType));
 }
 
+@Override
+public <T> Stream<T> streamAll(Class<T> domainType) {
+return collect(das -> das.streamAll(domainType));
+}
+
 @Override
 public <T> Iterable<T> findAllById(Iterable<?> ids, Class<T> domainType) {
 return collect(das -> das.findAllById(ids, domainType));
 }
 
+@Override
+public <T> Stream<T> streamAllByIds(Iterable<?> ids, Class<T> domainType) {
+return collect(das -> das.streamAllByIds(ids, domainType));
+}
+
 @Override
 public Iterable<Object> findAllByPath(Identifier identifier,
 PersistentPropertyPath<? extends RelationalPersistentProperty> path) {
@@ -153,6 +165,11 @@ public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
 return collect(das -> das.findAll(domainType, sort));
 }
 
+@Override
+public <T> Stream<T> streamAll(Class<T> domainType, Sort sort) {
+return collect(das -> das.streamAll(domainType, sort));
+}
+
 @Override
 public <T> Iterable<T> findAll(Class<T> domainType, Pageable pageable) {
 return collect(das -> das.findAll(domainType, pageable));
@@ -168,6 +185,11 @@ public <T> Iterable<T> findAll(Query query, Class<T> domainType) {
 return collect(das -> das.findAll(query, domainType));
 }
 
+@Override
+public <T> Stream<T> streamAll(Query query, Class<T> domainType) {
+return collect(das -> das.streamAll(query, domainType));
+}
+
 @Override
 public <T> Iterable<T> findAll(Query query, Class<T> domainType, Pageable pageable) {
 return collect(das -> das.findAll(query, domainType, pageable));

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DataAccessStrategy.java

@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.domain.Pageable;
@@ -41,6 +42,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Diego Krupitza
+ * @author Sergey Korotaev
  */
 public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationResolver {
 
@@ -252,6 +254,16 @@ public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationR
 @Override
 <T> Iterable<T> findAll(Class<T> domainType);
 
+/**
+ * Loads all entities of the given type to a {@link Stream}.
+ *
+ * @param domainType the type of entities to load. Must not be {@code null}.
+ * @param <T> the type of entities to load.
+ * @return Guaranteed to be not {@code null}.
+ */
+@Override
+<T> Stream<T> streamAll(Class<T> domainType);
+
 /**
  * Loads all entities that match one of the ids passed as an argument. It is not guaranteed that the number of ids
  * passed in matches the number of entities returned.
@@ -264,6 +276,18 @@ public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationR
 @Override
 <T> Iterable<T> findAllById(Iterable<?> ids, Class<T> domainType);
 
+/**
+ * Loads all entities that match one of the ids passed as an argument to a {@link Stream}.
+ * It is not guaranteed that the number of ids passed in matches the number of entities returned.
+ *
+ * @param ids the Ids of the entities to load. Must not be {@code null}.
+ * @param domainType the type of entities to load. Must not be {@code null}.
+ * @param <T> type of entities to load.
+ * @return the loaded entities. Guaranteed to be not {@code null}.
+ */
+@Override
+<T> Stream<T> streamAllByIds(Iterable<?> ids, Class<T> domainType);
+
 @Override
 Iterable<Object> findAllByPath(Identifier identifier,
 PersistentPropertyPath<? extends RelationalPersistentProperty> path);
@@ -280,6 +304,18 @@ Iterable<Object> findAllByPath(Identifier identifier,
 @Override
 <T> Iterable<T> findAll(Class<T> domainType, Sort sort);
 
+/**
+ * Loads all entities of the given type to a {@link Stream}, sorted.
+ *
+ * @param domainType the type of entities to load. Must not be {@code null}.
+ * @param <T> the type of entities to load.
+ * @param sort the sorting information. Must not be {@code null}.
+ * @return Guaranteed to be not {@code null}.
+ * @since 2.0
+ */
+@Override
+<T> Stream<T> streamAll(Class<T> domainType, Sort sort);
+
 /**
  * Loads all entities of the given type, paged and sorted.
  *
@@ -316,6 +352,18 @@ Iterable<Object> findAllByPath(Identifier identifier,
 @Override
 <T> Iterable<T> findAll(Query query, Class<T> domainType);
 
+/**
+ * Execute a {@code SELECT} query and convert the resulting items to a {@link Stream}.
+ *
+ * @param query must not be {@literal null}.
+ * @param domainType the type of entities. Must not be {@code null}.
+ * @return a non-null list with all the matching results.
+ * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+ * @since 3.0
+ */
+@Override
+<T> Stream<T> streamAll(Query query, Class<T> domainType);
+
 /**
  * Execute a {@code SELECT} query and convert the resulting items to a {@link Iterable}. Applies the {@link Pageable}
  * to the result.

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java

@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.dao.OptimisticLockingFailureException;
@@ -60,6 +61,7 @@
  * @author Radim Tlusty
  * @author Chirag Tailor
  * @author Diego Krupitza
+ * @author Sergey Korotaev
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -276,6 +278,11 @@ public <T> List<T> findAll(Class<T> domainType) {
 return operations.query(sql(domainType).getFindAll(), getEntityRowMapper(domainType));
 }
 
+@Override
+public <T> Stream<T> streamAll(Class<T> domainType) {
+return operations.queryForStream(sql(domainType).getFindAll(), new MapSqlParameterSource(), getEntityRowMapper(domainType));
+}
+
 @Override
 public <T> List<T> findAllById(Iterable<?> ids, Class<T> domainType) {
 
@@ -288,6 +295,19 @@ public <T> List<T> findAllById(Iterable<?> ids, Class<T> domainType) {
 return operations.query(findAllInListSql, parameterSource, getEntityRowMapper(domainType));
 }
 
+@Override
+public <T> Stream<T> streamAllByIds(Iterable<?> ids, Class<T> domainType) {
+
+if (!ids.iterator().hasNext()) {
+return Stream.empty();
+}
+
+SqlParameterSource parameterSource = sqlParametersFactory.forQueryByIds(ids, domainType);
+String findAllInListSql = sql(domainType).getFindAllInList();
+
+return operations.queryForStream(findAllInListSql, parameterSource, getEntityRowMapper(domainType));
+}
+
 @Override
 @SuppressWarnings("unchecked")
 public List<Object> findAllByPath(Identifier identifier,
@@ -342,6 +362,11 @@ public <T> List<T> findAll(Class<T> domainType, Sort sort) {
 return operations.query(sql(domainType).getFindAll(sort), getEntityRowMapper(domainType));
 }
 
+@Override
+public <T> Stream<T> streamAll(Class<T> domainType, Sort sort) {
+return operations.queryForStream(sql(domainType).getFindAll(sort), new MapSqlParameterSource(), getEntityRowMapper(domainType));
+}
+
 @Override
 public <T> List<T> findAll(Class<T> domainType, Pageable pageable) {
 return operations.query(sql(domainType).getFindAll(pageable), getEntityRowMapper(domainType));
@@ -369,6 +394,15 @@ public <T> List<T> findAll(Query query, Class<T> domainType) {
 return operations.query(sqlQuery, parameterSource, getEntityRowMapper(domainType));
 }
 
+@Override
+public <T> Stream<T> streamAll(Query query, Class<T> domainType) {
+
+MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+String sqlQuery = sql(domainType).selectByQuery(query, parameterSource);
+
+return operations.queryForStream(sqlQuery, parameterSource, getEntityRowMapper(domainType));
+}
+
 @Override
 public <T> List<T> findAll(Query query, Class<T> domainType, Pageable pageable) {