Use concurrent lists for cleanup collections. (#501)

Allow cleanup collections to be accessed by the housekeeper while closing is in progress.

Author: kdubb

driver/src/main/java/com/impossibl/postgres/jdbc/PGDirectConnection.java

@@ -124,15 +124,16 @@
 import java.sql.Savepoint;
 import java.sql.Struct;
 import java.text.ParseException;
-import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledFuture;
@@ -178,11 +179,11 @@ public class PGDirectConnection extends BasicContext implements PGConnection {
   private static class Cleanup implements CleanupRunnable {
 
     ServerConnection serverConnection;
-    List<WeakReference<PGStatement>> statements;
+    Collection<WeakReference<PGStatement>> statements;
     StackTraceElement[] allocationStackTrace;
     String connectionInfo;
 
-    private Cleanup(ServerConnection serverConnection, List<WeakReference<PGStatement>> statements, String connectionInfo) {
+    private Cleanup(ServerConnection serverConnection, Collection<WeakReference<PGStatement>> statements, String connectionInfo) {
       this.serverConnection = serverConnection;
       this.statements = statements;
       this.allocationStackTrace = new Exception().getStackTrace();
@@ -217,7 +218,7 @@ public void run() {
   boolean autoCommit = true;
   private int networkTimeout;
   private SQLWarning warningChain;
-  private List<WeakReference<PGStatement>> activeStatements;
+  private Collection<WeakReference<PGStatement>> activeStatements;
   private Map<StatementCacheKey, StatementDescription> descriptionCache;
   private Map<StatementCacheKey, PreparedStatementDescription> preparedStatementCache;
   private int preparedStatementCacheThreshold;
@@ -234,7 +235,7 @@ public void run() {
 
     this.strict = getSetting(STRICT_MODE);
     this.networkTimeout = getSetting(DEFAULT_NETWORK_TIMEOUT);
-    this.activeStatements = new ArrayList<>();
+    this.activeStatements = new ConcurrentLinkedQueue<>();
     this.notificationListeners = new ConcurrentHashMap<>();
 
     final int descriptionCacheSize = getSetting(DESCRIPTION_CACHE_SIZE);
@@ -394,7 +395,7 @@ else if (s == statement) {
    *
    * @param statements Statements to close
    */
-  private static void closeStatements(List<WeakReference<PGStatement>> statements) {
+  private static void closeStatements(Collection<WeakReference<PGStatement>> statements) {
 
     for (WeakReference<PGStatement> statementRef : statements) {
 

driver/src/main/java/com/impossibl/postgres/jdbc/PGStatement.java

@@ -48,10 +48,11 @@
 import java.sql.SQLException;
 import java.sql.SQLWarning;
 import java.sql.Statement;
-import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 import static java.lang.Long.min;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -74,10 +75,10 @@ private static class Cleanup implements CleanupRunnable {
 
     PGDirectConnection connection;
     String name;
-    List<WeakReference<PGResultSet>> resultSets;
+    Collection<WeakReference<PGResultSet>> resultSets;
     StackTraceElement[] allocationStackTrace;
 
-    private Cleanup(PGDirectConnection connection, String name, List<WeakReference<PGResultSet>> resultSets) {
+    private Cleanup(PGDirectConnection connection, String name, Collection<WeakReference<PGResultSet>> resultSets) {
       this.connection = connection;
       this.name = name;
       this.resultSets = resultSets;
@@ -129,7 +130,7 @@ public void run() {
   Query query;
   List<ResultBatch> resultBatches;
   boolean autoClose;
-  List<WeakReference<PGResultSet>> activeResultSets;
+  Collection<WeakReference<PGResultSet>> activeResultSets;
   PGResultSet generatedKeysResultSet;
   SQLWarning warningChain;
   int queryTimeout;
@@ -148,7 +149,7 @@ public void run() {
     this.name = name;
     this.processEscapes = true;
     this.resultFields = resultFields;
-    this.activeResultSets = new ArrayList<>();
+    this.activeResultSets = new ConcurrentLinkedQueue<>();
     this.generatedKeysResultSet = null;
     this.fetchSize = connection.getDefaultFetchSize();
 
@@ -199,7 +200,7 @@ static void closeCursor(PGDirectConnection connection, String cursorName) throws
    * Closes the given list of result-sets
    *
    */
-  private static void closeResultSets(List<WeakReference<PGResultSet>> resultSets) {
+  private static void closeResultSets(Collection<WeakReference<PGResultSet>> resultSets) {
 
     for (WeakReference<PGResultSet> resultSetRef : resultSets) {