diff --git a/src/Browser.php b/src/Browser.php
index 3e3458af..12bce6b5 100644
--- a/src/Browser.php
+++ b/src/Browser.php
@@ -23,6 +23,7 @@ class Browser
     private $baseUrl;
     private $protocolVersion = '1.1';
     private $defaultHeaders = array(
+        'Connection' => 'close',
         'User-Agent' => 'ReactPHP/1'
     );
 
diff --git a/src/Client/Client.php b/src/Client/Client.php
index c3fd4570..7a5180ab 100644
--- a/src/Client/Client.php
+++ b/src/Client/Client.php
@@ -3,30 +3,25 @@
 namespace React\Http\Client;
 
 use Psr\Http\Message\RequestInterface;
-use React\EventLoop\LoopInterface;
+use React\Http\Io\ClientConnectionManager;
 use React\Http\Io\ClientRequestStream;
-use React\Socket\Connector;
-use React\Socket\ConnectorInterface;
 
 /**
  * @internal
  */
 class Client
 {
-    private $connector;
+    /** @var ClientConnectionManager */
+    private $connectionManager;
 
-    public function __construct(LoopInterface $loop, ConnectorInterface $connector = null)
+    public function __construct(ClientConnectionManager $connectionManager)
     {
-        if ($connector === null) {
-            $connector = new Connector(array(), $loop);
-        }
-
-        $this->connector = $connector;
+        $this->connectionManager = $connectionManager;
     }
 
     /** @return ClientRequestStream */
     public function request(RequestInterface $request)
     {
-        return new ClientRequestStream($this->connector, $request);
+        return new ClientRequestStream($this->connectionManager, $request);
     }
 }
diff --git a/src/Io/ClientConnectionManager.php b/src/Io/ClientConnectionManager.php
new file mode 100644
index 00000000..faac98b6
--- /dev/null
+++ b/src/Io/ClientConnectionManager.php
@@ -0,0 +1,137 @@
+<?php
+
+namespace React\Http\Io;
+
+use Psr\Http\Message\UriInterface;
+use React\EventLoop\LoopInterface;
+use React\EventLoop\TimerInterface;
+use React\Promise\PromiseInterface;
+use React\Socket\ConnectionInterface;
+use React\Socket\ConnectorInterface;
+
+/**
+ * [Internal] Manages outgoing HTTP connections for the HTTP client
+ *
+ * @internal
+ * @final
+ */
+class ClientConnectionManager
+{
+    /** @var ConnectorInterface */
+    private $connector;
+
+    /** @var LoopInterface */
+    private $loop;
+
+    /** @var string[] */
+    private $idleUris = array();
+
+    /** @var ConnectionInterface[] */
+    private $idleConnections = array();
+
+    /** @var TimerInterface[] */
+    private $idleTimers = array();
+
+    /** @var \Closure[] */
+    private $idleStreamHandlers = array();
+
+    /** @var float */
+    private $maximumTimeToKeepAliveIdleConnection = 0.001;
+
+    public function __construct(ConnectorInterface $connector, LoopInterface $loop)
+    {
+        $this->connector = $connector;
+        $this->loop = $loop;
+    }
+
+    /**
+     * @return PromiseInterface<ConnectionInterface>
+     */
+    public function connect(UriInterface $uri)
+    {
+        $scheme = $uri->getScheme();
+        if ($scheme !== 'https' && $scheme !== 'http') {
+            return \React\Promise\reject(new \InvalidArgumentException(
+                'Invalid request URL given'
+            ));
+        }
+
+        $port = $uri->getPort();
+        if ($port === null) {
+            $port = $scheme === 'https' ? 443 : 80;
+        }
+        $uri = ($scheme === 'https' ? 'tls://' : '') . $uri->getHost() . ':' . $port;
+
+        // Reuse idle connection for same URI if available
+        foreach ($this->idleConnections as $id => $connection) {
+            if ($this->idleUris[$id] === $uri) {
+                assert($this->idleStreamHandlers[$id] instanceof \Closure);
+                $connection->removeListener('close', $this->idleStreamHandlers[$id]);
+                $connection->removeListener('data', $this->idleStreamHandlers[$id]);
+                $connection->removeListener('error', $this->idleStreamHandlers[$id]);
+
+                assert($this->idleTimers[$id] instanceof TimerInterface);
+                $this->loop->cancelTimer($this->idleTimers[$id]);
+                unset($this->idleUris[$id], $this->idleConnections[$id], $this->idleTimers[$id], $this->idleStreamHandlers[$id]);
+
+                return \React\Promise\resolve($connection);
+            }
+        }
+
+        // Create new connection if no idle connection to same URI is available
+        return $this->connector->connect($uri);
+    }
+
+    /**
+     * Hands back an idle connection to the connection manager for possible future reuse.
+     *
+     * @return void
+     */
+    public function keepAlive(UriInterface $uri, ConnectionInterface $connection)
+    {
+        $scheme = $uri->getScheme();
+        assert($scheme === 'https' || $scheme === 'http');
+
+        $port = $uri->getPort();
+        if ($port === null) {
+            $port = $scheme === 'https' ? 443 : 80;
+        }
+
+        $this->idleUris[] = ($scheme === 'https' ? 'tls://' : '') . $uri->getHost() . ':' . $port;
+        $this->idleConnections[] = $connection;
+
+        $that = $this;
+        $cleanUp = function () use ($connection, $that) {
+            // call public method to support legacy PHP 5.3
+            $that->cleanUpConnection($connection);
+        };
+
+        // clean up and close connection when maximum time to keep-alive idle connection has passed
+        $this->idleTimers[] = $this->loop->addTimer($this->maximumTimeToKeepAliveIdleConnection, $cleanUp);
+
+        // clean up and close connection when unexpected close/data/error event happens during idle time
+        $this->idleStreamHandlers[] = $cleanUp;
+        $connection->on('close', $cleanUp);
+        $connection->on('data', $cleanUp);
+        $connection->on('error', $cleanUp);
+    }
+
+    /**
+     * @internal
+     * @return void
+     */
+    public function cleanUpConnection(ConnectionInterface $connection) // private (PHP 5.4+)
+    {
+        $id = \array_search($connection, $this->idleConnections, true);
+        if ($id === false) {
+            return;
+        }
+
+        assert(\is_int($id));
+        assert($this->idleTimers[$id] instanceof TimerInterface);
+        $this->loop->cancelTimer($this->idleTimers[$id]);
+        unset($this->idleUris[$id], $this->idleConnections[$id], $this->idleTimers[$id], $this->idleStreamHandlers[$id]);
+
+        $connection->close();
+    }
+}
diff --git a/src/Io/ClientRequestStream.php b/src/Io/ClientRequestStream.php
index bdaa54f1..0220f008 100644
--- a/src/Io/ClientRequestStream.php
+++ b/src/Io/ClientRequestStream.php
@@ -3,12 +3,10 @@
 namespace React\Http\Io;
 
 use Evenement\EventEmitter;
+use Psr\Http\Message\MessageInterface;
 use Psr\Http\Message\RequestInterface;
-use Psr\Http\Message\ResponseInterface;
 use React\Http\Message\Response;
