From 8a84eaf7a74379f34a8b47c007506949080bac45 Mon Sep 17 00:00:00 2001 From: Robin Bygrave Date: Thu, 28 Apr 2016 22:15:41 +1200 Subject: [PATCH] #675 - ENH: Add SqlQuery.findEach() and SqlQuery.findEachWhile() methods ... to replace setListener() --- .../java/com/avaje/ebean/EbeanServer.java | 19 + src/main/java/com/avaje/ebean/SqlQuery.java | 19 + .../server/core/DefaultServer.java | 24 ++ .../server/core/RelationalQueryEngine.java | 22 +- .../server/core/RelationalQueryRequest.java | 364 ++++++++++++++---- .../query/DefaultRelationalQueryEngine.java | 207 +++------- .../querydefn/DefaultRelationalQuery.java | 12 + .../ebeaninternal/api/TDSpiEbeanServer.java | 8 + .../tests/query/sqlquery/SqlQueryTests.java | 56 ++- 9 files changed, 500 insertions(+), 231 deletions(-) diff --git a/src/main/java/com/avaje/ebean/EbeanServer.java b/src/main/java/com/avaje/ebean/EbeanServer.java index 6d995c955..7bdb5975f 100644 --- a/src/main/java/com/avaje/ebean/EbeanServer.java +++ b/src/main/java/com/avaje/ebean/EbeanServer.java @@ -1192,6 +1192,25 @@ public interface EbeanServer { */ List findList(SqlQuery query, Transaction transaction); + /** + * Execute the SqlQuery iterating a row at a time. + *

+ * This streaming type query is useful for large query execution as only 1 row needs to be held in memory. + *

