added run, commit, rollback and pull to connection

Author: transistive

src/Bolt/BoltResult.php

@@ -13,7 +13,6 @@
 
 namespace Laudis\Neo4j\Bolt;
 
-use function array_key_exists;
 use function array_splice;
 use Bolt\protocol\V4;
 use function call_user_func;
@@ -109,24 +108,9 @@ public function consume(): array
         return $this->meta ?? [];
     }
 
-    /**
-     * @return non-empty-list<list>
-     */
-    private function pull(): array
-    {
-        $protocol = $this->connection->getImplementation();
-        if (!$protocol instanceof V4) {
-            /** @var non-empty-list<list> */
-            return $protocol->pullAll(['qid' => $this->qid]);
-        }
-
-        /** @var non-empty-list<list> */
-        return $protocol->pull(['n' => $this->fetchSize, 'qid' => $this->qid]);
-    }
-
     private function fetchResults(): void
     {
-        $meta = $this->pull();
+        $meta = $this->connection->pull($this->qid, $this->fetchSize);
 
         /** @var list<list> $rows */
         $rows = array_splice($meta, 0, count($meta) - 1);

src/Bolt/BoltUnmanagedTransaction.php

@@ -50,7 +50,7 @@ final class BoltUnmanagedTransaction implements UnmanagedTransactionInterface
     /** @psalm-readonly */
     private BoltConnection $connection;
     /** @psalm-readonly */
-    private string $database;
+    private ?string $database;
 
     private bool $isRolledBack = false;
 
@@ -61,7 +61,7 @@ final class BoltUnmanagedTransaction implements UnmanagedTransactionInterface
     /**
      * @param FormatterInterface<T> $formatter
      */
-    public function __construct(string $database, FormatterInterface $formatter, BoltConnection $connection, SessionConfiguration $config, TransactionConfiguration $tsxConfig)
+    public function __construct(?string $database, FormatterInterface $formatter, BoltConnection $connection, SessionConfiguration $config, TransactionConfiguration $tsxConfig)
     {
         $this->formatter = $formatter;
         $this->connection = $connection;
@@ -82,7 +82,7 @@ public function commit(iterable $statements = []): CypherList
         });
 
         try {
-            $this->getBolt()->commit();
+            $this->connection->commit();
             $this->isCommitted = true;
         } catch (MessageException $e) {
             $this->handleMessageException($e);
@@ -96,7 +96,7 @@ public function commit(iterable $statements = []): CypherList
     public function rollback(): void
     {
         try {
-            $this->connection->getImplementation()->rollback();
+            $this->connection->rollback();
             $this->isRolledBack = true;
         } catch (MessageException $e) {
             $this->handleMessageException($e);
@@ -118,13 +118,16 @@ public function run(string $statement, iterable $parameters = [])
      */
     public function runStatement(Statement $statement)
     {
-        $extra = ['db' => $this->database, 'tx_timeout' => (int) ($this->tsxConfig->getTimeout() * 1000)];
         $parameters = ParameterHelper::formatParameters($statement->getParameters(), true);
         $start = microtime(true);
 
         try {
-            /** @var BoltMeta $meta */
-            $meta = $this->getBolt()->run($statement->getText(), $parameters->toArray(), $extra);
+            $meta = $this->connection->run(
+                $statement->getText(),
+                $parameters->toArray(),
+                $this->database,
+                $this->tsxConfig->getTimeout()
+            );
             $run = microtime(true);
         } catch (MessageException $e) {
             $this->handleMessageException($e);
@@ -156,14 +159,6 @@ public function runStatements(iterable $statements): CypherList
         return new CypherList($tbr);
     }
 
-    /**
-     * @psalm-immutable
-     */
-    private function getBolt(): V3
-    {
-        return $this->connection->getImplementation();
-    }
-
     /**
      * @throws Neo4jException
      *

src/Bolt/Session.php

@@ -170,7 +170,7 @@ private function startTransaction(TransactionConfiguration $config, SessionConfi
         try {
             $connection = $this->acquireConnection($config, $sessionConfig);
 
-            $connection->getImplementation()->begin(['db' => $this->config->getDatabase(), 'tx_timeout' => (int) ($config->getTimeout() * 1000)]);
+            $connection->begin($this->config->getDatabase(), $config->getTimeout());
         } catch (MessageException $e) {
             if (isset($connection)) {
                 $connection->reset();

src/Common/BoltConnection.php

@@ -14,6 +14,7 @@
 namespace Laudis\Neo4j\Common;
 
 use Bolt\protocol\V3;
+use Bolt\protocol\V4;
 use Laudis\Neo4j\BoltFactory;
 use Laudis\Neo4j\Contracts\ConnectionInterface;
 use Laudis\Neo4j\Databags\DatabaseInfo;
@@ -25,6 +26,8 @@
 
 /**
  * @implements ConnectionInterface<V3>
+ *
+ * @psalm-import-type BoltMeta from \Laudis\Neo4j\Contracts\FormatterInterface
  */
 final class BoltConnection implements ConnectionInterface
 {
@@ -144,9 +147,7 @@ public function isOpen(): bool
 
     public function open(): void
     {
-        if ($this->boltProtocol === null) {
-            $this->boltProtocol = $this->factory->build()[0];
-        }
+        $this->boltProtocol = $this->factory->build()[0];
     }
 
     public function setTimeout(float $timeout): void
@@ -156,15 +157,15 @@ public function setTimeout(float $timeout): void
 
     public function close(): void
     {
-        if ($this->ownerCount === 0 && $this->boltProtocol !== null) {
+        if ($this->ownerCount === 0) {
             $this->boltProtocol = null;
             $this->hasBeenReset = false;
         }
     }
 
     public function reset(): void
     {
-        if ($this->boltProtocol !== null) {
+        if ($this->boltProtocol) {
             $this->boltProtocol->reset();
             $this->boltProtocol = $this->factory->build()[0];
             if ($this->ownerCount > 0) {
@@ -179,17 +180,69 @@ public function reset(): void
      */
     public function begin(?string $database, ?float $timeout): void
     {
-        $params = [];
-        if ($database) {
-            $params['db'] = $database;
+        if ($this->boltProtocol === null) {
+            throw new RuntimeException('Cannot begin on a closed connection');
         }
-        if ($timeout) {
-            $params['tx_timeout'] = $timeout * 1000;
+
+        $this->boltProtocol->begin($this->buildExtra($database, $timeout));
+    }
+
+    /**
+     * @return BoltMeta
+     */
+    public function run(string $text, array $parameters, ?string $database, ?float $timeout): array
+    {
+        if ($this->boltProtocol === null) {
+            throw new RuntimeException('Cannot run on a closed connection');
         }
 
-        if ($this->boltProtocol) {
-            $this->boltProtocol->begin($params);
+        /** @var BoltMeta */
+        return $this->boltProtocol->run($text, $parameters, $this->buildExtra($database, $timeout));
+    }
+
+    public function commit(): void
+    {
+        if ($this->boltProtocol === null) {
+            throw new RuntimeException('Cannot commit on a closed connection');
         }
+
+        $this->boltProtocol->commit();
+    }
+
+    public function rollback(): void
+    {
+        if ($this->boltProtocol === null) {
+            throw new RuntimeException('Cannot commit on a closed connection');
+        }
+
+        $this->boltProtocol->rollback();
+    }
+
+    /**
+     * @return non-empty-list<list>
+     */
+    public function pull(?int $qid, ?int $fetchSize): array
+    {
+        if ($this->boltProtocol === null) {
+            throw new RuntimeException('Cannot pull on a closed connection');
+        }
+
+        $extra = [];
+        if ($fetchSize) {
+            $extra['n'] = $fetchSize;
+        }
+
+        if ($qid) {
+            $extra['qid'] = $qid;
+        }
+
+        if (!$this->boltProtocol instanceof V4) {
+            /** @var non-empty-list<list> */
+            return $this->boltProtocol->pullAll($extra);
+        }
+
+        /** @var non-empty-list<list> */
+        return $this->boltProtocol->pull($extra);
     }
 
     /**
@@ -215,4 +268,17 @@ public function decrementOwner(): void
     {
         --$this->ownerCount;
     }
+
+    private function buildExtra(?string $database, ?float $timeout): array
+    {
+        $extra = [];
+        if ($database) {
+            $extra['db'] = $database;
+        }
+        if ($timeout) {
+            $extra['tx_timeout'] = $timeout * 1000;
+        }
+
+        return $extra;
+    }
 }