-use React\Promise;
 use React\Socket\ConnectionInterface;
-use React\Socket\ConnectorInterface;
 use React\Stream\WritableStreamInterface;
 use RingCentral\Psr7 as gPsr;
 
@@ -26,8 +24,8 @@ class ClientRequestStream extends EventEmitter implements WritableStreamInterfac
     const STATE_HEAD_WRITTEN = 2;
     const STATE_END = 3;
 
-    /** @var ConnectorInterface */
-    private $connector;
+    /** @var ClientConnectionManager */
+    private $connectionManager;
 
     /** @var RequestInterface */
     private $request;
@@ -44,9 +42,9 @@ class ClientRequestStream extends EventEmitter implements WritableStreamInterfac
 
     private $pendingWrites = '';
 
-    public function __construct(ConnectorInterface $connector, RequestInterface $request)
+    public function __construct(ClientConnectionManager $connectionManager, RequestInterface $request)
     {
-        $this->connector = $connector;
+        $this->connectionManager = $connectionManager;
         $this->request = $request;
     }
 
@@ -65,7 +63,7 @@ private function writeHead()
         $pendingWrites = &$this->pendingWrites;
         $that = $this;
 
-        $promise = $this->connect();
+        $promise = $this->connectionManager->connect($this->request->getUri());
         $promise->then(
             function (ConnectionInterface $connection) use ($request, &$connectionRef, &$stateRef, &$pendingWrites, $that) {
                 $connectionRef = $connection;
@@ -174,11 +172,20 @@ public function handleData($data)
             $this->connection = null;
             $this->buffer = '';
 
-            // take control over connection handling and close connection once response body closes
+            // take control over connection handling and check if we can reuse the connection once response body closes
             $that = $this;
+            $request = $this->request;
+            $connectionManager = $this->connectionManager;
+            $successfulEndReceived = false;
             $input = $body = new CloseProtectionStream($connection);
-            $input->on('close', function () use ($connection, $that) {
-                $connection->close();
+            $input->on('close', function () use ($connection, $that, $connectionManager, $request, $response, &$successfulEndReceived) {
+                // only reuse connection after successful response and both request and response allow keep alive
+                if ($successfulEndReceived && $connection->isReadable() && $that->hasMessageKeepAliveEnabled($response) && $that->hasMessageKeepAliveEnabled($request)) {
+                    $connectionManager->keepAlive($request->getUri(), $connection);
+                } else {
+                    $connection->close();
+                }
+
                 $that->close();
             });
 
@@ -193,6 +200,9 @@ public function handleData($data)
                 $length = (int) $response->getHeaderLine('Content-Length');
             }
             $response = $response->withBody($body = new ReadableBodyStream($body, $length));
+            $body->on('end', function () use (&$successfulEndReceived) {
+                $successfulEndReceived = true;
+            });
 
             // emit response with streaming response body (see `Sender`)
             $this->emit('response', array($response, $body));
@@ -253,27 +263,28 @@ public function close()
         $this->removeAllListeners();
     }
 
-    protected function connect()
+    /**
+     * @internal
+     * @return bool
+     * @link https://www.rfc-editor.org/rfc/rfc9112#section-9.3
+     * @link https://www.rfc-editor.org/rfc/rfc7230#section-6.1
+     */
+    public function hasMessageKeepAliveEnabled(MessageInterface $message)
     {
-        $scheme = $this->request->getUri()->getScheme();
-        if ($scheme !== 'https' && $scheme !== 'http') {
-            return Promise\reject(
-                new \InvalidArgumentException('Invalid request URL given')
-            );
-        }
+        $connectionOptions = \RingCentral\Psr7\normalize_header(\strtolower($message->getHeaderLine('Connection')));
 
-        $host = $this->request->getUri()->getHost();
-        $port = $this->request->getUri()->getPort();
+        if (\in_array('close', $connectionOptions, true)) {
+            return false;
+        }
 
-        if ($scheme === 'https') {
-            $host = 'tls://' . $host;
+        if ($message->getProtocolVersion() === '1.1') {
+            return true;
         }
 
-        if ($port === null) {
-            $port = $scheme === 'https' ? 443 : 80;
+        if (\in_array('keep-alive', $connectionOptions, true)) {
+            return true;
         }
 
-        return $this->connector
-            ->connect($host . ':' . $port);
+        return false;
     }
 }
diff --git a/src/Io/Sender.php b/src/Io/Sender.php
index acbb6e7d..3598d31a 100644
--- a/src/Io/Sender.php
+++ b/src/Io/Sender.php
@@ -8,6 +8,7 @@
 use React\Http\Client\Client as HttpClient;
 use React\Promise\PromiseInterface;
 use React\Promise\Deferred;
+use React\Socket\Connector;
 use React\Socket\ConnectorInterface;
 use React\Stream\ReadableStreamInterface;
 
@@ -49,7 +50,11 @@ class Sender
      */
     public static function createFromLoop(LoopInterface $loop, ConnectorInterface $connector = null)
     {
-        return new self(new HttpClient($loop, $connector));
+        if ($connector === null) {
+            $connector = new Connector(array(), $loop);
+        }
+
+        return new self(new HttpClient(new ClientConnectionManager($connector, $loop)));
     }
 
     private $http;
@@ -93,13 +98,6 @@ public function send(RequestInterface $request)
             $size = 0;
         }
 
-        // automatically add `Connection: close` request header for HTTP/1.1 requests to avoid connection reuse
-        if ($request->getProtocolVersion() === '1.1') {
-            $request = $request->withHeader('Connection', 'close');
-        } else {
-            $request = $request->withoutHeader('Connection');
-        }
-
         // automatically add `Authorization: Basic …` request header if URL includes `user:pass@host`
         if ($request->getUri()->getUserInfo() !== '' && !$request->hasHeader('Authorization')) {
             $request = $request->withHeader('Authorization', 'Basic ' . \base64_encode($request->getUri()->getUserInfo()));
diff --git a/src/Io/Transaction.php b/src/Io/Transaction.php
index bfa42241..b93c490c 100644
--- a/src/Io/Transaction.php
+++ b/src/Io/Transaction.php
@@ -302,7 +302,7 @@ private function makeRedirectRequest(RequestInterface $request, UriInterface $lo
                 ->withMethod($request->getMethod() === 'HEAD' ? 'HEAD' : 'GET')
                 ->withoutHeader('Content-Type')
                 ->withoutHeader('Content-Length')
-                ->withBody(new EmptyBodyStream());
+                ->withBody(new BufferedBody(''));
         }
 
         return $request;
diff --git a/tests/BrowserTest.php b/tests/BrowserTest.php
index ad61cf9b..d01de9c5 100644
--- a/tests/BrowserTest.php
+++ b/tests/BrowserTest.php
@@ -60,9 +60,13 @@ public function testConstructWithConnectorAssignsGivenConnector()
         $ref->setAccessible(true);
         $client = $ref->getValue($sender);
 
-        $ref = new \ReflectionProperty($client, 'connector');
+        $ref = new \ReflectionProperty($client, 'connectionManager');
         $ref->setAccessible(true);
-        $ret = $ref->getValue($client);
+        $connectionManager = $ref->getValue($client);
+
+        $ref = new \ReflectionProperty($connectionManager, 'connector');
+        $ref->setAccessible(true);
+        $ret = $ref->getValue($connectionManager);
 
         $this->assertSame($connector, $ret);
     }
@@ -85,9 +89,13 @@ public function testConstructWithConnectorWithLegacySignatureAssignsGivenConnect
         $ref->setAccessible(true);
         $client = $ref->getValue($sender);
 
-        $ref = new \ReflectionProperty($client, 'connector');
+        $ref = new \ReflectionProperty($client, 'connectionManager');
+        $ref->setAccessible(true);
+        $connectionManager = $ref->getValue($client);
+
+        $ref = new \ReflectionProperty($connectionManager, 'connector');
         $ref->setAccessible(true);
-        $ret = $ref->getValue($client);
+        $ret = $ref->getValue($connectionManager);
 
         $this->assertSame($connector, $ret);
     }
@@ -548,6 +556,8 @@ public function testWithMultipleHeadersShouldBeMergedCorrectlyWithMultipleDefaul
                 'user-Agent' => array('ABC'),
                 'another-header' => array('value'),
                 'custom-header' => array('data'),
+
+                'Connection' => array('close')
             );
 
             $that->assertEquals($expectedHeaders, $request->getHeaders());
