Basic routing functionality

Author: technige

.gitignore

@@ -13,3 +13,8 @@ docs/build
 dist
 *.egg-info
 build
+
+*/run/*
+
+neo4j-community-*
+neo4j-enterprise-*

.gitmodules

@@ -1,3 +0,0 @@
-[submodule "neokit"]
-	path = neokit
-	url = https://github.com/neo-technology/neokit.git

examples/test_examples.py

@@ -32,7 +32,13 @@
 # (* "good reason" is defined as knowing what you are doing)
 
 
-auth_token = basic_auth("neo4j", "neo4j")
+auth_token = basic_auth("neotest", "neotest")
+
+
+# Deliberately shadow the built-in print function to
+# mute noise from example code.
+def print(*args, **kwargs):
+    pass
 
 
 class FreshDatabaseTestCase(ServerTestCase):
@@ -48,7 +54,7 @@ class MinimalWorkingExampleTestCase(FreshDatabaseTestCase):
 
     def test_minimal_working_example(self):
         # tag::minimal-example[]
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "neo4j"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neotest", "neotest"))
         session = driver.session()
 
         session.run("CREATE (a:Person {name:'Arthur', title:'King'})")
@@ -65,33 +71,33 @@ class ExamplesTestCase(FreshDatabaseTestCase):
 
     def test_construct_driver(self):
         # tag::construct-driver[]
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "neo4j"))
+        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neotest", "neotest"))
         # end::construct-driver[]
         return driver
 
     def test_configuration(self):
         # tag::configuration[]
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "neo4j"), max_pool_size=10)
+        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neotest", "neotest"), max_pool_size=10)
         # end::configuration[]
         return driver
 
     @skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
     def test_tls_require_encryption(self):
         # tag::tls-require-encryption[]
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "neo4j"), encrypted=True)
+        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neotest", "neotest"), encrypted=True)
         # end::tls-require-encryption[]
 
     @skipUnless(SSL_AVAILABLE, "Bolt over TLS is not supported by this version of Python")
     def test_tls_trust_on_first_use(self):
         # tag::tls-trust-on-first-use[]
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "neo4j"), encrypted=True, trust=TRUST_ON_FIRST_USE)
+        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neotest", "neotest"), encrypted=True, trust=TRUST_ON_FIRST_USE)
         # end::tls-trust-on-first-use[]
         assert driver
 
     @skip("testing verified certificates not yet supported ")
     def test_tls_signed(self):
         # tag::tls-signed[]
-        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neo4j", "neo4j"), encrypted=True, trust=TRUST_SIGNED_CERTIFICATES)
+        driver = GraphDatabase.driver("bolt://localhost", auth=basic_auth("neotest", "neotest"), encrypted=True, trust=TRUST_SIGNED_CERTIFICATES)
         # end::tls-signed[]
         assert driver
 

neo4j/util.py

@@ -19,6 +19,12 @@
 # limitations under the License.
 
 
+try:
+    from collections.abc import MutableSet
+except ImportError:
+    from collections import MutableSet, OrderedDict
+else:
+    from collections import OrderedDict
 import logging
 from sys import stdout
 
@@ -55,6 +61,13 @@ def __init__(self, logger_name):
         self.logger = logging.getLogger(self.logger_name)
         self.formatter = ColourFormatter("%(asctime)s  %(message)s")
 
+    def __enter__(self):
+        self.watch()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+
     def watch(self, level=logging.INFO, out=stdout):
         self.stop()
         handler = logging.StreamHandler(out)
@@ -81,3 +94,58 @@ def watch(logger_name, level=logging.INFO, out=stdout):
     watcher = Watcher(logger_name)
     watcher.watch(level, out)
     return watcher
+
+
+class RoundRobinSet(MutableSet):
+
+    def __init__(self, elements=()):
+        self._elements = OrderedDict.fromkeys(elements)
+        self._current = None
+
+    def __repr__(self):
+        return "{%s}" % ", ".join(map(repr, self._elements))
+
+    def __contains__(self, element):
+        return element in self._elements
+
+    def __next__(self):
+        current = None
+        if self._elements:
+            if self._current is None:
+                self._current = 0
+            else:
+                self._current = (self._current + 1) % len(self._elements)
+            current = list(self._elements.keys())[self._current]
+        return current
+
+    def __iter__(self):
+        return iter(self._elements)
+
+    def __len__(self):
+        return len(self._elements)
+
+    def add(self, element):
+        self._elements[element] = None
+
+    def clear(self):
+        self._elements.clear()
+
+    def discard(self, element):
+        try:
+            del self._elements[element]
+        except KeyError:
+            pass
+
+    def remove(self, element):
+        try:
+            del self._elements[element]
+        except KeyError:
+            raise ValueError(element)
+
+    def update(self, elements=()):
+        self._elements.update(OrderedDict.fromkeys(elements))
+
+    def replace(self, elements=()):
+        e = self._elements
+        e.clear()
+        e.update(OrderedDict.fromkeys(elements))

neo4j/v1/bolt.py

@@ -37,6 +37,7 @@
 from select import select
 from socket import create_connection, SHUT_RDWR, error as SocketError
 from struct import pack as struct_pack, unpack as struct_unpack, unpack_from as struct_unpack_from
+from threading import Lock
 
 from .constants import DEFAULT_USER_AGENT, KNOWN_HOSTS, MAGIC_PREAMBLE, TRUST_DEFAULT, TRUST_ON_FIRST_USE
 from .compat import hex2
@@ -200,19 +201,23 @@ def on_ignored(self, metadata=None):
 
 
 class Connection(object):
-    """ Server connection through which all protocol messages
-    are sent and received. This class is designed for protocol
-    version 1.
+    """ Server connection for Bolt protocol v1.
+
+    A :class:`.Connection` should be constructed following a
+    successful Bolt handshake and takes the socket over which
+    the handshake was carried out.
 
     .. note:: logs at INFO level
     """
 
     def __init__(self, sock, **config):
-        self.defunct = False
+        self.address = sock.getpeername()
         self.channel = ChunkChannel(sock)
         self.packer = Packer(self.channel)
         self.responses = deque()
+        self.in_use = False
         self.closed = False
+        self.defunct = False
 
         # Determine the user agent and ensure it is a Unicode value
         user_agent = config.get("user_agent", DEFAULT_USER_AGENT)
@@ -245,13 +250,6 @@ def on_failure(metadata):
     def __del__(self):
         self.close()
 
-    @property
-    def healthy(self):
-        """ Return ``True`` if this connection is healthy, ``False`` if
-        unhealthy and ``None`` if closed.
-        """
-        return None if self.closed else not self.defunct
-
     def append(self, signature, fields=(), response=None):
         """ Add a message to the outgoing queue.
 
@@ -362,6 +360,57 @@ def close(self):
             self.closed = True
 
 
+class ConnectionPool(object):
+    """ A collection of connections to one or more server addresses.
+    """
+
+    def __init__(self, connector):
+        self.connector = connector
+        self.connections = {}
+        self.lock = Lock()
+
+    def acquire(self, address):
+        """ Acquire a connection to a given address from the pool.
+        This method is thread safe.
+        """
+        with self.lock:
+            try:
+                connections = self.connections[address]
+            except KeyError:
+                connections = self.connections[address] = deque()
+            for connection in list(connections):
+                if connection.closed or connection.defunct:
+                    connections.remove(connection)
+                    continue
+                if not connection.in_use:
+                    connection.in_use = True
+                    return connection
+            connection = self.connector(address)
+            connection.in_use = True
+            connections.append(connection)
+            return connection
+
+    def release(self, connection):
+        """ Release a connection back into the pool.
+        This method is thread safe.
+        """
+        with self.lock:
+            connection.in_use = False
+
+    def close(self):
+        """ Close all connections and empty the pool.
+        This method is thread safe.
+        """
+        with self.lock:
+            for _, connections in self.connections.items():
+                for connection in connections:
+                    try:
+                        connection.close()
+                    except IOError:
+                        pass
+            self.connections.clear()
+
+
 class CertificateStore(object):
 
     def match_or_trust(self, host, der_encoded_certificate):

neo4j/v1/constants.py

@@ -34,9 +34,15 @@
 
 ENCRYPTION_OFF = 0
 ENCRYPTION_ON = 1
-ENCRYPTION_NON_LOCAL = 2
-ENCRYPTION_DEFAULT = ENCRYPTION_NON_LOCAL if SSL_AVAILABLE else ENCRYPTION_OFF
-
-TRUST_ON_FIRST_USE = 0
-TRUST_SIGNED_CERTIFICATES = 1
-TRUST_DEFAULT = TRUST_ON_FIRST_USE
+ENCRYPTION_DEFAULT = ENCRYPTION_ON if SSL_AVAILABLE else ENCRYPTION_OFF
+
+TRUST_ON_FIRST_USE = 0          # Deprecated
+TRUST_SIGNED_CERTIFICATES = 1   # Deprecated
+TRUST_ALL_CERTIFICATES = 2
+TRUST_CUSTOM_CA_SIGNED_CERTIFICATES = 3
+TRUST_SYSTEM_CA_SIGNED_CERTIFICATES = 4
+TRUST_DEFAULT = TRUST_ALL_CERTIFICATES
+
+READ_ACCESS = "READ"
+WRITE_ACCESS = "WRITE"
+ACCESS_DEFAULT = WRITE_ACCESS

neo4j/v1/exceptions.py

@@ -43,6 +43,16 @@ def __init__(self, data):
                 setattr(self, key, value)
 
 
+class TransactionError(Exception):
+    """ Raised when an error occurs while using a transaction.
+    """
+
+
 class ResultError(Exception):
     """ Raised when an error occurs while consuming a result.
     """
+
+
+class ServiceUnavailable(Exception):
+    """ Raised when no database service is available.
+    """