#2246 - ENH: Add DtoQuery findStream() and findIterate()

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

@@ -1,14 +1,6 @@
 package io.ebeaninternal.api;
 
-import io.ebean.DtoQuery;
-import io.ebean.EbeanServer;
-import io.ebean.ExtendedServer;
-import io.ebean.PersistenceContextScope;
-import io.ebean.Query;
-import io.ebean.RowConsumer;
-import io.ebean.RowMapper;
-import io.ebean.Transaction;
-import io.ebean.TxScope;
+import io.ebean.*;
 import io.ebean.bean.BeanCollectionLoader;
 import io.ebean.bean.CallOrigin;
 import io.ebean.config.DatabaseConfig;
@@ -25,6 +17,7 @@ import io.ebeaninternal.server.transaction.RemoteTransactionEvent;
 import java.util.List;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
+import java.util.stream.Stream;
 
 /**
  * Service Provider extension to EbeanServer.
@@ -254,6 +247,16 @@ public interface SpiEbeanServer extends ExtendedServer, EbeanServer, BeanCollect
    */
   void findEachRow(SpiSqlQuery query, RowConsumer consumer);
 
+  /**
+   * DTO findIterate query.
+   */
+  <T> QueryIterator<T> findDtoIterate(SpiDtoQuery<T> query);
+
+  /**
+   * DTO findStream query.
+   */
+  <T> Stream<T> findDtoStream(SpiDtoQuery<T> query);
+
   /**
    * DTO findList query.
    */

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

@@ -1433,14 +1433,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   @Nonnull
   @Override
   public <T> Stream<T> findStream(Query<T> query, Transaction transaction) {
-    SpiOrmQueryRequest<T> request = createQueryRequest(Type.ITERATE, query, transaction);
-    try {
-      request.initTransIfRequired();
-      return toStream(request.findIterate());
-    } catch (RuntimeException ex) {
-      request.endTransIfRequired();
-      throw ex;
-    }
+    return toStream(findIterate(query, transaction));
   }
 
   private <T> Stream<T> toStream(QueryIterator<T> queryIterator) {
@@ -1640,6 +1633,23 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
     }
   }
 
+  @Override
+  public <T> QueryIterator<T> findDtoIterate(SpiDtoQuery<T> query) {
+    DtoQueryRequest<T> request = new DtoQueryRequest<>(this, dtoQueryEngine, query);
+    try {
+      request.initTransIfRequired();
+      return request.findIterate();
+    } catch (RuntimeException ex) {
+      request.endTransIfRequired();
+      throw ex;
+    }
+  }
+
+  @Override
+  public <T> Stream<T> findDtoStream(SpiDtoQuery<T> query) {
+    return toStream(findDtoIterate(query));
+  }
+
   @Override
   public <T> List<T> findDtoList(SpiDtoQuery<T> query) {
     DtoQueryRequest<T> request = new DtoQueryRequest<>(this, dtoQueryEngine, query);

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

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.core;
 
+import io.ebean.QueryIterator;
 import io.ebean.core.type.DataReader;
 import io.ebeaninternal.api.SpiDtoQuery;
 import io.ebeaninternal.api.SpiEbeanServer;
@@ -90,6 +91,11 @@ public final class DtoQueryRequest<T> extends AbstractSqlQueryRequest {
     }
   }
 
+  public QueryIterator<T> findIterate() {
+    flushJdbcBatchOnQuery();
+    return queryEngine.findIterate(this);
+  }
+
   public void findEach(Consumer<T> consumer) {
     flushJdbcBatchOnQuery();
     queryEngine.findEach(this, consumer);

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

@@ -1,10 +1,12 @@
 package io.ebeaninternal.server.query;
 
+import io.ebean.QueryIterator;
 import io.ebeaninternal.api.SpiQuery;
 import io.ebeaninternal.server.core.DtoQueryRequest;
 import io.ebeaninternal.server.persist.Binder;
 
 import javax.persistence.PersistenceException;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
@@ -34,6 +36,15 @@ public class DtoQueryEngine {
     }
   }
 
+  public <T> QueryIterator<T> findIterate(DtoQueryRequest<T> request) {
+    try {
+      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      return new DtoQueryIterator<>(request);
+    } catch (SQLException e) {
+      throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
+    }
+  }
+
   public <T> void findEach(DtoQueryRequest<T> request, Consumer<T> consumer) {
      try {
       request.executeSql(binder, SpiQuery.Type.ITERATE);

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

@@ -0,0 +1,55 @@
+package io.ebeaninternal.server.query;
+
+import io.ebean.QueryIterator;
+import io.ebeaninternal.server.core.DtoQueryRequest;
+
+import javax.persistence.PersistenceException;
+import java.sql.SQLException;
+
+class DtoQueryIterator<T> implements QueryIterator<T> {
+
+  private final DtoQueryRequest<T> request;
+  private boolean closed;
+
+  DtoQueryIterator(DtoQueryRequest<T> request) {
+    this.request = request;
+  }
+
+  @Override
+  public boolean hasNext() {
+    boolean result = false;
+    try {
+      result = request.next();
+      return result;
+    } catch (SQLException e) {
+      throw new PersistenceException(e);
+    } finally {
+      if (!result) {
+        close();
+      }
+    }
+  }
+
+  @Override
+  public T next() {
+    try {
+      return request.readNextBean();
+    } catch (SQLException e) {
+      throw new PersistenceException(e);
+    }
+  }
+
+  @Override
+  public void close() {
+    if (!closed) {
+      closed = true;
+      request.close();
+      request.endTransIfRequired();
+    }
+  }
+
+  @Override
+  public void remove() {
+    throw new IllegalStateException("remove() not allowed");
+  }
+}

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

@@ -2,6 +2,7 @@ package io.ebeaninternal.server.querydefn;
 
 import io.ebean.DtoQuery;
 import io.ebean.ProfileLocation;
+import io.ebean.QueryIterator;
 import io.ebean.Transaction;
 import io.ebeaninternal.api.BindParams;
 import io.ebeaninternal.api.SpiDtoQuery;
@@ -15,6 +16,7 @@ import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
+import java.util.stream.Stream;
 
 /**
  * Default implementation of DtoQuery.
@@ -112,6 +114,16 @@ public class DefaultDtoQuery<T> implements SpiDtoQuery<T> {
     server.findDtoEachWhile(this, consumer);
   }
 
+  @Override
+  public QueryIterator<T> findIterate() {
+    return server.findDtoIterate(this);
+  }
+
+  @Override
+  public Stream<T> findStream() {
+    return server.findDtoStream(this);
+  }
+
   @Override
   public List<T> findList() {
     return server.findDtoList(this);