@@ -576,6 +586,32 @@ public function testWithoutHeaderShouldRemoveExistingHeader()
         $this->browser->get('http://example.com/');
     }
 
+    public function testWithoutHeaderConnectionShouldRemoveDefaultConnectionHeader()
+    {
+        $this->browser = $this->browser->withoutHeader('Connection');
+
+        $that = $this;
+        $this->sender->expects($this->once())->method('send')->with($this->callback(function (RequestInterface $request) use ($that) {
+            $that->assertEquals(array(), $request->getHeader('Connection'));
+            return true;
+        }))->willReturn(new Promise(function () { }));
+
+        $this->browser->get('http://example.com/');
+    }
+
+    public function testWithHeaderConnectionShouldOverwriteDefaultConnectionHeader()
+    {
+        $this->browser = $this->browser->withHeader('Connection', 'keep-alive');
+
+        $that = $this;
+        $this->sender->expects($this->once())->method('send')->with($this->callback(function (RequestInterface $request) use ($that) {
+            $that->assertEquals(array('keep-alive'), $request->getHeader('Connection'));
+            return true;
+        }))->willReturn(new Promise(function () { }));
+
+        $this->browser->get('http://example.com/');
+    }
+
     public function testBrowserShouldSendDefaultUserAgentHeader()
     {
         $that = $this;
diff --git a/tests/Client/FunctionalIntegrationTest.php b/tests/Client/FunctionalIntegrationTest.php
index 90d8444b..4925239c 100644
--- a/tests/Client/FunctionalIntegrationTest.php
+++ b/tests/Client/FunctionalIntegrationTest.php
@@ -5,10 +5,13 @@
 use Psr\Http\Message\ResponseInterface;
 use React\EventLoop\Loop;
 use React\Http\Client\Client;
+use React\Http\Io\ClientConnectionManager;
 use React\Http\Message\Request;
 use React\Promise\Deferred;
+use React\Promise\Promise;
 use React\Promise\Stream;
 use React\Socket\ConnectionInterface;
+use React\Socket\Connector;
 use React\Socket\SocketServer;
 use React\Stream\ReadableStreamInterface;
 use React\Tests\Http\TestCase;
@@ -45,7 +48,7 @@ public function testRequestToLocalhostEmitsSingleRemoteConnection()
         });
         $port = parse_url($socket->getAddress(), PHP_URL_PORT);
 
-        $client = new Client(Loop::get());
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
         $request = $client->request(new Request('GET', 'http://localhost:' . $port, array(), '', '1.0'));
 
         $promise = Stream\first($request, 'close');
@@ -54,6 +57,58 @@ public function testRequestToLocalhostEmitsSingleRemoteConnection()
         \React\Async\await(\React\Promise\Timer\timeout($promise, self::TIMEOUT_LOCAL));
     }
 
+    public function testRequestToLocalhostWillConnectAndCloseConnectionAfterResponseWhenKeepAliveTimesOut()
+    {
+        $socket = new SocketServer('127.0.0.1:0');
+        $socket->on('connection', $this->expectCallableOnce());
+
+        $promise = new Promise(function ($resolve) use ($socket) {
+            $socket->on('connection', function (ConnectionInterface $conn) use ($socket, $resolve) {
+                $conn->write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK");
+                $conn->on('close', function () use ($resolve) {
+                    $resolve(null);
+                });
+                $socket->close();
+            });
+        });
+        $port = parse_url($socket->getAddress(), PHP_URL_PORT);
+
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
+        $request = $client->request(new Request('GET', 'http://localhost:' . $port, array(), '', '1.1'));
+
+        $request->end();
+
+        \React\Async\await(\React\Promise\Timer\timeout($promise, self::TIMEOUT_LOCAL));
+    }
+
+    public function testRequestToLocalhostWillReuseExistingConnectionForSecondRequest()
+    {
+        $socket = new SocketServer('127.0.0.1:0');
+        $socket->on('connection', $this->expectCallableOnce());
+
+        $socket->on('connection', function (ConnectionInterface $connection) use ($socket) {
+            $connection->on('data', function () use ($connection) {
+                $connection->write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK");
+            });
+            $socket->close();
+        });
+        $port = parse_url($socket->getAddress(), PHP_URL_PORT);
+
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
+
+        $request = $client->request(new Request('GET', 'http://localhost:' . $port, array(), '', '1.1'));
+        $promise = Stream\first($request, 'close');
+        $request->end();
+
+        \React\Async\await(\React\Promise\Timer\timeout($promise, self::TIMEOUT_LOCAL));
+
+        $request = $client->request(new Request('GET', 'http://localhost:' . $port, array(), '', '1.1'));
+        $promise = Stream\first($request, 'close');
+        $request->end();
+
+        \React\Async\await(\React\Promise\Timer\timeout($promise, self::TIMEOUT_LOCAL));
+    }
+
     public function testRequestLegacyHttpServerWithOnlyLineFeedReturnsSuccessfulResponse()
     {
         $socket = new SocketServer('127.0.0.1:0');
@@ -62,7 +117,7 @@ public function testRequestLegacyHttpServerWithOnlyLineFeedReturnsSuccessfulResp
             $socket->close();
         });
 
-        $client = new Client(Loop::get());
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
         $request = $client->request(new Request('GET', str_replace('tcp:', 'http:', $socket->getAddress()), array(), '', '1.0'));
 
         $once = $this->expectCallableOnceWith('body');
@@ -82,7 +137,7 @@ public function testSuccessfulResponseEmitsEnd()
         // max_nesting_level was set to 100 for PHP Versions < 5.4 which resulted in failing test for legacy PHP
         ini_set('xdebug.max_nesting_level', 256);
 
-        $client = new Client(Loop::get());
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
 
         $request = $client->request(new Request('GET', 'http://www.google.com/', array(), '', '1.0'));
 
