Fix bug in ConnectionPool._closeQuery

There was a defect which meant that if you closed a prepared
query it could end up trying to close the query on a connection
which was still in use.

Fixes #49

Author: jamesots

lib/src/connection_pool.dart

@@ -18,12 +18,18 @@ class ConnectionPool extends Object with _ConnectionHelpers implements Queriable
   final int _maxPacketSize;
   int _max;
 
-/*
+  /*
    * The pool maintains a queue of connection requests. When a connection completes, if there
    * is a connection in the queue then it is 'activated' - that is, the future returned
    * by _getConnection() completes.
    */
   final Queue<Completer<_Connection>> _pendingConnections;
+  /*
+   * If you need a particular connection, put an entry in _requestedConnections. As soon as
+   * that connection is free then the completer completes. _requestedConnections is
+   * checked before _pendingConnections.
+   */
+  final Map<_Connection, Queue<Completer>> _requestedConnections;
   final List<_Connection> _pool;
 
 /**
@@ -40,6 +46,7 @@ class ConnectionPool extends Object with _ConnectionHelpers implements Queriable
 //      bool useCompression: false,
       bool useSSL: false}) :
   _pendingConnections = new Queue<Completer<_Connection>>(),
+  _requestedConnections = new Map<_Connection, Queue<Completer>>(),
   _pool = new List<_Connection>(),
   _host = host,
   _port = port,
@@ -125,6 +132,14 @@ class ConnectionPool extends Object with _ConnectionHelpers implements Queriable
       return;
     }
 
+    if (_requestedConnections.containsKey(cnx) && _requestedConnections[cnx].length > 0) {
+      _log.finest("Reusing cnx#${cnx.number} for a requested operation");
+      var c = _requestedConnections[cnx].removeFirst();
+      cnx.use();
+      c.complete(cnx);
+      return;
+    }
+
     if (_pendingConnections.length > 0) {
       _log.finest("Reusing cnx#${cnx.number} for a queued operation");
       var c = _pendingConnections.removeFirst();
@@ -217,13 +232,17 @@ class ConnectionPool extends Object with _ConnectionHelpers implements Queriable
     }
   }
 
+  // Close a prepared query on all connections which have this query.
+  // This may take some time if it has to wait a long time for a
+  // connection to become free.
   _closeQuery(Query q, bool retain) async {
     _log.finest("Closing query: ${q.sql}");
     var thePool = new List<_Connection>();
     thePool.addAll(_pool); // prevent concurrent modification
     for (var cnx in thePool) {
       var preparedQuery = cnx.removePreparedQueryFromCache(q.sql);
       if (preparedQuery != null) {
+        _log.finest("Connection not ready");
         await _waitUntilReady(cnx);
         _log.finest("Connection ready - closing query: ${q.sql}");
         var handler = new _CloseStatementHandler(preparedQuery.statementHandlerId);
@@ -233,17 +252,22 @@ class ConnectionPool extends Object with _ConnectionHelpers implements Queriable
     }
   }
 
-/**
-   * The future returned by [_waitUntilReady] fires when all queued operations in the pool
-   * have completed, and the connection is free to be used again.
+  /**
+   * The future returned by [_waitUntilReady] fires when the connection is next available
+   * to be used.
    */
   Future<_Connection> _waitUntilReady(_Connection cnx) {
     var c = new Completer<_Connection>();
     if (!cnx.inUse) {
+      // connection isn't in use, so use it straight away
       cnx.use();
       c.complete(cnx);
     } else {
-      _pendingConnections.add(c);
+      // Connection is in use, so request we get it the next time it's available.
+      if (!_requestedConnections.containsKey(cnx)) {
+        _requestedConnections[cnx] = new Queue<Completer>();
+      }
+      _requestedConnections[cnx].add(c);
     }
     return c.future;
   }

lib/src/query.dart

@@ -83,7 +83,7 @@ class Query extends Object with _ConnectionHelpers {
   }
 
   /// Closes this query and removes it from all connections in the pool.
-  void close() {
+  close() async {
     _pool._closeQuery(this, _inTransaction);
   }
 

test/integration/prepared_query.dart

@@ -0,0 +1,26 @@
+part of integrationtests;
+
+void runPreparedQueryTests(String user, String password, String db, int port, String host) {
+  group('prepared query', ()
+  {
+    ConnectionPool pool;
+
+    setUp(() {
+      pool = new ConnectionPool(user:user, password:password, db:db, port:port, host:host, max:2);
+      return setup(pool, "row",
+        "create table row (id integer, name text)",
+        "insert into row values (0, 'One'), (1, 'One')");
+    });
+
+    tearDown(() {
+      pool.closeConnectionsNow();
+    });
+
+    test('can close prepared query on in-use connections', () async {
+      var cnx = await pool.getConnection();
+      var query = await pool.prepare("select * from row");
+      await query.close();
+      cnx.release();
+    });
+  });
+}

test/integration_test.dart

@@ -22,6 +22,7 @@ part 'integration/row.dart';
 part 'integration/errors.dart';
 part 'integration/stored_procedures.dart';
 part 'integration/execute_multi.dart';
+part 'integration/prepared_query.dart';
 
 void main(List<String> args) {
   hierarchicalLoggingEnabled = true;
@@ -51,6 +52,7 @@ void main(List<String> args) {
   var db = options.getString('db');
   var host = options.getString('host', 'localhost');
 
+  runPreparedQueryTests(user, password, db, port, host);
   runIntTests(user, password, db, port, host);
   runIntTests2(user, password, db, port, host);
   runCharsetTests(user, password, db, port, host);