feat: Add ability to close all connections attached to a driver

src/Basic/Driver.php

@@ -57,4 +57,9 @@ public static function create(string|UriInterface $uri, ?DriverConfiguration $co
 
         return new self($driver);
     }
+
+    public function closeConnections(): void
+    {
+        $this->driver->closeConnections();
+    }
 }

src/Bolt/BoltConnection.php

@@ -143,7 +143,23 @@ public function getAuthentication(): AuthenticateInterface
      */
     public function isOpen(): bool
     {
-        return !in_array($this->protocol()->serverState, [ServerState::DISCONNECTED, ServerState::DEFUNCT], true);
+        return !in_array(
+            $this->protocol()->serverState,
+            [ServerState::DISCONNECTED, ServerState::DEFUNCT],
+            true
+        );
+    }
+
+    /**
+     * @psalm-mutation-free
+     */
+    public function isStreaming(): bool
+    {
+        return in_array(
+            $this->protocol()->serverState,
+            [ServerState::STREAMING, ServerState::TX_STREAMING],
+            true
+        );
     }
 
     public function setTimeout(float $timeout): void
@@ -154,7 +170,7 @@ public function setTimeout(float $timeout): void
     public function consumeResults(): void
     {
         $this->logger?->log(LogLevel::DEBUG, 'Consuming results');
-        if ($this->protocol()->serverState !== ServerState::STREAMING && $this->protocol()->serverState !== ServerState::TX_STREAMING) {
+        if (!$this->isStreaming()) {
             $this->subscribedResults = [];
 
             return;
@@ -225,8 +241,14 @@ public function discard(?int $qid): void
      *
      * @return BoltMeta
      */
-    public function run(string $text, array $parameters, ?string $database, ?float $timeout, BookmarkHolder $holder, ?AccessMode $mode): array
-    {
+    public function run(
+        string $text,
+        array $parameters,
+        ?string $database,
+        ?float $timeout,
+        BookmarkHolder $holder,
+        ?AccessMode $mode
+    ): array {
         $extra = $this->buildRunExtra($database, $timeout, $holder, $mode);
         $this->logger?->log(LogLevel::DEBUG, 'RUN', $extra);
         $response = $this->protocol()
@@ -298,10 +320,15 @@ public function pull(?int $qid, ?int $fetchSize): array
     }
 
     public function __destruct()
+    {
+        $this->close();
+    }
+
+    public function close(): void
     {
         try {
             if ($this->isOpen()) {
-                if ($this->protocol()->serverState === ServerState::STREAMING || $this->protocol()->serverState === ServerState::TX_STREAMING) {
+                if ($this->isStreaming()) {
                     $this->consumeResults();
                 }
 

src/Bolt/BoltDriver.php

@@ -109,4 +109,9 @@ public function verifyConnectivity(?SessionConfiguration $config = null): bool
 
         return true;
     }
+
+    public function closeConnections(): void
+    {
+        $this->pool->close();
+    }
 }

src/Bolt/ConnectionPool.php

@@ -46,8 +46,12 @@ public function __construct(
         private readonly ?Neo4jLogger $logger
     ) {}
 
-    public static function create(UriInterface $uri, AuthenticateInterface $auth, DriverConfiguration $conf, SemaphoreInterface $semaphore): self
-    {
+    public static function create(
+        UriInterface $uri,
+        AuthenticateInterface $auth,
+        DriverConfiguration $conf,
+        SemaphoreInterface $semaphore
+    ): self {
         return new self(
             $semaphore,
             BoltFactory::create($conf->getLogger()),
@@ -165,4 +169,12 @@ private function returnAnyAvailableConnection(SessionConfiguration $config): ?Co
 
         return null;
     }
+
+    public function close(): void
+    {
+        foreach ($this->activeConnections as $activeConnection) {
+            $activeConnection->close();
+        }
+        $this->activeConnections = [];
+    }
 }

src/Contracts/ConnectionInterface.php

@@ -112,4 +112,9 @@ public function getEncryptionLevel(): string;
      * Returns the user agent handling this connection.
      */
     public function getUserAgent(): string;
+
+    /**
+     * Closes the connection.
+     */
+    public function close(): void;
 }

src/Contracts/ConnectionPoolInterface.php

@@ -44,4 +44,9 @@ public function acquire(SessionConfiguration $config): Generator;
      * Releases a connection back to the pool.
      */
     public function release(ConnectionInterface $connection): void;
+
+    /**
+     * Closes all connections in the pool.
+     */
+    public function close(): void;
 }

src/Contracts/DriverInterface.php

@@ -38,4 +38,9 @@ public function createSession(?SessionConfiguration $config = null): SessionInte
      * Returns true if the driver can make a valid connection with the server.
      */
     public function verifyConnectivity(?SessionConfiguration $config = null): bool;
+
+    /**
+     * Closes all connections in the pool.
+     */
+    public function closeConnections(): void;
 }

src/Http/HttpConnectionPool.php

@@ -127,4 +127,9 @@ public function release(ConnectionInterface $connection): void
     {
         // Nothing to release in the current HTTP Protocol implementation
     }
+
+    public function close(): void
+    {
+        // Nothing to close in the current HTTP Protocol implementation
+    }
 }

src/Http/HttpDriver.php

@@ -198,4 +198,9 @@ private function tsxUrl(SessionConfiguration $config): Resolvable
             return str_replace('{databaseName}', $database, $tsx);
         });
     }
+
+    public function closeConnections(): void
+    {
+        // Nothing to close in the current HTTP Protocol implementation
+    }
 }

src/Neo4j/Neo4jConnectionPool.php

@@ -233,4 +233,13 @@ private function createKey(ConnectionRequestData $data, ?SessionConfiguration $c
             ':',
         ], '|', $key);
     }
+
+    public function close(): void
+    {
+        foreach (self::$pools as $pool) {
+            $pool->close();
+        }
+        self::$pools = [];
+        $this->cache->clear();
+    }
 }

src/Neo4j/Neo4jDriver.php

@@ -111,4 +111,9 @@ public function verifyConnectivity(?SessionConfiguration $config = null): bool
 
         return true;
     }
+
+    public function closeConnections(): void
+    {
+        $this->pool->close();
+    }
 }