#2200 - ENH: Add DtoQuery findEach with batch consumer - findEach(int batchSize, Consumer<List<T>> consumer)

ebean-core/src/main/java/io/ebeaninternal/api/SpiEbeanServer.java

@@ -264,6 +264,11 @@ public interface SpiEbeanServer extends ExtendedServer, EbeanServer, BeanCollect
    */
   <T> void findDtoEach(SpiDtoQuery<T> query, Consumer<T> consumer);
 
+  /**
+   * DTO findEach batch query.
+   */
+  <T> void findDtoEach(SpiDtoQuery<T> query, int batch, Consumer<List<T>> consumer);
+
   /**
    * DTO findEachWhile query.
    */

ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultServer.java

@@ -1597,6 +1597,17 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
     }
   }
 
+  @Override
+  public <T> void findDtoEach(SpiDtoQuery<T> query, int batch, Consumer<List<T>> consumer) {
+    DtoQueryRequest<T> request = new DtoQueryRequest<>(this, dtoQueryEngine, query);
+    try {
+      request.initTransIfRequired();
+      request.findEach(batch, consumer);
+    } finally {
+      request.endTransIfRequired();
+    }
+  }
+
   @Override
   public <T> void findDtoEachWhile(SpiDtoQuery<T> query, Predicate<T> consumer) {
     DtoQueryRequest<T> request = new DtoQueryRequest<>(this, dtoQueryEngine, query);

ebean-core/src/main/java/io/ebeaninternal/server/core/DtoQueryRequest.java

@@ -92,6 +92,11 @@ public final class DtoQueryRequest<T> extends AbstractSqlQueryRequest {
     queryEngine.findEach(this, consumer);
   }
 
+  public void findEach(int batch, Consumer<List<T>> consumer) {
+    flushJdbcBatchOnQuery();
+    queryEngine.findEach(this, batch, consumer);
+  }
+
   public void findEachWhile(Predicate<T> consumer) {
     flushJdbcBatchOnQuery();
     queryEngine.findEachWhile(this, consumer);

ebean-core/src/main/java/io/ebeaninternal/server/query/DtoQueryEngine.java

@@ -42,7 +42,28 @@ public class DtoQueryEngine {
       }
     } catch (Exception e) {
       throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
+    } finally {
+      request.close();
+    }
+  }
 
+  public <T> void findEach(DtoQueryRequest<T> request, int batchSize, Consumer<List<T>> consumer) {
+    try {
+      List<T> buffer = new ArrayList<>();
+      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      while (request.next()) {
+        buffer.add(request.readNextBean());
+        if (buffer.size() >= batchSize) {
+          consumer.accept(buffer);
+          buffer.clear();
+        }
+      }
+      if (!buffer.isEmpty()) {
+        // consume the remainder
+        consumer.accept(buffer);
+      }
+    } catch (Exception e) {
+      throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
     } finally {
       request.close();
     }

ebean-core/src/main/java/io/ebeaninternal/server/querydefn/DefaultDtoQuery.java

@@ -102,6 +102,11 @@ public class DefaultDtoQuery<T> implements SpiDtoQuery<T> {
     server.findDtoEach(this, consumer);
   }
 
+  @Override
+  public void findEach(int batch, Consumer<List<T>> consumer) {
+    server.findDtoEach(this, batch, consumer);
+  }
+
   @Override
   public void findEachWhile(Predicate<T> consumer) {
     server.findDtoEachWhile(this, consumer);