@@ -107,7 +162,7 @@ public function testPostDataReturnsData()
         // max_nesting_level was set to 100 for PHP Versions < 5.4 which resulted in failing test for legacy PHP
         ini_set('xdebug.max_nesting_level', 256);
 
-        $client = new Client(Loop::get());
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
 
         $data = str_repeat('.', 33000);
         $request = $client->request(new Request('POST', 'https://' . (mt_rand(0, 1) === 0 ? 'eu.' : '') . 'httpbin.org/post', array('Content-Length' => strlen($data)), '', '1.0'));
@@ -139,7 +194,7 @@ public function testPostJsonReturnsData()
             $this->markTestSkipped('Not supported on HHVM');
         }
 
-        $client = new Client(Loop::get());
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
 
         $data = json_encode(array('numbers' => range(1, 50)));
         $request = $client->request(new Request('POST', 'https://httpbin.org/post', array('Content-Length' => strlen($data), 'Content-Type' => 'application/json'), '', '1.0'));
@@ -169,7 +224,7 @@ public function testCancelPendingConnectionEmitsClose()
         // max_nesting_level was set to 100 for PHP Versions < 5.4 which resulted in failing test for legacy PHP
         ini_set('xdebug.max_nesting_level', 256);
 
-        $client = new Client(Loop::get());
+        $client = new Client(new ClientConnectionManager(new Connector(), Loop::get()));
 
         $request = $client->request(new Request('GET', 'http://www.google.com/', array(), '', '1.0'));
         $request->on('error', $this->expectCallableNever());
diff --git a/tests/FunctionalBrowserTest.php b/tests/FunctionalBrowserTest.php
index 6def2ecc..7ab909de 100644
--- a/tests/FunctionalBrowserTest.php
+++ b/tests/FunctionalBrowserTest.php
@@ -553,6 +553,60 @@ public function testReceiveStreamAndExplicitlyCloseConnectionEvenWhenServerKeeps
         $socket->close();
     }
 
