Naive attempt to implement keepalive connections

src/Client.php

@@ -13,16 +13,42 @@ class Client
     public function __construct(LoopInterface $loop, ConnectorInterface $connector = null)
     {
         if ($connector === null) {
-            $connector = new Connector($loop);
+            $connector = new KeepaliveConnector(new Connector($loop));
         }
 
-        $this->connector = $connector;
+        $this->connector = new KeepaliveConnector($connector);
     }
 
     public function request($method, $url, array $headers = array(), $protocolVersion = '1.0')
     {
         $requestData = new RequestData($method, $url, $headers, $protocolVersion);
 
-        return new Request($this->connector, $requestData);
+        $that = $this;
+        $request = new Request($this->connector, $requestData);
+        $request->on('close', function() use ($that, $url) {
+            $that->connector->handleConnectionClose($url);
+        });
+        $request->on('response',
+            function(Response $response) use ($url, $headers, $that) {
+                if ($that->isConnectionClose($headers)
+                    || $that->isConnectionClose($response->getHeaders())) {
+                    $that->connector->handleConnectionClose($url);
+                }
+            }
+        );
+
+        return $request;
+    }
+
+    /** @internal */
+    public function isConnectionClose($headers)
+    {
+        $normalizedHeaders = array_change_key_case($headers, CASE_LOWER);
+
+        if (!isset($normalizedHeaders['connection'])) {
+            return false;
+        }
+
+        return strtolower($normalizedHeaders['connection']) === 'close';
     }
 }

src/KeepaliveConnector.php

@@ -0,0 +1,52 @@
+<?php
+
+namespace React\HttpClient;
+
+use React\Promise\PromiseInterface;
+use React\Promise;
+use React\Socket\ConnectionInterface;
+use React\Socket\ConnectorInterface;
+
+/**
+ * @internal
+ */
+class KeepaliveConnector implements ConnectorInterface
+{
+    private $directConnector;
+
+    /** @var PromiseInterface[] */
+    private $aliveConnections = array();
+
+    public function __construct(ConnectorInterface $directConnector)
+    {
+        $this->directConnector = $directConnector;
+    }
+
+    public function connect($uri)
+    {
+        return isset($this->aliveConnections[$uri])
+            ? $this->tryReuseAliveConnection($uri)
+            : $this->createNewConnection($uri);
+    }
+
+    public function handleConnectionClose($uri)
+    {
+        unset($this->aliveConnections[$uri]);
+    }
+
+    private function tryReuseAliveConnection($uri)
+    {
+        return Promise\resolve($this->aliveConnections[$uri]);
+    }
+
+    private function createNewConnection($uri)
+    {
+        $connection = $this->directConnector->connect($uri);
+
+        $that = $this;
+        return $connection->then(function(ConnectionInterface $connection) use ($that, $uri) {
+            $that->aliveConnections[$uri] = $connection;
+            return $connection;
+        });
+    }
+}

src/Request.php

@@ -132,10 +132,10 @@ public function handleDrain()
     /** @internal */
     public function handleData($data)
     {
-        $this->buffer .= $data;
-
         // buffer until double CRLF (or double LF for compatibility with legacy servers)
-        if (false !== strpos($this->buffer, "\r\n\r\n") || false !== strpos($this->buffer, "\n\n")) {
+        if (false !== strpos($data, "\r\n\r\n") || false !== strpos($data, "\n\n")) {
+            $this->buffer .= $data;
+
             try {
                 list($response, $bodyChunk) = $this->parseResponse($this->buffer);
             } catch (\InvalidArgumentException $exception) {