Timeout configuration option works with transaction functions

references #64

Author: transistive

src/Bolt/BoltConnectionPool.php

@@ -19,6 +19,7 @@
 use Laudis\Neo4j\Common\TransactionHelper;
 use Laudis\Neo4j\Contracts\AuthenticateInterface;
 use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
+use Laudis\Neo4j\Databags\TransactionConfiguration;
 use Laudis\Neo4j\Enum\AccessMode;
 use Psr\Http\Message\UriInterface;
 use function str_starts_with;
@@ -31,10 +32,10 @@ final class BoltConnectionPool implements ConnectionPoolInterface
     /**
      * @throws Exception
      */
-    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate): StreamSocket
+    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate, TransactionConfiguration $config): StreamSocket
     {
         $host = $uri->getHost();
-        $socket = new StreamSocket($host, $uri->getPort() ?? 7687);
+        $socket = new StreamSocket($host, $uri->getPort() ?? 7687, $config->getTimeout());
 
         $scheme = $uri->getScheme();
         $explosion = explode('+', $scheme, 2);

src/Bolt/Session.php

@@ -15,7 +15,6 @@
 
 use Bolt\Bolt;
 use Bolt\connection\StreamSocket;
-use Closure;
 use Ds\Vector;
 use Exception;
 use Laudis\Neo4j\Common\TransactionHelper;
@@ -92,7 +91,7 @@ public function run(string $statement, iterable $parameters = [], ?TransactionCo
     public function writeTransaction(callable $tsxHandler, ?TransactionConfiguration $config = null)
     {
         return TransactionHelper::retry(
-            Closure::fromCallable([$this, 'beginTransaction']),
+            fn () => $this->beginTransaction([], $config),
             $tsxHandler,
             $config ?? TransactionConfiguration::default()
         );
@@ -110,8 +109,9 @@ public function transaction(callable $tsxHandler, ?TransactionConfiguration $con
 
     public function beginTransaction(?iterable $statements = null, ?TransactionConfiguration $config = null): UnmanagedTransactionInterface
     {
+        $config ??= TransactionConfiguration::default();
         try {
-            $bolt = $this->acquireBolt();
+            $bolt = $this->acquireBolt($config);
 
             $begin = $bolt->begin(['db' => $this->config->getDatabase()]);
 
@@ -137,12 +137,12 @@ public function beginTransaction(?iterable $statements = null, ?TransactionConfi
      */
     private function beginInstantTransaction(): TransactionInterface
     {
-        return new BoltUnmanagedTransaction($this->config->getDatabase(), $this->formatter, $this->acquireBolt());
+        return new BoltUnmanagedTransaction($this->config->getDatabase(), $this->formatter, $this->acquireBolt(TransactionConfiguration::default()));
     }
 
-    private function acquireBolt(): Bolt
+    private function acquireBolt(TransactionConfiguration $config): Bolt
     {
-        $bolt = new Bolt($this->pool->acquire($this->uri, $this->config->getAccessMode(), $this->auth));
+        $bolt = new Bolt($this->pool->acquire($this->uri, $this->config->getAccessMode(), $this->auth, $config));
         $this->auth->authenticateBolt($bolt, $this->uri, $this->userAgent);
 
         return $bolt;

src/Contracts/ConnectionPoolInterface.php

@@ -13,6 +13,7 @@
 
 namespace Laudis\Neo4j\Contracts;
 
+use Laudis\Neo4j\Databags\TransactionConfiguration;
 use Laudis\Neo4j\Enum\AccessMode;
 use Psr\Http\Message\UriInterface;
 
@@ -24,5 +25,5 @@ interface ConnectionPoolInterface
     /**
      * @return T
      */
-    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate);
+    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate, TransactionConfiguration $config);
 }

src/Http/HttpConnectionPool.php

@@ -15,6 +15,7 @@
 
 use Laudis\Neo4j\Contracts\AuthenticateInterface;
 use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
+use Laudis\Neo4j\Databags\TransactionConfiguration;
 use Laudis\Neo4j\Enum\AccessMode;
 use Psr\Http\Client\ClientInterface;
 use Psr\Http\Message\UriInterface;
@@ -31,7 +32,7 @@ public function __construct(ClientInterface $client)
         $this->client = $client;
     }
 
-    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate): ClientInterface
+    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate, TransactionConfiguration $config): ClientInterface
     {
         return $this->client;
     }

src/Http/HttpSession.php

@@ -73,8 +73,10 @@ public function __construct(
      */
     public function runStatements(iterable $statements, ?TransactionConfiguration $config = null): CypherList
     {
+        $config ??= TransactionConfiguration::default();
+
         $request = $this->requestFactory->createRequest('POST', $this->uri);
-        $client = $this->pool->acquire($request->getUri(), $this->config->getAccessMode(), $this->auth);
+        $client = $this->pool->acquire($request->getUri(), $this->config->getAccessMode(), $this->auth, $config);
         $content = HttpHelper::statementsToString($this->formatter, $statements);
         $request = $this->instantCommitRequest($request)->withBody($this->streamFactory->createStream($content));
 
@@ -137,9 +139,11 @@ public function run(string $statement, iterable $parameters = [], ?TransactionCo
      */
     public function beginTransaction(?iterable $statements = null, ?TransactionConfiguration $config = null): UnmanagedTransactionInterface
     {
+        $config ??= TransactionConfiguration::default();
+
         $request = $this->requestFactory->createRequest('POST', $this->uri);
         $request->getBody()->write(HttpHelper::statementsToString($this->formatter, $statements ?? []));
-        $client = $this->pool->acquire($request->getUri(), $this->config->getAccessMode(), $this->auth);
+        $client = $this->pool->acquire($request->getUri(), $this->config->getAccessMode(), $this->auth, $config);
         $response = $client->sendRequest($request);
 
         /** @var array{commit: string} $data */

src/Neo4j/Neo4jConnectionPool.php

@@ -21,6 +21,7 @@
 use Laudis\Neo4j\Common\Uri;
 use Laudis\Neo4j\Contracts\AuthenticateInterface;
 use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
+use Laudis\Neo4j\Databags\TransactionConfiguration;
 use Laudis\Neo4j\Enum\AccessMode;
 use Laudis\Neo4j\Enum\RoutingRoles;
 use Laudis\Neo4j\Types\CypherList;
@@ -51,12 +52,12 @@ public function __construct(ConnectionPoolInterface $pool)
     /**
      * @throws Exception
      */
-    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate): StreamSocket
+    public function acquire(UriInterface $uri, AccessMode $mode, AuthenticateInterface $authenticate, TransactionConfiguration $config): StreamSocket
     {
         $table = $this->routingTable($uri, $authenticate);
         $server = $this->getNextServer($table, $mode);
 
-        $socket = $this->pool->acquire($server, $mode, $authenticate);
+        $socket = $this->pool->acquire($server, $mode, $authenticate, $config);
 
         $scheme = $uri->getScheme();
         $explosion = explode('+', $scheme, 2);

tests/Integration/ComplexQueryTests.php

@@ -17,6 +17,8 @@
 use function getenv;
 use InvalidArgumentException;
 use Laudis\Neo4j\Contracts\FormatterInterface;
+use Laudis\Neo4j\Contracts\TransactionInterface;
+use Laudis\Neo4j\Databags\TransactionConfiguration;
 use Laudis\Neo4j\Exception\Neo4jException;
 use Laudis\Neo4j\Formatter\BasicFormatter;
 use Laudis\Neo4j\ParameterHelper;
@@ -264,4 +266,15 @@ public function testPeriodicCommitFail(string $alias): void
 CYPHER);
         $tsx->commit();
     }
+
+    /**
+     * @dataProvider connectionAliases
+     */
+    public function testLongQuery(string $alias): void
+    {
+        $this->client->writeTransaction(static function (TransactionInterface $tsx) {
+            $tsx->run('UNWIND range(1, 10000) AS x MERGE (:Number {value: x})');
+        }, $alias, TransactionConfiguration::default()->withTimeout(100000));
+        self::assertTrue(true);
+    }
 }