Fix for Issue 35 Connection pool leak with findFutureRowCount(),

findFutureList() and findFutureIds()

src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java

@@ -1340,10 +1340,9 @@ public final class DefaultServer implements SpiEbeanServer {
     Transaction newTxn = createTransaction();
 
     CallableQueryRowCount<T> call = new CallableQueryRowCount<T>(this, copy, newTxn);
-    FutureTask<Integer> futureTask = new FutureTask<Integer>(call);
 
-    QueryFutureRowCount<T> queryFuture = new QueryFutureRowCount<T>(copy, futureTask);
-    backgroundExecutor.execute(futureTask);
+    QueryFutureRowCount<T> queryFuture = new QueryFutureRowCount<T>(call);
+    backgroundExecutor.execute(queryFuture.getFutureTask());
 
     return queryFuture;
   }
@@ -1362,11 +1361,9 @@ public final class DefaultServer implements SpiEbeanServer {
     Transaction newTxn = createTransaction();
 
     CallableQueryIds<T> call = new CallableQueryIds<T>(this, copy, newTxn);
-    FutureTask<List<Object>> futureTask = new FutureTask<List<Object>>(call);
+    QueryFutureIds<T> queryFuture = new QueryFutureIds<T>(call);
 
-    QueryFutureIds<T> queryFuture = new QueryFutureIds<T>(copy, futureTask);
-
-    backgroundExecutor.execute(futureTask);
+    backgroundExecutor.execute(queryFuture.getFutureTask());
 
     return queryFuture;
   }
