Merge pull request #339 from neo4j/4.0-bolt-4.0

Adds Bolt 4.0 protocol support at connector level

Author: martin-neotech

neo4j/io/__init__.py

@@ -124,6 +124,9 @@ def open(cls, address, *, auth=None, timeout=None, **config):
         if config.protocol_version == (3, 0):
             from neo4j.io._bolt3 import Bolt3
             connection = Bolt3(address, s, auth=auth, **config)
+        elif config.protocol_version == (4, 0):
+            from neo4j.io._bolt4x0 import Bolt4x0
+            connection = Bolt4x0(address, s, auth=auth, **config)
         else:
             log.debug("[#%04X]  S: <CLOSE>", s.getpeername()[1])
             s.shutdown(SHUT_RDWR)
@@ -160,17 +163,50 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_value, traceback):
         self.close()
 
-    def run(self, statement, parameters=None, mode=None, bookmarks=None, metadata=None, timeout=None, **handlers):
-        raise NotImplementedError
+    def run(self, statement, parameters=None, mode=None, bookmarks=None, metadata=None,
+            timeout=None, db=None, **handlers):
+        """ Appends a RUN message to the output stream.
+
+        :param statement: Cypher query string
+        :param parameters: dictionary of Cypher parameters
+        :param mode: access mode for routing - "READ" or "WRITE" (default)
+        :param bookmarks: iterable of bookmark values after which this transaction should begin
+        :param metadata: custom metadata dictionary to attach to the transaction
+        :param timeout: timeout for transaction execution (seconds)
+        :param db: name of the database against which to begin the transaction
+        :param handlers: handler functions passed into the returned Response object
+        :return: Response object
+        """
 
-    def discard_all(self, **handlers):
-        raise NotImplementedError
+    def discard(self, n=-1, qid=-1, **handlers):
+        """ Appends a DISCARD message to the output stream.
 
-    def pull_all(self, **handlers):
-        raise NotImplementedError
+        :param n: number of records to discard, default = -1 (ALL)
+        :param qid: query ID to discard for, default = -1 (last query)
+        :param handlers: handler functions passed into the returned Response object
+        :return: Response object
+        """
 
-    def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None, **handlers):
-        raise NotImplementedError
+    def pull(self, n=-1, qid=-1, **handlers):
+        """ Appends a PULL message to the output stream.
+
+        :param n: number of records to pull, default = -1 (ALL)
+        :param qid: query ID to pull for, default = -1 (last query)
+        :param handlers: handler functions passed into the returned Response object
+        :return: Response object
+        """
+
+    def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None, db=None, **handlers):
+        """ Appends a BEGIN message to the output stream.
+
+        :param mode: access mode for routing - "READ" or "WRITE" (default)
+        :param bookmarks: iterable of bookmark values after which this transaction should begin
+        :param metadata: custom metadata dictionary to attach to the transaction
+        :param timeout: timeout for transaction execution (seconds)
+        :param db: name of the database against which to begin the transaction
+        :param handlers: handler functions passed into the returned Response object
+        :return: Response object
+        """
 
     def commit(self, **handlers):
         raise NotImplementedError
@@ -462,7 +498,7 @@ def fail(md):
                 log.debug("[#%04X]  C: <ROUTING> query=%r", cx.local_port, self.routing_context or {})
                 cx.run("CALL dbms.cluster.routing.getRoutingTable($context)",
                        {"context": self.routing_context}, on_success=metadata.update, on_failure=fail)
-                cx.pull_all(on_success=metadata.update, on_records=records.extend)
+                cx.pull(on_success=metadata.update, on_records=records.extend)
                 cx.send_all()
                 cx.fetch_all()
                 routing_info = [dict(zip(metadata.get("fields", ()), values)) for values in records]

neo4j/io/_bolt3.py

@@ -26,6 +26,7 @@
 from neo4j.api import (
     Version,
 )
