Integrate "pre-ping" into connection pool.

Added native "pessimistic disconnection" handling to the :class:`.Pool`
object.  The new parameter :paramref:`.Pool.pre_ping`, available from
the engine as :paramref:`.create_engine.pool_pre_ping`, applies an
efficient form of the "pre-ping" recipe featured in the pooling
documentation, which upon each connection check out, emits a simple
statement, typically "SELECT 1", to test the connection for liveness.
If the existing connection is no longer able to respond to commands,
the connection is transparently recycled, and all other connections
made prior to the current timestamp are invalidated.

Change-Id: I89700d0075e60abd2250e54b9bd14daf03c71c00
Fixes: #3919

1 files changed, 19 insertions, 3 deletions

diff --git a/lib/sqlalchemy/testing/engines.py b/lib/sqlalchemy/testing/engines.py
index 6bca75fb1..4510ba6e5 100644
--- a/lib/sqlalchemy/testing/engines.py
+++ b/lib/sqlalchemy/testing/engines.py
@@ -161,14 +161,24 @@ class ReconnectFixture(object):
     def __init__(self, dbapi):
         self.dbapi = dbapi
         self.connections = []
+        self.is_stopped = False
 
     def __getattr__(self, key):
         return getattr(self.dbapi, key)
 
     def connect(self, *args, **kwargs):
+
         conn = self.dbapi.connect(*args, **kwargs)
-        self.connections.append(conn)
-        return conn
+        if self.is_stopped:
+            self._safe(conn.close)
+            curs = conn.cursor()  # should fail on Oracle etc.
+            # should fail for everything that didn't fail
+            # above, connection is closed
+            curs.execute("select 1")
+            assert False, "simulated connect failure didn't work"
+        else:
+            self.connections.append(conn)
+            return conn
 
     def _safe(self, fn):
         try:
@@ -178,16 +188,20 @@ class ReconnectFixture(object):
                 "ReconnectFixture couldn't "
                 "close connection: %s" % e)
 
-    def shutdown(self):
+    def shutdown(self, stop=False):
         # TODO: this doesn't cover all cases
         # as nicely as we'd like, namely MySQLdb.
         # would need to implement R. Brewer's
         # proxy server idea to get better
         # coverage.
+        self.is_stopped = stop
         for c in list(self.connections):
             self._safe(c.close)
         self.connections = []
 
+    def restart(self):
+        self.is_stopped = False
+
 
 def reconnecting_engine(url=None, options=None):
     url = url or config.db.url
@@ -200,9 +214,11 @@ def reconnecting_engine(url=None, options=None):
 
     def dispose():
         engine.dialect.dbapi.shutdown()
+        engine.dialect.dbapi.is_stopped = False
         _dispose()
 
     engine.test_shutdown = engine.dialect.dbapi.shutdown
+    engine.test_restart = engine.dialect.dbapi.restart
     engine.dispose = dispose
     return engine