max_pool_size

Author: technige

neo4j/v1/session.py

@@ -36,6 +36,8 @@ class which can be used to obtain `Driver` instances that are used for
 from .typesystem import hydrated
 
 
+DEFAULT_MAX_POOL_SIZE = 50
+
 STATEMENT_TYPE_READ_ONLY = "r"
 STATEMENT_TYPE_READ_WRITE = "rw"
 STATEMENT_TYPE_WRITE_ONLY = "w"
@@ -91,7 +93,8 @@ def __init__(self, url, **config):
         else:
             raise ValueError("Unsupported URL scheme: %s" % parsed.scheme)
         self.config = config
-        self.sessions = deque()
+        self.max_pool_size = config.get("max_pool_size", DEFAULT_MAX_POOL_SIZE)
+        self.session_pool = deque()
 
     def session(self):
         """ Create a new session based on the graph database details
@@ -102,14 +105,33 @@ def session(self):
             >>> session = driver.session()
 
         """
-        try:
-            session = self.sessions.pop()
-        except IndexError:
-            session = Session(self)
-        else:
-            session.connection.reset()
+        session = None
+        done = False
+        while not done:
+            try:
+                session = self.session_pool.pop()
+            except IndexError:
+                session = Session(self)
+                done = True
+            else:
+                if session.healthy:
+                    session.connection.reset()
+                    done = session.healthy
         return session
 
+    def recycle(self, session):
+        """ Pass a session back to the driver for recycling, if healthy.
+
+        :param session:
+        :return:
+        """
+        pool = self.session_pool
+        for s in pool:
+            if not s.healthy:
+                pool.remove(s)
+        if session.healthy and len(pool) < self.max_pool_size and session not in pool:
+            pool.appendleft(session)
+
 
 class Result(list):
     """ A handler for the result of Cypher statement execution.
@@ -344,19 +366,25 @@ def __init__(self, driver):
         self.connection = connect(driver.host, driver.port, **driver.config)
         self.transaction = None
         self.bench_tests = []
-        self.closed = False
 
     def __del__(self):
-        if not self.closed:
+        if not self.connection.closed:
             self.connection.close()
-        self.closed = True
 
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
         self.close()
 
+    @property
+    def healthy(self):
+        """ Return ``True`` if this session is healthy, ``False`` if
+        unhealthy and ``None`` if closed.
+        """
+        connection = self.connection
+        return None if connection.closed else not connection.defunct
+
     def run(self, statement, parameters=None):
         """ Run a parameterised Cypher statement.
 
@@ -411,9 +439,7 @@ def run(self, statement, parameters=None):
     def close(self):
         """ If still usable, return this session to the driver pool it came from.
         """
-        if not self.connection.defunct:
-            self.driver.sessions.appendleft(self)
-        self.closed = True
+        self.driver.recycle(self)
 
     def begin_transaction(self):
         """ Create a new :class:`.Transaction` within this session.

test/test_session.py

@@ -24,10 +24,35 @@
 from mock import patch
 from neo4j.v1.session import GraphDatabase, CypherError, Record, record
 from neo4j.v1.typesystem import Node, Relationship, Path
-from test.util import watch
+
+
+class DriverTestCase(TestCase):
+
+    def test_healthy_session_will_be_returned_to_the_pool_on_close(self):
+        driver = GraphDatabase.driver("bolt://localhost")
+        assert len(driver.session_pool) == 0
+        driver.session().close()
+        assert len(driver.session_pool) == 1
+
+    def test_unhealthy_session_will_not_be_returned_to_the_pool_on_close(self):
+        driver = GraphDatabase.driver("bolt://localhost")
+        assert len(driver.session_pool) == 0
+        session = driver.session()
+        session.connection.defunct = True
+        session.close()
+        assert len(driver.session_pool) == 0
+
+    def session_pool_cannot_exceed_max_size(self):
+        driver = GraphDatabase.driver("bolt://localhost", max_pool_size=1)
+        assert len(driver.session_pool) == 0
+        driver.session().close()
+        assert len(driver.session_pool) == 1
+        driver.session().close()
+        assert len(driver.session_pool) == 1
 
 
 class RunTestCase(TestCase):
+
     def test_must_use_valid_url_scheme(self):
         with self.assertRaises(ValueError):
             GraphDatabase.driver("x://xxx")