@@ -1376,6 +1373,7 @@ public final class DefaultServer implements SpiEbeanServer {
     SpiQuery<T> spiQuery = (SpiQuery<T>) query;
     spiQuery.setFutureFetch(true);
 
+    // transfer the persistence content from the transaction
     if (spiQuery.getPersistenceContext() == null) {
       if (t != null) {
         spiQuery.setPersistenceContext(((SpiTransaction) t).getPersistenceContext());
@@ -1387,14 +1385,13 @@ public final class DefaultServer implements SpiEbeanServer {
       }
     }
 
+    // Create a new transaction solely to execute the findList() at some future time
     Transaction newTxn = createTransaction();
-    CallableQueryList<T> call = new CallableQueryList<T>(this, query, newTxn);
+    CallableQueryList<T> call = new CallableQueryList<T>(this, spiQuery, newTxn);
+    QueryFutureList<T> queryFuture = new QueryFutureList<T>(call);
+    backgroundExecutor.execute(queryFuture.getFutureTask());
 
-    FutureTask<List<T>> futureTask = new FutureTask<List<T>>(call);
-
-    backgroundExecutor.execute(futureTask);
-
-    return new QueryFutureList<T>(query, futureTask);
+    return queryFuture;
   }
 
   public <T> PagingList<T> findPagingList(Query<T> query, Transaction t, int pageSize) {

src/main/java/com/avaje/ebeaninternal/server/query/BaseFuture.java

@@ -15,7 +15,7 @@ import java.util.concurrent.TimeoutException;
  */
 public abstract class BaseFuture<T> implements Future<T> {
 	
-	private final FutureTask<T> futureTask;
+	protected final FutureTask<T> futureTask;
 	
 	public BaseFuture(FutureTask<T> futureTask) {
 		this.futureTask = futureTask;
@@ -43,5 +43,4 @@ public abstract class BaseFuture<T> implements Future<T> {
 		return futureTask.isDone();
 	}
 
-	
 }

src/main/java/com/avaje/ebeaninternal/server/query/CallableQuery.java

@@ -1,8 +1,8 @@
 package com.avaje.ebeaninternal.server.query;
 
-import com.avaje.ebean.Query;
 import com.avaje.ebean.Transaction;
 import com.avaje.ebeaninternal.api.SpiEbeanServer;
+import com.avaje.ebeaninternal.api.SpiQuery;
 
 /**
  * Base object for making query execution into Callable's.
@@ -13,16 +13,24 @@ import com.avaje.ebeaninternal.api.SpiEbeanServer;
  */
 public abstract class CallableQuery<T> {
 
-	protected final Query<T> query;
+	protected final SpiQuery<T> query;
 	
 	protected final SpiEbeanServer server;
 	
-	protected final Transaction t;
+	protected final Transaction transaction;
 	
-	public CallableQuery(SpiEbeanServer server, Query<T> query, Transaction t) {
+	public CallableQuery(SpiEbeanServer server, SpiQuery<T> query, Transaction t) {
 		this.server = server;
 		this.query = query;
-		this.t = t;
+		this.transaction = t;
 	}
+
+  public SpiQuery<T> getQuery() {
+    return query;
+  }
+
+  public Transaction getTransaction() {
+    return transaction;
+  }
 	
 }

src/main/java/com/avaje/ebeaninternal/server/query/CallableQueryIds.java

@@ -3,9 +3,9 @@ package com.avaje.ebeaninternal.server.query;
 import java.util.List;
 import java.util.concurrent.Callable;
 
-import com.avaje.ebean.Query;
 import com.avaje.ebean.Transaction;
 import com.avaje.ebeaninternal.api.SpiEbeanServer;
+import com.avaje.ebeaninternal.api.SpiQuery;
 
 /**
  * Represent the fetch Id's query as a Callable.
@@ -15,7 +15,7 @@ import com.avaje.ebeaninternal.api.SpiEbeanServer;
 public class CallableQueryIds<T> extends CallableQuery<T> implements Callable<List<Object>> {
 
 	
-	public CallableQueryIds(SpiEbeanServer server, Query<T> query, Transaction t) {
+	public CallableQueryIds(SpiEbeanServer server, SpiQuery<T> query, Transaction t) {
 		super(server, query, t);
 	}
 	
@@ -26,7 +26,11 @@ public class CallableQueryIds<T> extends CallableQuery<T> implements Callable<Li
 		// we have already made a copy of the query
 		// this way the same query instance is available to the
 		// QueryFutureIds (as so has access to the List before it is done)
-		return server.findIdsWithCopy(query, t);
+	  try {
+	    return server.findIdsWithCopy(query, transaction);
+	  } finally {
+	    transaction.end();
+	  }
 	}
 
 }

src/main/java/com/avaje/ebeaninternal/server/query/CallableQueryList.java

@@ -3,9 +3,9 @@ package com.avaje.ebeaninternal.server.query;
 import java.util.List;
 import java.util.concurrent.Callable;
 
-import com.avaje.ebean.Query;
 import com.avaje.ebean.Transaction;
 import com.avaje.ebeaninternal.api.SpiEbeanServer;
+import com.avaje.ebeaninternal.api.SpiQuery;
 
 /**
  * Represent the findList query as a Callable.
@@ -15,7 +15,7 @@ import com.avaje.ebeaninternal.api.SpiEbeanServer;
 public class CallableQueryList<T> extends CallableQuery<T> implements Callable<List<T>> {
 
 	
-	public CallableQueryList(SpiEbeanServer server, Query<T> query, Transaction t) {
+	public CallableQueryList(SpiEbeanServer server, SpiQuery<T> query, Transaction t) {
 		super(server, query, t);
 	}
 	
@@ -23,7 +23,12 @@ public class CallableQueryList<T> extends CallableQuery<T> implements Callable<L
 	 * Execute the query returning the resulting List.
 	 */
 	public List<T> call() throws Exception {
-		return server.findList(query, t);
+	  try {
+	    return server.findList(query, transaction);
+	  } finally {
+	    // cleanup the underlying connection
+	    transaction.end();
+	  }
 	}
 
 	

src/main/java/com/avaje/ebeaninternal/server/query/CallableQueryRowCount.java

@@ -2,29 +2,36 @@ package com.avaje.ebeaninternal.server.query;
 
 import java.util.concurrent.Callable;
 
-import com.avaje.ebean.Query;
 import com.avaje.ebean.Transaction;
 import com.avaje.ebeaninternal.api.SpiEbeanServer;
+import com.avaje.ebeaninternal.api.SpiQuery;
 
 /**
  * Represent the findRowCount query as a Callable.
- *
- * @param <T> the entity bean type
+ * 
+ * @param <T>
+ *          the entity bean type
  */
 public class CallableQueryRowCount<T> extends CallableQuery<T> implements Callable<Integer> {
 
-	
-	public CallableQueryRowCount(SpiEbeanServer server, Query<T> query, Transaction t) {
-		super(server, query, t);
-	}
-	
-	/**
-	 * Execute the query returning the row count.
-	 */
-	public Integer call() throws Exception {
-		return server.findRowCountWithCopy(query, t);
-	}
+  /**
+   * Note that the transaction passed in is always a new transaction solely to
+   * find the row count so it must be cleaned up by this CallableQueryRowCount.
+   */
+  public CallableQueryRowCount(SpiEbeanServer server, SpiQuery<T> query, Transaction t) {
+    super(server, query, t);
+  }
+
+  /**
+   * Execute the query returning the row count.
+   */
+  public Integer call() throws Exception {
+    try {
+      return server.findRowCountWithCopy(query, transaction);
+    } finally {
+      // cleanup the underlying connection
+      transaction.end();
+    }
+  }
 
-	
-	
 }

src/main/java/com/avaje/ebeaninternal/server/query/CallableSqlQueryList.java

@@ -17,19 +17,23 @@ public class CallableSqlQueryList implements Callable<List<SqlRow>> {
 	
 	private final EbeanServer server;
 	
-	private final Transaction t;
+	private final Transaction transaction;
 	
 	public CallableSqlQueryList(EbeanServer server, SqlQuery query, Transaction t) {
 		this.server = server;
 		this.query = query;
-		this.t = t;
+		this.transaction = t;
 	}
 
 	/**
 	 * Execute the query returning the resulting list.
 	 */
 	public List<SqlRow> call() throws Exception {
-		return server.findList(query, t);
+	  try {
+		return server.findList(query, transaction);
+	  } finally {
+	    transaction.end();
+	  }
 	}
 
 	

src/main/java/com/avaje/ebeaninternal/server/query/QueryFutureIds.java

@@ -5,30 +5,38 @@ import java.util.concurrent.FutureTask;
 
 import com.avaje.ebean.FutureIds;
 import com.avaje.ebean.Query;
-import com.avaje.ebeaninternal.api.SpiQuery;
+import com.avaje.ebean.Transaction;
 
 /**
  * Default implementation of FutureIds.
  */
 public class QueryFutureIds<T> extends BaseFuture<List<Object>> implements FutureIds<T> {
 
-	private final SpiQuery<T> query;
+	private final CallableQueryIds<T> call;
 	
-	public QueryFutureIds(SpiQuery<T> query, FutureTask<List<Object>> futureTask) {
-		super(futureTask);
-		this.query = query;
+	public QueryFutureIds(CallableQueryIds<T> call ) {
+		super(new FutureTask<List<Object>>(call));
+		this.call = call;
+	}
+	
+	public FutureTask<List<Object>> getFutureTask() {
+	  return futureTask;
+	}
+	
+	public Transaction getTransaction() {
+	  return call.transaction;
 	}
 	
 	public Query<T> getQuery() {
-		return query;
+		return call.query;
 	}
 	
 	public List<Object> getPartialIds() {
-		return query.getIdList();
+		return call.query.getIdList();
 	}
 
 	public boolean cancel(boolean mayInterruptIfRunning) {
-		query.cancel();
+	  call.query.cancel();
 		return super.cancel(mayInterruptIfRunning);
 	}
 

src/main/java/com/avaje/ebeaninternal/server/query/QueryFutureList.java

@@ -5,26 +5,34 @@ import java.util.concurrent.FutureTask;
 
 import com.avaje.ebean.FutureList;
 import com.avaje.ebean.Query;
+import com.avaje.ebean.Transaction;
 
 /**
  * Default implementation for FutureList.
  */
 public class QueryFutureList<T> extends BaseFuture<List<T>> implements FutureList<T> {
 
-	private final Query<T> query;
+	private final CallableQueryList<T> call;
 	
+	public QueryFutureList(CallableQueryList<T> call) {
+		super(new FutureTask<List<T>>(call));
+		this.call = call;
+	}
 	
-	public QueryFutureList(Query<T> query, FutureTask<List<T>> futureTask) {
-		super(futureTask);
-		this.query = query;
+	public FutureTask<List<T>> getFutureTask() {
+	  return futureTask;
+	}
+	
+	public Transaction getTransaction() {
+	  return call.transaction;
 	}
 	
 	public Query<T> getQuery() {
-		return query;
+		return call.query;
 	}
 
 	public boolean cancel(boolean mayInterruptIfRunning) {
-		query.cancel();
+	  call.query.cancel();
 		return super.cancel(mayInterruptIfRunning);
 	}
 

src/main/java/com/avaje/ebeaninternal/server/query/QueryFutureRowCount.java

@@ -4,25 +4,34 @@ import java.util.concurrent.FutureTask;
 
 import com.avaje.ebean.FutureRowCount;
 import com.avaje.ebean.Query;
+import com.avaje.ebean.Transaction;
 
 /**
  * Future implementation for the row count query.
  */
 public class QueryFutureRowCount<T> extends BaseFuture<Integer> implements FutureRowCount<T> {
 
-	private final Query<T> query;
+  private final CallableQueryRowCount<T> call;
 	
-	public QueryFutureRowCount(Query<T> query, FutureTask<Integer> futureTask) {
-		super(futureTask);
-		this.query = query;
+	public QueryFutureRowCount(CallableQueryRowCount<T> call ) {
+		super(new FutureTask<Integer>(call));
+		this.call = call;
+	}
+	
+	public FutureTask<Integer> getFutureTask() {
+	  return futureTask;
+	}
+	
+	public Transaction getTransaction() {
+	  return call.transaction; 
 	}
 	
 	public Query<T> getQuery() {
-		return query;
+		return call.query;
 	}
 
 	public boolean cancel(boolean mayInterruptIfRunning) {
-		query.cancel();
+	  call.query.cancel();
 		return super.cancel(mayInterruptIfRunning);
 	}