ENH: Add Query findEach() with batch consumer

This is a variation of findEach() that makes it easy to have a batch consumer processing a large result in batches.  For example, process in batches of 50 beans.

Note that the last batch consumed/processed will often have less than the batch size.

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

@@ -1446,6 +1446,18 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
     // no try finally - findEach guarantee's cleanup of the transaction if required
   }
 
+  @Override
+  public <T> void findEach(Query<T> query, int batch, Consumer<List<T>> consumer, Transaction t) {
+    SpiOrmQueryRequest<T> request = createQueryRequest(Type.ITERATE, query, t);
+//    if (request.isUseDocStore()) {
+//      docStore().findEach(request, consumer);
+//      return;
+//    }
+    request.initTransIfRequired();
+    request.findEach(batch, consumer);
+    // no try finally - findEach guarantee's cleanup of the transaction if required
+  }
+
   @Override
   public <T> void findEachWhile(Query<T> query, Predicate<T> consumer, Transaction t) {
     SpiOrmQueryRequest<T> request = createQueryRequest(Type.ITERATE, query, t);

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

@@ -430,6 +430,24 @@ public final class OrmQueryRequest<T> extends BeanRequest implements SpiOrmQuery
     }
   }
 
+  @Override
+  public void findEach(int batch, Consumer<List<T>> batchConsumer) {
+    final List<T> buffer = new ArrayList<>(batch);
+    try (QueryIterator<T> it = queryEngine.findIterate(this)) {
+      while (it.hasNext()) {
+        buffer.add(it.next());
+        if (buffer.size() >= batch) {
+          batchConsumer.accept(buffer);
+          buffer.clear();
+        }
+      }
+      if (!buffer.isEmpty()) {
+        // consume the remainder
+        batchConsumer.accept(buffer);
+      }
+    }
+  }
+
   @Override
   public void findEachWhile(Predicate<T> consumer) {
     try (QueryIterator<T> it = queryEngine.findIterate(this)) {

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

@@ -80,10 +80,15 @@ public interface SpiOrmQueryRequest<T> extends BeanQueryRequest<T>, DocQueryRequ
   <A> List<A> findIds();
 
   /**
-   * Execute the find returning a QueryIterator and visitor pattern.
+   * Execute findEach iterating results one bean at a time.
    */
   void findEach(Consumer<T> consumer);
 
+  /**
+   * Execute findEach with a batch consumer.
+   */
+  void findEach(int batch, Consumer<List<T>> batchConsumer);
+
   /**
    * Execute the find returning a QueryIterator and visitor pattern.
    */

ebean-core/src/main/java/io/ebeaninternal/server/expression/DefaultExpressionList.java

@@ -436,6 +436,11 @@ public class DefaultExpressionList<T> implements SpiExpressionList<T> {
     query.findEach(consumer);
   }
 
+  @Override
+  public void findEach(int batch, Consumer<List<T>> consumer) {
+    query.findEach(batch, consumer);
+  }
+
   @Override
   public void findEachWhile(Predicate<T> consumer) {
     query.findEachWhile(consumer);

ebean-core/src/main/java/io/ebeaninternal/server/expression/JunctionExpression.java

@@ -446,6 +446,11 @@ class JunctionExpression<T> implements SpiJunction<T>, SpiExpression, Expression
     exprList.findEach(consumer);
   }
 
+  @Override
+  public void findEach(int batch, Consumer<List<T>> consumer) {
+    exprList.findEach(batch, consumer);
+  }
+
   @Override
   public void findEachWhile(Predicate<T> consumer) {
     exprList.findEachWhile(consumer);

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

@@ -263,6 +263,11 @@ class DefaultFetchGroupQuery<T> implements SpiFetchGroupQuery<T>, SpiQueryFetch
     throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup");
   }
 
+  @Override
+  public void findEach(int batch, Consumer<List<T>> consumer) {
+    throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup");
+  }
+
   @Override
   public void findEachWhile(Predicate<T> consumer) {
     throw new RuntimeException("EB102: Only select() and fetch() clause is allowed on FetchGroup");

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

@@ -1551,6 +1551,11 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     server.findEach(this, consumer, transaction);
   }
 
+  @Override
+  public void findEach(int batch, Consumer<List<T>> consumer) {
+    server.findEach(this, batch, consumer, transaction);
+  }
+
   @Override
   public QueryIterator<T> findIterate() {
     return server.findIterate(this, transaction);