+    public function testRequestWillCreateNewConnectionForSecondRequestByDefaultEvenWhenServerKeepsConnectionOpen()
+    {
+        $twice = $this->expectCallableOnce();
+        $socket = new SocketServer('127.0.0.1:0');
+        $socket->on('connection', function (\React\Socket\ConnectionInterface $connection) use ($socket, $twice) {
+            $connection->on('data', function () use ($connection) {
+                $connection->write("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello");
+            });
+
+            $socket->on('connection', $twice);
+            $socket->on('connection', function () use ($socket) {
+                $socket->close();
+            });
+        });
+
+        $this->base = str_replace('tcp:', 'http:', $socket->getAddress()) . '/';
+
+        $response = \React\Async\await($this->browser->get($this->base . 'get'));
+        assert($response instanceof ResponseInterface);
+        $this->assertEquals('hello', (string)$response->getBody());
+
+        $response = \React\Async\await($this->browser->get($this->base . 'get'));
+        assert($response instanceof ResponseInterface);
+        $this->assertEquals('hello', (string)$response->getBody());
+    }
+
+    public function testRequestWithoutConnectionHeaderWillReuseExistingConnectionForSecondRequest()
+    {
+        $this->socket->on('connection', $this->expectCallableOnce());
+
+        // remove default `Connection: close` request header to enable keep-alive
+        $this->browser = $this->browser->withoutHeader('Connection');
+
+        $response = \React\Async\await($this->browser->get($this->base . 'get'));
+        assert($response instanceof ResponseInterface);
+        $this->assertEquals('hello', (string)$response->getBody());
+
+        $response = \React\Async\await($this->browser->get($this->base . 'get'));
+        assert($response instanceof ResponseInterface);
+        $this->assertEquals('hello', (string)$response->getBody());
+    }
+
+    public function testRequestWithoutConnectionHeaderWillReuseExistingConnectionForRedirectedRequest()
+    {
+        $this->socket->on('connection', $this->expectCallableOnce());
+
+        // remove default `Connection: close` request header to enable keep-alive
+        $this->browser = $this->browser->withoutHeader('Connection');
+
+        $response = \React\Async\await($this->browser->get($this->base . 'redirect-to?url=get'));
+        assert($response instanceof ResponseInterface);
+        $this->assertEquals('hello', (string)$response->getBody());
+    }
+
     public function testPostStreamChunked()
     {
         $stream = new ThroughStream();
diff --git a/tests/Io/ClientConnectionManagerTest.php b/tests/Io/ClientConnectionManagerTest.php
new file mode 100644
index 00000000..b28c7964
--- /dev/null
+++ b/tests/Io/ClientConnectionManagerTest.php
@@ -0,0 +1,389 @@
+<?php
+
+namespace React\Tests\Http\Io;
+
+use RingCentral\Psr7\Uri;
+use React\Http\Io\ClientConnectionManager;
+use React\Promise\Promise;
+use React\Promise\PromiseInterface;
+use React\Tests\Http\TestCase;
+
+class ClientConnectionManagerTest extends TestCase
+{
+    public function testConnectWithHttpsUriShouldConnectToTlsWithDefaultPort()
+    {
+        $promise = new Promise(function () { });
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('tls://reactphp.org:443')->willReturn($promise);
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $ret = $connectionManager->connect(new Uri('https://reactphp.org/'));
+
+        assert($ret instanceof PromiseInterface);
+        $this->assertSame($promise, $ret);
+    }
+
+    public function testConnectWithHttpUriShouldConnectToTcpWithDefaultPort()
+    {
+        $promise = new Promise(function () { });
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('reactphp.org:80')->willReturn($promise);
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $ret = $connectionManager->connect(new Uri('http://reactphp.org/'));
+        $this->assertSame($promise, $ret);
+    }
+
+    public function testConnectWithExplicitPortShouldConnectWithGivenPort()
+    {
+        $promise = new Promise(function () { });
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('reactphp.org:8080')->willReturn($promise);
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $ret = $connectionManager->connect(new Uri('http://reactphp.org:8080/'));
+        $this->assertSame($promise, $ret);
+    }
+
+    public function testConnectWithInvalidSchemeShouldRejectWithException()
+    {
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->never())->method('connect');
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $promise = $connectionManager->connect(new Uri('ftp://reactphp.org/'));
+
+        $exception = null;
+        $promise->then(null, function ($reason) use (&$exception) {
+            $exception = $reason;
+        });
+
+        $this->assertInstanceOf('InvalidArgumentException', $exception);
+        $this->assertEquals('Invalid request URL given', $exception->getMessage());
+    }
+
+    public function testConnectWithoutSchemeShouldRejectWithException()
+    {
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->never())->method('connect');
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $promise = $connectionManager->connect(new Uri('reactphp.org'));
+
+        $exception = null;
+        $promise->then(null, function ($reason) use (&$exception) {
+            $exception = $reason;
+        });
+
+        $this->assertInstanceOf('InvalidArgumentException', $exception);
+        $this->assertEquals('Invalid request URL given', $exception->getMessage());
+    }
+
+    public function testConnectReusesIdleConnectionFromPreviousKeepAliveCallWithoutUsingConnectorAndWillAddAndRemoveStreamEventsAndAddAndCancelIdleTimer()
+    {
+        $connectionToReuse = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+
+        $streamHandler = null;
+        $connectionToReuse->expects($this->exactly(3))->method('on')->withConsecutive(
+            array(
+                'close',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    $streamHandler = $cb;
+                    return true;
+                })
+            ),
+            array(
+                'data',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            ),
+            array(
+                'error',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            )
+        );
+
+        $connectionToReuse->expects($this->exactly(3))->method('removeListener')->withConsecutive(
+            array(
+                'close',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            ),
+            array(
+                'data',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            ),
+            array(
+                'error',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            )
+        );
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->never())->method('connect');
+
+        $timer = $this->getMockBuilder('React\EventLoop\TimerInterface')->getMock();
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
+        $loop->expects($this->once())->method('cancelTimer')->with($timer);
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/'), $connectionToReuse);
+
+        $promise = $connectionManager->connect(new Uri('https://reactphp.org/'));
+        assert($promise instanceof PromiseInterface);
+
+        $connection = null;
+        $promise->then(function ($value) use (&$connection) {
+            $connection = $value;
+        });
+
+        $this->assertSame($connectionToReuse, $connection);
+    }
+
+    public function testConnectReusesIdleConnectionFromPreviousKeepAliveCallWithoutUsingConnectorAlsoWhenUriPathAndQueryAndFragmentIsDifferent()
+    {
+        $connectionToReuse = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->never())->method('connect');
+
+        $timer = $this->getMockBuilder('React\EventLoop\TimerInterface')->getMock();
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
+        $loop->expects($this->once())->method('cancelTimer')->with($timer);
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/http?foo#bar'), $connectionToReuse);
+
+        $promise = $connectionManager->connect(new Uri('https://reactphp.org/http/'));
+        assert($promise instanceof PromiseInterface);
+
+        $connection = null;
+        $promise->then(function ($value) use (&$connection) {
+            $connection = $value;
+        });
+
+        $this->assertSame($connectionToReuse, $connection);
+    }
+
+    public function testConnectUsesConnectorWithSameUriAndReturnsPromiseForNewConnectionFromConnectorWhenPreviousKeepAliveCallUsedDifferentUri()
+    {
+        $connectionToReuse = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+
+        $promise = new Promise(function () { });
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('tls://reactphp.org:443')->willReturn($promise);
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('http://reactphp.org/'), $connectionToReuse);
+
+        $ret = $connectionManager->connect(new Uri('https://reactphp.org/'));
+
+        assert($ret instanceof PromiseInterface);
+        $this->assertSame($promise, $ret);
+    }
+
+    public function testConnectUsesConnectorForNewConnectionWhenPreviousConnectReusedIdleConnectionFromPreviousKeepAliveCall()
+    {
+        $firstConnection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $secondConnection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('tls://reactphp.org:443')->willReturn(\React\Promise\resolve($secondConnection));
+
+        $timer = $this->getMockBuilder('React\EventLoop\TimerInterface')->getMock();
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
+        $loop->expects($this->once())->method('cancelTimer')->with($timer);
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/'), $firstConnection);
+
+        $promise = $connectionManager->connect(new Uri('https://reactphp.org/'));
+        assert($promise instanceof PromiseInterface);
+
+        $promise = $connectionManager->connect(new Uri('https://reactphp.org/'));
+        assert($promise instanceof PromiseInterface);
+
+        $connection = null;
+        $promise->then(function ($value) use (&$connection) {
+            $connection = $value;
+        });
+
+        $this->assertSame($secondConnection, $connection);
+    }
+
+    public function testKeepAliveAddsTimerAndDoesNotCloseConnectionImmediately()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->never())->method('close');
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->with(0.001, $this->anything());
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/'), $connection);
+    }
+
+    public function testKeepAliveClosesConnectionAfterIdleTimeout()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('close');
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+
+        $timerCallback = null;
+        $timer = $this->getMockBuilder('React\EventLoop\TimerInterface')->getMock();
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->with($this->anything(), $this->callback(function ($cb) use (&$timerCallback) {
+            $timerCallback = $cb;
+            return true;
+        }))->willReturn($timer);
+        $loop->expects($this->once())->method('cancelTimer')->with($timer);
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/'), $connection);
+
+        // manually invoker timer function to emulate time has passed
+        $this->assertNotNull($timerCallback);
+        call_user_func($timerCallback); // $timerCallback() (PHP 5.4+)
+    }
+
+    public function testConnectUsesConnectorForNewConnectionWhenIdleConnectionFromPreviousKeepAliveCallHasAlreadyTimedOut()
+    {
+        $firstConnection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $firstConnection->expects($this->once())->method('close');
+
+        $secondConnection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $secondConnection->expects($this->never())->method('close');
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('tls://reactphp.org:443')->willReturn(\React\Promise\resolve($secondConnection));
+
+        $timerCallback = null;
+        $timer = $this->getMockBuilder('React\EventLoop\TimerInterface')->getMock();
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->with($this->anything(), $this->callback(function ($cb) use (&$timerCallback) {
+            $timerCallback = $cb;
+            return true;
+        }))->willReturn($timer);
+        $loop->expects($this->once())->method('cancelTimer')->with($timer);
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/'), $firstConnection);
+
+        // manually invoker timer function to emulate time has passed
+        $this->assertNotNull($timerCallback);
+        call_user_func($timerCallback); // $timerCallback() (PHP 5.4+)
+
+        $promise = $connectionManager->connect(new Uri('https://reactphp.org/'));
+        assert($promise instanceof PromiseInterface);
+
+        $connection = null;
+        $promise->then(function ($value) use (&$connection) {
+            $connection = $value;
+        });
+
+        $this->assertSame($secondConnection, $connection);
+    }
+
+    public function testConnectUsesConnectorForNewConnectionWhenIdleConnectionFromPreviousKeepAliveCallHasAlreadyFiredUnexpectedStreamEventBeforeIdleTimeoutThatClosesConnection()
+    {
+        $firstConnection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $firstConnection->expects($this->once())->method('close');
+
+        $streamHandler = null;
+        $firstConnection->expects($this->exactly(3))->method('on')->withConsecutive(
+            array(
+                'close',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    $streamHandler = $cb;
+                    return true;
+                })
+            ),
+            array(
+                'data',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            ),
+            array(
+                'error',
+                $this->callback(function ($cb) use (&$streamHandler) {
+                    assert($streamHandler instanceof \Closure);
+                    return $cb === $streamHandler;
+                })
+            )
+        );
+
+        $secondConnection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $secondConnection->expects($this->never())->method('close');
+
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->with('tls://reactphp.org:443')->willReturn(\React\Promise\resolve($secondConnection));
+
+        $timer = $this->getMockBuilder('React\EventLoop\TimerInterface')->getMock();
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
+        $loop->expects($this->once())->method('cancelTimer')->with($timer);
+
+        $connectionManager = new ClientConnectionManager($connector, $loop);
+
+        $connectionManager->keepAlive(new Uri('https://reactphp.org/'), $firstConnection);
+
+        // manually invoke connection close to emulate server closing idle connection before idle timeout
+        $this->assertNotNull($streamHandler);
+        call_user_func($streamHandler); // $streamHandler() (PHP 5.4+)
+
+        $promise = $connectionManager->connect(new Uri('https://reactphp.org/'));
+        assert($promise instanceof PromiseInterface);
+
+        $connection = null;
+        $promise->then(function ($value) use (&$connection) {
+            $connection = $value;
+        });
+
+        $this->assertSame($secondConnection, $connection);
+    }
+}
diff --git a/tests/Io/ClientRequestStreamTest.php b/tests/Io/ClientRequestStreamTest.php
index 93220d10..4649087a 100644
--- a/tests/Io/ClientRequestStreamTest.php
+++ b/tests/Io/ClientRequestStreamTest.php
@@ -3,6 +3,7 @@
 namespace React\Tests\Http\Io;
 
 use Psr\Http\Message\ResponseInterface;