+from neo4j.io._courier import MessageInbox
 from neo4j.meta import get_user_agent
 from neo4j.exceptions import (
     ProtocolError,
@@ -39,7 +40,6 @@
     SessionExpired,
 )
 from neo4j.packstream import (
-    UnpackableBuffer,
     Unpacker,
     Packer,
 )
@@ -142,7 +142,10 @@ def hello(self):
         self.send_all()
         self.fetch_all()
 
-    def run(self, statement, parameters=None, mode=None, bookmarks=None, metadata=None, timeout=None, **handlers):
+    def run(self, statement, parameters=None, mode=None, bookmarks=None, metadata=None,
+            timeout=None, db=None, **handlers):
+        if db is not None:
+            raise ValueError("Database selection is not supported in Bolt 3")
         if not parameters:
             parameters = {}
         extra = {}
@@ -170,15 +173,26 @@ def run(self, statement, parameters=None, mode=None, bookmarks=None, metadata=No
         else:
             self._append(b"\x10", fields, Response(self, **handlers))
 
-    def discard_all(self, **handlers):
+    def discard(self, n=-1, qid=-1, **handlers):
+        if n != -1:
+            raise ValueError("Incremental discard is not supported in Bolt 3")
+        if qid != -1:
+            raise ValueError("Query selection on discard is not supported in Bolt 3")
         log.debug("[#%04X]  C: DISCARD_ALL", self.local_port)
         self._append(b"\x2F", (), Response(self, **handlers))
 
-    def pull_all(self, **handlers):
+    def pull(self, n=-1, qid=-1, **handlers):
+        if n != -1:
+            raise ValueError("Incremental pull is not supported in Bolt 3")
+        if qid != -1:
+            raise ValueError("Query selection on pull is not supported in Bolt 3")
         log.debug("[#%04X]  C: PULL_ALL", self.local_port)
         self._append(b"\x3F", (), Response(self, **handlers))
 
-    def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None, **handlers):
+    def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
+              db=None, **handlers):
+        if db is not None:
+            raise ValueError("Database selection is not supported in Bolt 3")
         extra = {}
         if mode:
             extra["mode"] = mode
@@ -538,56 +552,16 @@ def recv_into(self, buffer, n_bytes=0, flags=0):
         return n_bytes
 
 
-class Inbox:
-
-    def __init__(self, s, on_error):
-        super(Inbox, self).__init__()
-        self.on_error = on_error
-        self._messages = self._yield_messages(s)
-
-    def __iter__(self):
-        return self
+class Inbox(MessageInbox):
 
     def __next__(self):
-        return next(self._messages)
-
-    @classmethod
-    def _load_chunks(cls, sock, buffer):
-        chunk_size = 0
-        while True:
-            if chunk_size == 0:
-                buffer.receive(sock, 2)
-            chunk_size = buffer.pop_u16()
-            if chunk_size > 0:
-                buffer.receive(sock, chunk_size + 2)
-            yield chunk_size
-
-    def _yield_messages(self, sock):
-        try:
-            buffer = UnpackableBuffer()
-            chunk_loader = self._load_chunks(sock, buffer)
-            unpacker = Unpacker(buffer)
-            details = []
-            while True:
-                unpacker.reset()
-                details[:] = ()
-                chunk_size = -1
-                while chunk_size != 0:
-                    chunk_size = next(chunk_loader)
-                summary_signature = None
-                summary_metadata = None
-                size, signature = unpacker.unpack_structure_header()
-                if size > 1:
-                    raise ProtocolError("Expected one field")
-                if signature == b"\x71":
-                    data = unpacker.unpack()
-                    details.append(data)
-                else:
-                    summary_signature = signature
-                    summary_metadata = unpacker.unpack_map()
-                yield details, summary_signature, summary_metadata
-        except OSError as error:
-            self.on_error(error)
+        tag, fields = self.pop()
+        if tag == b"\x71":
+            return fields, None, None
+        elif fields:
+            return [], tag, fields[0]
+        else:
+            return [], tag, None
 
 
 class Response: