[WIP] Connection Pool Refactor

src/Connection/DefaultConnectionPool.php

@@ -23,7 +23,7 @@ final class DefaultConnectionPool implements ConnectionPool
     /** @var Connector */
     private $connector;
 
-    /** @var Promise[][] */
+    /** @var \SplObjectStorage[] */
     private $connections = [];
 
     public function __construct(?Connector $connector = null)
@@ -41,23 +41,22 @@ public function getConnection(Request $request, CancellationToken $cancellation)
             $authority = $uri->getHost() . ':' . ($uri->getPort() ?: $defaultPort);
             $key = $uri->getScheme() . '://' . $authority;
 
-            if (!empty($this->connections[$key])) {
-                foreach ($this->connections[$key] as $index => $connection) {
+            if (isset($this->connections[$key])) {
+                foreach ($this->connections[$key] as $connection) {
                     $connection = yield $connection;
                     \assert($connection instanceof Connection);
 
                     if (!$connection->isBusy()) {
                         return $connection;
                     }
                 }
-
-                ++$index;
             } else {
-                $this->connections[$key] = [];
-                $index = 0;
+                $this->connections[$key] = new \SplObjectStorage;
             }
 
-            $promise = $this->connections[$key][$index] = call(function () use ($request, $isHttps, $authority, $cancellation, $key, $index) {
+            \assert($this->connections[$key] instanceof \SplObjectStorage);
+
+            $promise = call(function () use (&$promise, $request, $isHttps, $authority, $cancellation, $key) {
                 $connectContext = new ConnectContext;
 
                 if ($isHttps) {
@@ -113,27 +112,31 @@ public function getConnection(Request $request, CancellationToken $cancellation)
                     $connection = new Http1Connection($socket);
                 }
 
+                \assert($promise instanceof Promise);
+
                 $connections = &$this->connections;
-                $connection->onClose(static function () use (&$connections, $key, $index) {
-                    unset($connections[$key][$index]);
+                $connection->onClose(static function () use (&$connections, $key, $promise) {
+                    $connections[$key]->detach($promise);
 
-                    if (empty($connections[$key])) {
+                    if (!$connections[$key]->count()) {
                         unset($connections[$key]);
                     }
                 });
 
                 return $connection;
             });
 
-            $promise->onResolve(function (?\Throwable $exception) use ($key, $index): void {
+            $this->connections[$key]->attach($promise);
+
+            $promise->onResolve(function (?\Throwable $exception) use ($key, $promise): void {
                 if (!$exception) {
                     return;
                 }
 
                 // Connection failed, remove from list of connections.
-                unset($this->connections[$key][$index]);
+                $this->connections[$key]->detach($promise);
 
-                if (empty($this->connections[$key])) {
+                if (!$this->connections[$key]->count()) {
                     unset($this->connections[$key]);
                 }
             });

src/Connection/Http2Connection.php

@@ -226,12 +226,12 @@ public function request(Request $request, CancellationToken $token): Promise
 
     public function isBusy(): bool
     {
-        return $this->remainingStreams > 0 || $this->socket->isClosed();
+        return $this->remainingStreams <= 0 || $this->socket->isClosed();
     }
 
     public function onClose(callable $onClose): void
     {
-        if ($this->socket->isClosed()) {
+        if ($this->onClose === null) {
             Promise\rethrow(call($onClose, $this));
             return;
         }
@@ -243,6 +243,12 @@ public function close(): Promise
     {
         $this->socket->close();
 
+        if (!empty($this->streams)) {
+            foreach ($this->streams as $id => $stream) {
+                $this->releaseStream($id);
+            }
+        }
+
         if ($this->onClose !== null) {
             $onClose = $this->onClose;
             $this->onClose = null;
@@ -308,8 +314,12 @@ private function run(): \Generator
                     \assert($promise === null || $promise instanceof Promise);
                 }
             }
+
+            var_dump($chunk);
         } catch (\Throwable $exception) {
-           // ?
+           foreach ($this->streams as $id => $stream) {
+               $this->releaseStream($id, $exception);
+           }
         } finally {
             $this->close();
         }
@@ -954,6 +964,7 @@ private function parser(): \Generator
                     }
 
                     $status = $headers[":status"][0];
+                    unset($headers[":status"]);
 
                     if ($stream->state & Http2Stream::RESERVED) {
                         $error = self::PROTOCOL_ERROR;