+use RingCentral\Psr7\Uri;
 use React\Http\Io\ClientRequestStream;
 use React\Http\Message\Request;
 use React\Promise\Deferred;
@@ -13,26 +14,17 @@
 
 class ClientRequestStreamTest extends TestCase
 {
-    private $connector;
-
-    /**
-     * @before
-     */
-    public function setUpStream()
-    {
-        $this->connector = $this->getMockBuilder('React\Socket\ConnectorInterface')
-            ->getMock();
-    }
-
     /** @test */
-    public function requestShouldBindToStreamEventsAndUseconnector()
+    public function testRequestShouldUseConnectionManagerWithUriFromRequestAndBindToStreamEvents()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $uri = new Uri('http://www.example.com');
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->with($uri)->willReturn(\React\Promise\resolve($connection));
 
-        $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $requestData = new Request('GET', $uri);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $connection->expects($this->atLeast(5))->method('on')->withConsecutive(
             array('drain', $this->identicalTo(array($request, 'handleDrain'))),
@@ -57,26 +49,14 @@ public function requestShouldBindToStreamEventsAndUseconnector()
         $request->handleData("\r\nbody");
     }
 
-    /**
-     * @test
-     */
-    public function requestShouldConnectViaTlsIfUrlUsesHttpsScheme()
-    {
-        $this->connector->expects($this->once())->method('connect')->with('tls://www.example.com:443')->willReturn(new Promise(function () { }));
-
-        $requestData = new Request('GET', 'https://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
-
-        $request->end();
-    }
-
     /** @test */
     public function requestShouldEmitErrorIfConnectionFails()
     {
-        $this->connector->expects($this->once())->method('connect')->willReturn(\React\Promise\reject(new \RuntimeException()));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\reject(new \RuntimeException()));
 
         $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->on('error', $this->expectCallableOnceWith($this->isInstanceOf('RuntimeException')));
         $request->on('close', $this->expectCallableOnce());
@@ -89,10 +69,11 @@ public function requestShouldEmitErrorIfConnectionClosesBeforeResponseIsParsed()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->on('error', $this->expectCallableOnceWith($this->isInstanceOf('RuntimeException')));
         $request->on('close', $this->expectCallableOnce());
@@ -106,10 +87,11 @@ public function requestShouldEmitErrorIfConnectionEmitsError()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->on('error', $this->expectCallableOnceWith($this->isInstanceOf('Exception')));
         $request->on('close', $this->expectCallableOnce());
@@ -123,10 +105,11 @@ public function requestShouldEmitErrorIfRequestParserThrowsException()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->on('error', $this->expectCallableOnceWith($this->isInstanceOf('InvalidArgumentException')));
         $request->on('close', $this->expectCallableOnce());
@@ -135,48 +118,17 @@ public function requestShouldEmitErrorIfRequestParserThrowsException()
         $request->handleData("\r\n\r\n");
     }
 
-    /**
-     * @test
-     */
-    public function requestShouldEmitErrorIfUrlIsInvalid()
-    {
-        $this->connector->expects($this->never())->method('connect');
-
-        $requestData = new Request('GET', 'ftp://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
-
-        $request->on('error', $this->expectCallableOnceWith($this->isInstanceOf('InvalidArgumentException')));
-        $request->on('close', $this->expectCallableOnce());
-
-        $request->end();
-    }
-
-    /**
-     * @test
-     */
-    public function requestShouldEmitErrorIfUrlHasNoScheme()
-    {
-        $this->connector->expects($this->never())->method('connect');
-
-        $requestData = new Request('GET', 'www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
-
-        $request->on('error', $this->expectCallableOnceWith($this->isInstanceOf('InvalidArgumentException')));
-        $request->on('close', $this->expectCallableOnce());
-
-        $request->end();
-    }
-
     /** @test */
     public function getRequestShouldSendAGetRequest()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.0\r\nHost: www.example.com\r\n\r\n");
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.0');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->end();
     }
@@ -187,10 +139,11 @@ public function getHttp11RequestShouldSendAGetRequestWithGivenConnectionCloseHea
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->end();
     }
@@ -201,11 +154,12 @@ public function getOptionsAsteriskShouldSendAOptionsRequestAsteriskRequestTarget
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
         $connection->expects($this->once())->method('write')->with("OPTIONS * HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('OPTIONS', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
         $requestData = $requestData->withRequestTarget('*');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->end();
     }
@@ -216,10 +170,11 @@ public function testStreamShouldEmitResponseWithEmptyBodyWhenResponseContainsCon
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->once())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -240,10 +195,11 @@ public function testStreamShouldEmitResponseWithEmptyBodyWhenResponseContainsSta
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->once())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -264,10 +220,11 @@ public function testStreamShouldEmitResponseWithEmptyBodyWhenResponseContainsSta
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->once())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -288,10 +245,11 @@ public function testStreamShouldEmitResponseWithEmptyBodyWhenRequestMethodIsHead
         $connection->expects($this->once())->method('write')->with("HEAD / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->once())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('HEAD', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -312,10 +270,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyUntilEndWhenRespons
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->once())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -336,10 +295,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyWithoutDataWhenResp
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->never())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -360,10 +320,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyWithDataWithoutEndW
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->never())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -384,10 +345,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyUntilEndWhenRespons
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->once())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -408,10 +370,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyWithoutDataWhenResp
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->never())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -432,10 +395,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyWithDataWithoutEndW
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->never())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -456,10 +420,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyWithDataWithoutEndW
         $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n");
         $connection->expects($this->never())->method('close');
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -492,10 +457,11 @@ public function testStreamShouldEmitResponseWithStreamingBodyUntilEndWhenRespons
             return true;
         }));
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'close'), '', '1.1');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $that = $this;
         $request->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($that) {
@@ -513,16 +479,251 @@ public function testStreamShouldEmitResponseWithStreamingBodyUntilEndWhenRespons
         call_user_func($endEvent); // $endEvent() (PHP 5.4+)
     }
 