+ */ + void findEach(SqlQuery query, QueryEachConsumer consumer, Transaction transaction); + + /** + * Execute the SqlQuery iterating a row at a time with the ability to stop consuming part way through. + *

+ * Returning false after processing a row stops the iteration through the query results. + *

+ *

+ * This streaming type query is useful for large query execution as only 1 row needs to be held in memory. + *

+ */ + void findEachWhile(SqlQuery query, QueryEachWhileConsumer consumer, Transaction transaction); + /** * Execute the sql query returning a set of MapBean. *

diff --git a/src/main/java/com/avaje/ebean/SqlQuery.java b/src/main/java/com/avaje/ebean/SqlQuery.java index 4852e575b..c36ac942f 100644 --- a/src/main/java/com/avaje/ebean/SqlQuery.java +++ b/src/main/java/com/avaje/ebean/SqlQuery.java @@ -61,6 +61,25 @@ public interface SqlQuery extends Serializable { */ Map findMap(); + /** + * Execute the SqlQuery iterating a row at a time. + *

+ * This streaming type query is useful for large query execution as only 1 row needs to be held in memory. + *

+ */ + void findEach(QueryEachConsumer consumer); + + /** + * Execute the SqlQuery iterating a row at a time with the ability to stop consuming part way through. + *

+ * Returning false after processing a row stops the iteration through the query results. + *

+ *

+ * This streaming type query is useful for large query execution as only 1 row needs to be held in memory. + *

+ */ + void findEachWhile(QueryEachWhileConsumer consumer); + /** * Execute the query returning a single row or null. *

diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java index 60a550cdd..f67e6c64b 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java @@ -1404,6 +1404,30 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { return new SqlQueryFutureList(query, futureTask); } + @Override + public void findEach(SqlQuery query, QueryEachConsumer consumer, Transaction transaction) { + + RelationalQueryRequest request = new RelationalQueryRequest(this, relationalQueryEngine, query, transaction); + try { + request.initTransIfRequired(); + request.findEach(consumer); + } finally { + request.endTransIfRequired(); + } + } + + @Override + public void findEachWhile(SqlQuery query, QueryEachWhileConsumer consumer, Transaction transaction) { + + RelationalQueryRequest request = new RelationalQueryRequest(this, relationalQueryEngine, query, transaction); + try { + request.initTransIfRequired(); + request.findEachWhile(consumer); + } finally { + request.endTransIfRequired(); + } + } + public List findList(SqlQuery query, Transaction t) { RelationalQueryRequest request = new RelationalQueryRequest(this, relationalQueryEngine, query, t); diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryEngine.java b/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryEngine.java index 999b0fdf4..b46f39dbf 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryEngine.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryEngine.java @@ -1,11 +1,25 @@ package com.avaje.ebeaninternal.server.core; +import com.avaje.ebean.QueryEachConsumer; +import com.avaje.ebean.QueryEachWhileConsumer; +import com.avaje.ebean.SqlRow; + public interface RelationalQueryEngine { - /** - * Find a list of beans using relational query. - */ - Object findMany(RelationalQueryRequest request); + /** + * Find a list of beans using relational query. + */ + Object findMany(RelationalQueryRequest request); + + /** + * Find each query using relational query. + */ + void findEach(RelationalQueryRequest request, QueryEachConsumer consumer); + + /** + * Find each while query using relational query. + */ + void findEach(RelationalQueryRequest request, QueryEachWhileConsumer consumer); } \ No newline at end of file diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryRequest.java b/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryRequest.java index 0f3f3fb07..b8f15d377 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryRequest.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/RelationalQueryRequest.java @@ -1,114 +1,314 @@ package com.avaje.ebeaninternal.server.core; -import java.util.List; -import java.util.Map; -import java.util.Set; - import com.avaje.ebean.EbeanServer; +import com.avaje.ebean.QueryEachConsumer; +import com.avaje.ebean.QueryEachWhileConsumer; import com.avaje.ebean.SqlQuery; import com.avaje.ebean.SqlRow; import com.avaje.ebean.Transaction; +import com.avaje.ebeaninternal.api.BindParams; import com.avaje.ebeaninternal.api.SpiEbeanServer; import com.avaje.ebeaninternal.api.SpiQuery; import com.avaje.ebeaninternal.api.SpiSqlQuery; import com.avaje.ebeaninternal.api.SpiTransaction; +import com.avaje.ebeaninternal.server.lib.util.Str; +import com.avaje.ebeaninternal.server.persist.Binder; +import com.avaje.ebeaninternal.server.query.DefaultSqlRow; +import com.avaje.ebeaninternal.server.transaction.TransactionManager; +import com.avaje.ebeaninternal.server.util.BindParamsParser; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; /** * Wraps the objects involved in executing a SqlQuery. */ public final class RelationalQueryRequest { - private final SpiSqlQuery query; + private static final int GLOBAL_ROW_LIMIT = Integer.valueOf(System.getProperty("ebean.query.globallimit", "1000000")); - private final RelationalQueryEngine queryEngine; + private static final Logger logger = LoggerFactory.getLogger(RelationalQueryRequest.class); - private final SpiEbeanServer ebeanServer; + private final SpiSqlQuery query; - private SpiTransaction trans; + private final RelationalQueryEngine queryEngine; - private boolean createdTransaction; + private final SpiEbeanServer ebeanServer; - private SpiQuery.Type queryType; + private SpiTransaction trans; - /** - * Create the BeanFindRequest. - */ - public RelationalQueryRequest(SpiEbeanServer server, RelationalQueryEngine engine, SqlQuery q, Transaction t) { - this.ebeanServer = server; - this.queryEngine = engine; - this.query = (SpiSqlQuery) q; - this.trans = (SpiTransaction) t; + private boolean createdTransaction; + + private SpiQuery.Type queryType; + + private String sql; + + private ResultSet resultSet; + + private int rowCount; + + private String bindLog = ""; + + private String[] propertyNames; + + private int estimateCapacity; + + private PreparedStatement pstmt; + + /** + * Create the BeanFindRequest. + */ + public RelationalQueryRequest(SpiEbeanServer server, RelationalQueryEngine engine, SqlQuery q, Transaction t) { + this.ebeanServer = server; + this.queryEngine = engine; + this.query = (SpiSqlQuery) q; + this.trans = (SpiTransaction) t; + } + + /** + * Create a transaction if none currently exists. + */ + public void initTransIfRequired() { + if (trans == null) { + trans = ebeanServer.getCurrentServerTransaction(); + if (trans == null || !trans.isActive()) { + // create a local readOnly transaction + trans = ebeanServer.createServerTransaction(false, -1); + createdTransaction = true; + } } + } - /** - * Create a transaction if none currently exists. - */ - public void initTransIfRequired() { - if (trans == null) { - trans = ebeanServer.getCurrentServerTransaction(); - if (trans == null || !trans.isActive()) { - // create a local readOnly transaction - trans = ebeanServer.createServerTransaction(false, -1); - createdTransaction = true; - } + /** + * End the transaction if it was locally created. + */ + public void endTransIfRequired() { + if (createdTransaction) { + trans.commit(); + } + } + + public void findEach(QueryEachConsumer consumer) { + queryType = SpiQuery.Type.ITERATE; + queryEngine.findEach(this, consumer); + } + + public void findEachWhile(QueryEachWhileConsumer consumer) { + queryType = SpiQuery.Type.ITERATE; + queryEngine.findEach(this, consumer); + } + + @SuppressWarnings("unchecked") + public List findList() { + queryType = SpiQuery.Type.LIST; + return (List) queryEngine.findMany(this); + } + + @SuppressWarnings("unchecked") + public Set findSet() { + queryType = SpiQuery.Type.SET; + return (Set) queryEngine.findMany(this); + } + + @SuppressWarnings("unchecked") + public Map findMap() { + queryType = SpiQuery.Type.MAP; + return (Map) queryEngine.findMany(this); + } + + /** + * Return the find that is to be performed. + */ + public SpiSqlQuery getQuery() { + return query; + } + + /** + * Return the type (List, Set or Map) that this fetch returns. + */ + public SpiQuery.Type getQueryType() { + return queryType; + } + + public EbeanServer getEbeanServer() { + return ebeanServer; + } + + public SpiTransaction getTransaction() { + return trans; + } + + public boolean isLogSql() { + return trans.isLogSql(); + } + + public boolean isLogSummary() { + return trans.isLogSummary(); + } + + private void setResultSet(ResultSet resultSet) throws SQLException { + this.resultSet = resultSet; + this.propertyNames = getPropertyNames(); + // calculate the initialCapacity of the Map to reduce rehashing + float initCap = (propertyNames.length) / 0.7f; + this.estimateCapacity = (int) initCap + 1; + } + + /** + * Build the list of property names. + */ + private String[] getPropertyNames() throws SQLException { + + ArrayList propNames = new ArrayList(); + ResultSetMetaData metaData = resultSet.getMetaData(); + + int columnsPlusOne = metaData.getColumnCount() + 1; + for (int i = 1; i < columnsPlusOne; i++) { + propNames.add(metaData.getColumnLabel(i)); + } + return propNames.toArray(new String[propNames.size()]); + } + + /** + * Return the bindLog for this request. + */ + public String getBindLog() { + return bindLog; + } + + /** + * Return true if we can navigate to the next row. + */ + public boolean next() throws SQLException { + rowCount++; + return resultSet.next(); + } + + /** + * Close the underlying resources. + */ + public void close() { + try { + if (resultSet != null) { + resultSet.close(); + } + } catch (SQLException e) { + logger.error(null, e); + } + try { + if (pstmt != null) { + pstmt.close(); + } + } catch (SQLException e) { + logger.error(null, e); + } + } + + /** + * Read and return the next SqlRow. + */ + public SqlRow createNewRow(String dbTrueValue) throws SQLException { + + SqlRow sqlRow = new DefaultSqlRow(estimateCapacity, 0.75f, dbTrueValue); + + int index = 0; + for (int i = 0; i < propertyNames.length; i++) { + index++; + Object value = resultSet.getObject(index); + sqlRow.set(propertyNames[i], value); + } + return sqlRow; + } + + /** + * Prepare the SQL taking into account named bind parameters. + */ + private void prepareSql() { + + String sql = query.getQuery(); + BindParams bindParams = query.getBindParams(); + if (!bindParams.isEmpty()) { + // convert any named parameters if required + sql = BindParamsParser.parse(bindParams, sql); + } + this.sql = sql; + } + + /** + * Prepare and execute the SQL using the Binder. + */ + public boolean executeSql(Binder binder) throws SQLException { + + prepareSql(); + synchronized (query) { + if (query.isCancelled()) { + logger.trace("Query already cancelled"); + return false; + } + + Connection conn = trans.getInternalConnection(); + + // synchronise for query.cancel() support + pstmt = conn.prepareStatement(sql); + if (query.getTimeout() > 0) { + pstmt.setQueryTimeout(query.getTimeout()); + } + if (query.getBufferFetchSizeHint() > 0) { + pstmt.setFetchSize(query.getBufferFetchSizeHint()); + } + + BindParams bindParams = query.getBindParams(); + if (!bindParams.isEmpty()) { + this.bindLog = binder.bind(bindParams, pstmt); + } + + if (isLogSql()) { + String logSql = sql; + if (TransactionManager.SQL_LOGGER.isTraceEnabled()) { + logSql = Str.add(logSql, "; --bind(", bindLog, ")"); } - } + trans.logSql(logSql); + } - /** - * End the transaction if it was locally created. - */ - public void endTransIfRequired() { - if (createdTransaction) { - trans.commit(); - } + setResultSet(pstmt.executeQuery()); + return true; } + } - @SuppressWarnings("unchecked") - public List findList() { - queryType = SpiQuery.Type.LIST; - return (List) queryEngine.findMany(this); - } + /** + * Return true if the query has been cancelled. + */ + public boolean isCancelled() { + return query.isCancelled(); + } - @SuppressWarnings("unchecked") - public Set findSet() { - queryType = SpiQuery.Type.SET; - return (Set) queryEngine.findMany(this); - } + /** + * Return the maxRows allowed to fetch. + */ + public int getMaxRows() { + return query.getMaxRows() >= 1 ? query.getMaxRows() : GLOBAL_ROW_LIMIT; + } - @SuppressWarnings("unchecked") - public Map findMap() { - queryType = SpiQuery.Type.MAP; - return (Map) queryEngine.findMany(this); - } - - /** - * Return the find that is to be performed. - */ - public SpiSqlQuery getQuery() { - return query; - } - - /** - * Return the type (List, Set or Map) that this fetch returns. - */ - public SpiQuery.Type getQueryType() { - return queryType; - } - - public EbeanServer getEbeanServer() { - return ebeanServer; - } - - public SpiTransaction getTransaction() { - return trans; - } - - public boolean isLogSql() { - return trans.isLogSql(); - } - - public boolean isLogSummary() { - return trans.isLogSummary(); - } + /** + * Return the SQL executed for this query. + */ + public String getSql() { + return sql; + } + /** + * Return the rows read. + */ + public int getRowCount() { + return rowCount - 1; + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/query/DefaultRelationalQueryEngine.java b/src/main/java/com/avaje/ebeaninternal/server/query/DefaultRelationalQueryEngine.java index 733f35f6b..95fd32b83 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/query/DefaultRelationalQueryEngine.java +++ b/src/main/java/com/avaje/ebeaninternal/server/query/DefaultRelationalQueryEngine.java @@ -1,25 +1,19 @@ package com.avaje.ebeaninternal.server.query; +import com.avaje.ebean.QueryEachConsumer; +import com.avaje.ebean.QueryEachWhileConsumer; import com.avaje.ebean.SqlQueryListener; import com.avaje.ebean.SqlRow; -import com.avaje.ebean.bean.BeanCollection; -import com.avaje.ebeaninternal.api.BindParams; import com.avaje.ebeaninternal.api.SpiSqlQuery; -import com.avaje.ebeaninternal.api.SpiTransaction; import com.avaje.ebeaninternal.server.core.Message; import com.avaje.ebeaninternal.server.core.RelationalQueryEngine; import com.avaje.ebeaninternal.server.core.RelationalQueryRequest; -import com.avaje.ebeaninternal.server.lib.util.Str; import com.avaje.ebeaninternal.server.persist.Binder; -import com.avaje.ebeaninternal.server.transaction.TransactionManager; -import com.avaje.ebeaninternal.server.type.DataBind; -import com.avaje.ebeaninternal.server.util.BindParamsParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.persistence.PersistenceException; -import java.sql.*; -import java.util.ArrayList; +import java.sql.SQLException; /** * Perform native sql fetches. @@ -28,7 +22,6 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine { private static final Logger logger = LoggerFactory.getLogger(DefaultRelationalQueryEngine.class); - private static final int GLOBAL_ROW_LIMIT = Integer.valueOf(System.getProperty("ebean.query.globallimit", "1000000")); private final Binder binder; @@ -39,108 +32,85 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine { this.dbTrueValue = dbTrueValue == null ? "true" : dbTrueValue; } - public Object findMany(RelationalQueryRequest request) { + @Override + public void findEach(RelationalQueryRequest request, QueryEachWhileConsumer consumer) { - SpiSqlQuery query = request.getQuery(); + long startTime = System.currentTimeMillis(); + try { + request.executeSql(binder); + while (request.next()) { + if (!consumer.accept(readRow(request))) { + break; + } + } + logSummary(request, startTime); + + } catch (Exception e) { + throw new PersistenceException(Message.msg("fetch.error", e.getMessage(), request.getSql()), e); + + } finally { + request.close(); + } + } + + @Override + public void findEach(RelationalQueryRequest request, QueryEachConsumer consumer) { long startTime = System.currentTimeMillis(); - SpiTransaction t = request.getTransaction(); - Connection conn = t.getInternalConnection(); - ResultSet rset = null; - PreparedStatement pstmt = null; - - String sql = query.getQuery(); - - BindParams bindParams = query.getBindParams(); - - if (!bindParams.isEmpty()) { - // convert any named parameters if required - sql = BindParamsParser.parse(bindParams, sql); - } - try { + request.executeSql(binder); + while (request.next()) { + consumer.accept(readRow(request)); + } + logSummary(request, startTime); - String bindLog = ""; - String[] propNames; + } catch (Exception e) { + throw new PersistenceException(Message.msg("fetch.error", e.getMessage(), request.getSql()), e); - synchronized (query) { - if (query.isCancelled()) { - logger.trace("Query already cancelled"); - return null; - } + } finally { + request.close(); + } + } - // synchronise for query.cancel() support - pstmt = conn.prepareStatement(sql); + public Object findMany(RelationalQueryRequest request) { - if (query.getTimeout() > 0) { - pstmt.setQueryTimeout(query.getTimeout()); - } - if (query.getBufferFetchSizeHint() > 0) { - pstmt.setFetchSize(query.getBufferFetchSizeHint()); - } - - if (!bindParams.isEmpty()) { - bindLog = binder.bind(bindParams, pstmt); - } - - if (request.isLogSql()) { - String logSql = sql; - if (TransactionManager.SQL_LOGGER.isTraceEnabled()) { - logSql = Str.add(logSql, "; --bind(", bindLog, ")"); - } - t.logSql(logSql); - } - - rset = pstmt.executeQuery(); - - propNames = getPropertyNames(rset); + long startTime = System.currentTimeMillis(); + try { + if (!request.executeSql(binder)) { + return null; } - // calculate the initialCapacity of the Map to reduce - // rehashing for queries with 12+ columns - float initCap = (propNames.length) / 0.7f; - int estimateCapacity = (int) initCap + 1; - - // determine the maxRows limit - int maxRows = GLOBAL_ROW_LIMIT; - if (query.getMaxRows() >= 1) { - maxRows = query.getMaxRows(); - } + int maxRows = request.getMaxRows(); int loadRowCount = 0; - SqlQueryListener listener = query.getListener(); - BeanCollectionWrapper wrapper = new BeanCollectionWrapper(request); boolean isMap = wrapper.isMap(); - String mapKey = query.getMapKey(); + String mapKey = request.getQuery().getMapKey(); - SqlRow bean = null; + SpiSqlQuery query = request.getQuery(); + SqlQueryListener listener = query.getListener(); - while (rset.next()) { + while (request.next()) { + SqlRow bean; synchronized (query) { - // synchronise for query.cancel() support - if (!query.isCancelled()) { - bean = readRow(rset, propNames, estimateCapacity); + if (request.isCancelled()) { + break; } + bean = readRow(request); } if (bean != null) { - // bean can be null if query cancelled if (listener != null) { listener.process(bean); - } else { if (isMap) { - Object keyValue = bean.get(mapKey); - wrapper.addToMap(bean, keyValue); + wrapper.addToMap(bean, bean.get(mapKey)); } else { wrapper.addToCollection(bean); } } - loadRowCount++; - if (loadRowCount == maxRows) { // break, as we have hit the max rows to fetch... break; @@ -148,84 +118,35 @@ public class DefaultRelationalQueryEngine implements RelationalQueryEngine { } } - BeanCollection beanColl = wrapper.getBeanCollection(); + logSummary(request, startTime); - if (request.isLogSummary()) { - long exeTime = System.currentTimeMillis() - startTime; - String msg = "SqlQuery rows[" + loadRowCount + "] time[" + exeTime + "] bind[" + bindLog + "]"; - t.logSummary(msg); - } - - if (query.isCancelled()) { - logger.debug("Query was cancelled during execution rows:" + loadRowCount); - } - - return beanColl; + return wrapper.getBeanCollection(); } catch (Exception e) { - String m = Message.msg("fetch.error", e.getMessage(), sql); - throw new PersistenceException(m, e); + throw new PersistenceException(Message.msg("fetch.error", e.getMessage(), request.getSql()), e); } finally { - try { - if (rset != null) { - rset.close(); - } - } catch (SQLException e) { - logger.error(null, e); - } - try { - if (pstmt != null) { - pstmt.close(); - } - } catch (SQLException e) { - logger.error(null, e); - } + request.close(); } } - /** - * Build the list of property names. - */ - protected String[] getPropertyNames(ResultSet rset) throws SQLException { + private void logSummary(RelationalQueryRequest request, long startTime) { - ArrayList propNames = new ArrayList(); - - ResultSetMetaData rsmd = rset.getMetaData(); - - int columnsPlusOne = rsmd.getColumnCount() + 1; - - - for (int i = 1; i < columnsPlusOne; i++) { - String columnName = rsmd.getColumnLabel(i); - // will convert columnName to lower case - propNames.add(columnName); + if (request.isLogSummary()) { + long exeTime = System.currentTimeMillis() - startTime; + request.getTransaction().logSummary("SqlQuery rows[" + request.getRowCount() + "] time[" + exeTime + "] bind[" + request.getBindLog() + "]"); } - return propNames.toArray(new String[propNames.size()]); + if (request.isCancelled()) { + logger.debug("Query was cancelled during execution rows: {}", request.getRowCount()); + } } /** * Read the row from the ResultSet and return as a MapBean. */ - protected SqlRow readRow(ResultSet rset, String[] propNames, int initialCapacity) throws SQLException { - - // by default a map will rehash on the 12th entry - // it will be pretty common to have 12 or more entries so - // to reduce rehashing I am trying to estimate a good - // initial capacity for the MapBean to use. - SqlRow bean = new DefaultSqlRow(initialCapacity, 0.75f, dbTrueValue); - - int index = 0; - - for (int i = 0; i < propNames.length; i++) { - index++; - Object value = rset.getObject(index); - bean.set(propNames[i], value); - } - - return bean; - + protected SqlRow readRow(RelationalQueryRequest request) throws SQLException { + return request.createNewRow(dbTrueValue); } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/querydefn/DefaultRelationalQuery.java b/src/main/java/com/avaje/ebeaninternal/server/querydefn/DefaultRelationalQuery.java index bcff6ddb7..7be4e52f6 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/querydefn/DefaultRelationalQuery.java +++ b/src/main/java/com/avaje/ebeaninternal/server/querydefn/DefaultRelationalQuery.java @@ -9,6 +9,8 @@ import java.util.Set; import javax.persistence.PersistenceException; import com.avaje.ebean.EbeanServer; +import com.avaje.ebean.QueryEachConsumer; +import com.avaje.ebean.QueryEachWhileConsumer; import com.avaje.ebean.SqlFutureList; import com.avaje.ebean.SqlQueryListener; import com.avaje.ebean.SqlRow; @@ -73,6 +75,16 @@ public class DefaultRelationalQuery implements SpiSqlQuery { return this; } + @Override + public void findEach(QueryEachConsumer consumer) { + server.findEach(this, consumer, null); + } + + @Override + public void findEachWhile(QueryEachWhileConsumer consumer) { + server.findEachWhile(this, consumer, null); + } + public List findList() { return server.findList(this, null); } diff --git a/src/test/java/com/avaje/ebeaninternal/api/TDSpiEbeanServer.java b/src/test/java/com/avaje/ebeaninternal/api/TDSpiEbeanServer.java index 9d2e5c880..ea35c5885 100644 --- a/src/test/java/com/avaje/ebeaninternal/api/TDSpiEbeanServer.java +++ b/src/test/java/com/avaje/ebeaninternal/api/TDSpiEbeanServer.java @@ -540,6 +540,14 @@ public class TDSpiEbeanServer implements SpiEbeanServer { return null; } + @Override + public void findEach(SqlQuery query, QueryEachConsumer consumer, Transaction transaction) { + } + + @Override + public void findEachWhile(SqlQuery query, QueryEachWhileConsumer consumer, Transaction transaction) { + } + @Override public Set findSet(SqlQuery query, Transaction transaction) { return null; diff --git a/src/test/java/com/avaje/tests/query/sqlquery/SqlQueryTests.java b/src/test/java/com/avaje/tests/query/sqlquery/SqlQueryTests.java index fa70938a2..1775390de 100644 --- a/src/test/java/com/avaje/tests/query/sqlquery/SqlQueryTests.java +++ b/src/test/java/com/avaje/tests/query/sqlquery/SqlQueryTests.java @@ -1,29 +1,81 @@ package com.avaje.tests.query.sqlquery; import com.avaje.ebean.Ebean; +import com.avaje.ebean.QueryEachConsumer; +import com.avaje.ebean.QueryEachWhileConsumer; import com.avaje.ebean.SqlQuery; import com.avaje.ebean.SqlQueryListener; import com.avaje.ebean.SqlRow; +import com.avaje.tests.model.basic.Order; import com.avaje.tests.model.basic.ResetBasicData; import org.junit.Test; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; + public class SqlQueryTests { @Test public void setListener() { - ResetBasicData.reset(); + int expectedRows = Ebean.find(Order.class).findRowCount(); + + final AtomicInteger count = new AtomicInteger(); + SqlQuery sqlQuery = Ebean.createSqlQuery("select * from o_order"); sqlQuery.setListener(new SqlQueryListener() { @Override public void process(SqlRow bean) { System.out.println("process row "+bean); + count.incrementAndGet(); } }); // returns an empty list sqlQuery.findList(); - + assertEquals(expectedRows, count.get()); } + + @Test + public void findEach() { + + ResetBasicData.reset(); + + int expectedRows = Ebean.find(Order.class).findRowCount(); + + final AtomicInteger count = new AtomicInteger(); + + SqlQuery sqlQuery = Ebean.createSqlQuery("select * from o_order"); + sqlQuery.findEach(new QueryEachConsumer() { + @Override + public void accept(SqlRow bean) { + count.incrementAndGet(); + } + }); + + assertEquals(expectedRows, count.get()); + } + + @Test + public void findEachWhile() { + + ResetBasicData.reset(); + + final AtomicInteger count = new AtomicInteger(); + + SqlQuery sqlQuery = Ebean.createSqlQuery("select * from o_order order by id"); + sqlQuery.findEachWhile(new QueryEachWhileConsumer() { + @Override + public boolean accept(SqlRow bean) { + count.incrementAndGet(); + Integer id = bean.getInteger("id"); + return id.intValue() < 3; + } + }); + + assertEquals(3, count.get()); + } + }