WIP: send RESET when putting a connection back into the pool

Author: robsdedude

neo4j/io/__init__.py

@@ -80,6 +80,7 @@
     ClientError,
     ConfigurationError,
     DriverError,
+    Neo4jError,
     ReadServiceUnavailable,
     ServiceUnavailable,
     SessionExpired,
@@ -107,6 +108,8 @@ class Bolt:
 
     PROTOCOL_VERSION = None
 
+    _is_reset = True
+
     @classmethod
     def protocol_handlers(cls, protocol_version=None):
         """ Return a dictionary of available Bolt protocol handlers,
@@ -342,6 +345,10 @@ def commit(self, **handlers):
     def rollback(self, **handlers):
         raise NotImplementedError
 
+    @property
+    def is_reset(self):
+        return self._is_reset
+
     def reset(self):
         """ Add a RESET message to the outgoing queue, send
         it and consume all remaining messages.
@@ -483,6 +490,13 @@ def release(self, *connections):
         """
         with self.lock:
             for connection in connections:
+                if not connection.is_reset:
+                    try:
+                        connection.reset()
+                    except (Neo4jError, DriverError, BoltError) as e:
+                        log.debug(
+                            "Reset on IOPool.release failed: {}".format(e)
+                        )
                 connection.in_use = False
             self.cond.notify_all()
 

neo4j/io/_bolt3.py

@@ -99,7 +99,6 @@ def __init__(self, unresolved_address, sock, max_connection_lifetime, *, auth=No
         self._creation_timestamp = perf_counter()
         self.supports_multiple_results = False
         self.supports_multiple_databases = False
-        self._is_reset = True
         self.routing_context = routing_context
 
         # Determine the user agent

neo4j/io/_bolt4.py

@@ -99,7 +99,6 @@ def __init__(self, unresolved_address, sock, max_connection_lifetime, *, auth=No
         self._creation_timestamp = perf_counter()
         self.supports_multiple_results = True
         self.supports_multiple_databases = True
-        self._is_reset = True
         self.routing_context = routing_context
 
         # Determine the user agent

neo4j/work/transaction.py

@@ -160,7 +160,7 @@ def rollback(self):
             raise TransactionError("Transaction closed")
         metadata = {}
         try:
-            if not self._connection._is_reset:
+            if not self._connection.is_reset:
                 # DISCARD pending records then do a rollback.
                 self._consume_results()
                 self._connection.rollback(on_success=metadata.update)