+    public function testStreamShouldReuseConnectionForHttp11ByDefault()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->never())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager->expects($this->once())->method('keepAlive')->with(new Uri('http://www.example.com'), $connection);
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n");
+    }
+
+    public function testStreamShouldNotReuseConnectionWhenResponseContainsConnectionClose()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->once())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+    }
+
+    public function testStreamShouldNotReuseConnectionWhenRequestContainsConnectionCloseWithAdditionalOptions()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: FOO, CLOSE, BAR\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->once())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'FOO, CLOSE, BAR'), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: Foo, Close, Bar\r\n\r\n");
+    }
+
+    public function testStreamShouldNotReuseConnectionForHttp10ByDefault()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.0\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->once())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.0');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.0 200 OK\r\nContent-Length: 0\r\n\r\n");
+    }
+
+    public function testStreamShouldReuseConnectionForHttp10WhenBothRequestAndResponseContainConnectionKeepAlive()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.0\r\nHost: www.example.com\r\nConnection: keep-alive\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->never())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager->expects($this->once())->method('keepAlive')->with(new Uri('http://www.example.com'), $connection);
+
+        $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'keep-alive'), '', '1.0');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.0 200 OK\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n");
+    }
+
+    public function testStreamShouldReuseConnectionForHttp10WhenBothRequestAndResponseContainConnectionKeepAliveWithAdditionalOptions()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.0\r\nHost: www.example.com\r\nConnection: FOO, KEEP-ALIVE, BAR\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->never())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager->expects($this->once())->method('keepAlive')->with(new Uri('http://www.example.com'), $connection);
+
+        $requestData = new Request('GET', 'http://www.example.com', array('Connection' => 'FOO, KEEP-ALIVE, BAR'), '', '1.0');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.0 200 OK\r\nContent-Length: 0\r\nConnection: Foo, Keep-Alive, Bar\r\n\r\n");
+    }
+
+    public function testStreamShouldNotReuseConnectionWhenResponseContainsNoContentLengthAndResponseBodyTerminatedByConnectionEndEvent()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(false);
+        $connection->expects($this->once())->method('close');
+
+        $endEvent = null;
+        $eventName = null;
+        $connection->expects($this->any())->method('on')->with($this->callback(function ($name) use (&$eventName) {
+            $eventName = $name;
+            return true;
+        }), $this->callback(function ($cb) use (&$endEvent, &$eventName) {
+            if ($eventName === 'end') {
+                $endEvent = $cb;
+            }
+            return true;
+        }));
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\n\r\n");
+
+        $this->assertNotNull($endEvent);
+        call_user_func($endEvent); // $endEvent() (PHP 5.4+)
+    }
+
+    public function testStreamShouldNotReuseConnectionWhenResponseContainsContentLengthButIsTerminatedByUnexpectedCloseEvent()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->atMost(1))->method('isReadable')->willReturn(false);
+        $connection->expects($this->once())->method('close');
+
+        $closeEvent = null;
+        $eventName = null;
+        $connection->expects($this->any())->method('on')->with($this->callback(function ($name) use (&$eventName) {
+            $eventName = $name;
+            return true;
+        }), $this->callback(function ($cb) use (&$closeEvent, &$eventName) {
+            if ($eventName === 'close') {
+                $closeEvent = $cb;
+            }
+            return true;
+        }));
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n");
+
+        $this->assertNotNull($closeEvent);
+        call_user_func($closeEvent); // $closeEvent() (PHP 5.4+)
+    }
+
+    public function testStreamShouldReuseConnectionWhenResponseContainsTransferEncodingChunkedAndResponseBody()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->once())->method('isReadable')->willReturn(true);
+        $connection->expects($this->never())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager->expects($this->once())->method('keepAlive')->with(new Uri('http://www.example.com'), $connection);
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nOK\r\n0\r\n\r\n");
+    }
+
+    public function testStreamShouldNotReuseConnectionWhenResponseContainsTransferEncodingChunkedAndResponseBodyContainsInvalidData()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('write')->with("GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
+        $connection->expects($this->atMost(1))->method('isReadable')->willReturn(true);
+        $connection->expects($this->once())->method('close');
+
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $requestData = new Request('GET', 'http://www.example.com', array(), '', '1.1');
+        $request = new ClientRequestStream($connectionManager, $requestData);
+
+        $request->on('close', $this->expectCallableOnce());
+
+        $request->end();
+
+        $request->handleData("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nINVALID\r\n");
+    }
+
     /** @test */
     public function postRequestShouldSendAPostRequest()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
         $connection->expects($this->once())->method('write')->with($this->matchesRegularExpression("#^POST / HTTP/1\.0\r\nHost: www.example.com\r\n\r\nsome post data$#"));
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('POST', 'http://www.example.com', array(), '', '1.0');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->end('some post data');
 
@@ -541,10 +742,11 @@ public function writeWithAPostRequestShouldSendToTheStream()
             array($this->identicalTo("data"))
         );
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('POST', 'http://www.example.com', array(), '', '1.0');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->write("some");
         $request->write("post");
@@ -567,10 +769,11 @@ public function writeWithAPostRequestShouldSendBodyAfterHeadersAndEmitDrainEvent
         );
 
         $deferred = new Deferred();
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn($deferred->promise());
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn($deferred->promise());
 
         $requestData = new Request('POST', 'http://www.example.com', array(), '', '1.0');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $this->assertFalse($request->write("some"));
         $this->assertFalse($request->write("post"));
@@ -604,10 +807,11 @@ public function writeWithAPostRequestShouldForwardDrainEventIfFirstChunkExceedsB
         );
 
         $deferred = new Deferred();
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn($deferred->promise());
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn($deferred->promise());
 
         $requestData = new Request('POST', 'http://www.example.com', array(), '', '1.0');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $this->assertFalse($request->write("some"));
         $this->assertFalse($request->write("post"));
@@ -636,10 +840,11 @@ public function pipeShouldPipeDataIntoTheRequestBody()
             array($this->identicalTo("data"))
         );
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('POST', 'http://www.example.com', array(), '', '1.0');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $loop = $this
             ->getMockBuilder('React\EventLoop\LoopInterface')
@@ -663,13 +868,11 @@ public function pipeShouldPipeDataIntoTheRequestBody()
      */
     public function writeShouldStartConnecting()
     {
-        $this->connector->expects($this->once())
-                        ->method('connect')
-                        ->with('www.example.com:80')
-                        ->willReturn(new Promise(function () { }));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(new Promise(function () { }));
 
         $requestData = new Request('POST', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->write('test');
     }
@@ -679,10 +882,11 @@ public function writeShouldStartConnecting()
      */
     public function endShouldStartConnectingAndChangeStreamIntoNonWritableMode()
     {
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(new Promise(function () { }));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(new Promise(function () { }));
 
         $requestData = new Request('POST', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->end();
 
@@ -694,8 +898,10 @@ public function endShouldStartConnectingAndChangeStreamIntoNonWritableMode()
      */
     public function closeShouldEmitCloseEvent()
     {
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+
         $requestData = new Request('POST', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->on('close', $this->expectCallableOnce());
         $request->close();
@@ -706,8 +912,10 @@ public function closeShouldEmitCloseEvent()
      */
     public function writeAfterCloseReturnsFalse()
     {
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+
         $requestData = new Request('POST', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->close();
 
@@ -720,10 +928,11 @@ public function writeAfterCloseReturnsFalse()
      */
     public function endAfterCloseIsNoOp()
     {
-        $this->connector->expects($this->never())->method('connect');
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->never())->method('connect');
 
         $requestData = new Request('POST', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->close();
         $request->end();
@@ -737,10 +946,11 @@ public function closeShouldCancelPendingConnectionAttempt()
         $promise = new Promise(function () {}, function () {
             throw new \RuntimeException();
         });
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn($promise);
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn($promise);
 
         $requestData = new Request('POST', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->end();
 
@@ -754,8 +964,10 @@ public function closeShouldCancelPendingConnectionAttempt()
     /** @test */
     public function requestShouldRemoveAllListenerAfterClosed()
     {
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+
         $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $request->on('close', function () {});
         $this->assertCount(1, $request->listeners('close'));
@@ -769,10 +981,11 @@ public function multivalueHeader()
     {
         $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
 
-        $this->connector->expects($this->once())->method('connect')->with('www.example.com:80')->willReturn(\React\Promise\resolve($connection));
+        $connectionManager = $this->getMockBuilder('React\Http\Io\ClientConnectionManager')->disableOriginalConstructor()->getMock();
+        $connectionManager->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
 
         $requestData = new Request('GET', 'http://www.example.com');
-        $request = new ClientRequestStream($this->connector, $requestData);
+        $request = new ClientRequestStream($connectionManager, $requestData);
 
         $response = null;
         $request->on('response', $this->expectCallableOnce());
diff --git a/tests/Io/SenderTest.php b/tests/Io/SenderTest.php
index 4ef06442..3c8c4761 100644
--- a/tests/Io/SenderTest.php
+++ b/tests/Io/SenderTest.php
@@ -4,6 +4,7 @@
 
 use Psr\Http\Message\RequestInterface;
 use React\Http\Client\Client as HttpClient;
+use React\Http\Io\ClientConnectionManager;
 use React\Http\Io\ReadableBodyStream;
 use React\Http\Io\Sender;
 use React\Http\Message\Request;
@@ -13,6 +14,7 @@
 
 class SenderTest extends TestCase
 {
+    /** @var \React\EventLoop\LoopInterface */
     private $loop;
 
     /**
@@ -35,7 +37,7 @@ public function testSenderRejectsInvalidUri()
         $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
         $connector->expects($this->never())->method('connect');
 
-        $sender = new Sender(new HttpClient($this->loop, $connector));
+        $sender = new Sender(new HttpClient(new ClientConnectionManager($connector, $this->loop)));
 
         $request = new Request('GET', 'www.google.com');
 
@@ -54,7 +56,7 @@ public function testSenderConnectorRejection()
         $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
         $connector->expects($this->once())->method('connect')->willReturn(Promise\reject(new \RuntimeException('Rejected')));
 
-        $sender = new Sender(new HttpClient($this->loop, $connector));
+        $sender = new Sender(new HttpClient(new ClientConnectionManager($connector, $this->loop)));
 
         $request = new Request('GET', 'http://www.google.com/');
 
@@ -288,62 +290,6 @@ public function testSendCustomMethodWithExplicitContentLengthZeroWillBePassedAsI
         $sender->send($request);
     }
 
-    /** @test */
-    public function getHttp10RequestShouldSendAGetRequestWithoutConnectionHeaderByDefault()
-    {
-        $client = $this->getMockBuilder('React\Http\Client\Client')->disableOriginalConstructor()->getMock();
-        $client->expects($this->once())->method('request')->with($this->callback(function (RequestInterface $request) {
-            return !$request->hasHeader('Connection');
-        }))->willReturn($this->getMockBuilder('React\Http\Io\ClientRequestStream')->disableOriginalConstructor()->getMock());
-
-        $sender = new Sender($client);
-
-        $request = new Request('GET', 'http://www.example.com', array(), '', '1.0');
-        $sender->send($request);
-    }
-
-    /** @test */
-    public function getHttp10RequestShouldSendAGetRequestWithoutConnectionHeaderEvenWhenConnectionKeepAliveHeaderIsSpecified()
-    {
-        $client = $this->getMockBuilder('React\Http\Client\Client')->disableOriginalConstructor()->getMock();
-        $client->expects($this->once())->method('request')->with($this->callback(function (RequestInterface $request) {
-            return !$request->hasHeader('Connection');
-        }))->willReturn($this->getMockBuilder('React\Http\Io\ClientRequestStream')->disableOriginalConstructor()->getMock());
-
-        $sender = new Sender($client);
-
-        $request = new Request('GET', 'http://www.example.com', array('Connection' => 'keep-alive'), '', '1.0');
-        $sender->send($request);
-    }
-
-    /** @test */
-    public function getHttp11RequestShouldSendAGetRequestWithConnectionCloseHeaderByDefault()
-    {
-        $client = $this->getMockBuilder('React\Http\Client\Client')->disableOriginalConstructor()->getMock();
-        $client->expects($this->once())->method('request')->with($this->callback(function (RequestInterface $request) {
-            return $request->getHeaderLine('Connection') === 'close';
-        }))->willReturn($this->getMockBuilder('React\Http\Io\ClientRequestStream')->disableOriginalConstructor()->getMock());
-
-        $sender = new Sender($client);
-
-        $request = new Request('GET', 'http://www.example.com', array(), '', '1.1');
-        $sender->send($request);
-    }
-
-    /** @test */
-    public function getHttp11RequestShouldSendAGetRequestWithConnectionCloseHeaderEvenWhenConnectionKeepAliveHeaderIsSpecified()
-    {
-        $client = $this->getMockBuilder('React\Http\Client\Client')->disableOriginalConstructor()->getMock();
-        $client->expects($this->once())->method('request')->with($this->callback(function (RequestInterface $request) {
-            return $request->getHeaderLine('Connection') === 'close';
-        }))->willReturn($this->getMockBuilder('React\Http\Io\ClientRequestStream')->disableOriginalConstructor()->getMock());
-
-        $sender = new Sender($client);
-
-        $request = new Request('GET', 'http://www.example.com', array('Connection' => 'keep-alive'), '', '1.1');
-        $sender->send($request);
-    }
-
     /** @test */
     public function getRequestWithUserAndPassShouldSendAGetRequestWithBasicAuthorizationHeader()
     {
@@ -381,7 +327,7 @@ public function testCancelRequestWillCancelConnector()
         $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
         $connector->expects($this->once())->method('connect')->willReturn($promise);
 
-        $sender = new Sender(new HttpClient($this->loop, $connector));
+        $sender = new Sender(new HttpClient(new ClientConnectionManager($connector, $this->loop)));
 
         $request = new Request('GET', 'http://www.google.com/');
 
@@ -404,7 +350,7 @@ public function testCancelRequestWillCloseConnection()
         $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
         $connector->expects($this->once())->method('connect')->willReturn(Promise\resolve($connection));
 
-        $sender = new Sender(new HttpClient($this->loop, $connector));
+        $sender = new Sender(new HttpClient(new ClientConnectionManager($connector, $this->loop)));
 
         $request = new Request('GET', 'http://www.google.com/');