Forward compatibility with Stream v1.0 and v0.7

README.md

@@ -637,9 +637,9 @@ know what you're doing.
 Internally, the `SecureServer` has to set the required TLS context options on
 the underlying stream resources.
 These resources are not exposed through any of the interfaces defined in this
-package, but only through the `React\Stream\Stream` class.
+package, but only through the internal `Connection` class.
 The `TcpServer` class is guaranteed to emit connections that implement
-the `ConnectionInterface` and also extend the `Stream` class in order to
+the `ConnectionInterface` and uses the internal `Connection` class in order to
 expose these underlying resources.
 If you use a custom `ServerInterface` and its `connection` event does not
 meet this requirement, the `SecureServer` will emit an `error` event and
@@ -1237,8 +1237,11 @@ options as described above.
 All versions of PHP prior to 5.6.8 suffered from a buffering issue where reading
 from a streaming TLS connection could be one `data` event behind.
 This library implements a work-around to try to flush the complete incoming
-data buffers on these versions, but we have seen reports of people saying this
-could still affect some older versions (`5.5.23`, `5.6.7`, and `5.6.8`).
+data buffers on these legacy PHP versions, which has a penalty of around 10% of
+throughput on all connections.
+With this work-around, we have not been able to reproduce this issue anymore,
+but we have seen reports of people saying this could still affect some of the
+older PHP versions (`5.5.23`, `5.6.7`, and `5.6.8`).
 Note that this only affects *some* higher-level streaming protocols, such as
 IRC over TLS, but should not affect HTTP over TLS (HTTPS).
 Further investigation of this issue is needed.

composer.json

@@ -8,14 +8,14 @@
         "evenement/evenement": "~2.0|~1.0",
         "react/dns": "0.4.*|0.3.*",
         "react/event-loop": "0.4.*|0.3.*",
-        "react/stream": "^0.6 || ^0.5 || ^0.4.5",
+        "react/stream": "^1.0 || ^0.7 || ^0.6 || ^0.5 || ^0.4.5",
         "react/promise": "^2.1 || ^1.2",
         "react/promise-timer": "~1.0"
     },
     "require-dev": {
         "clue/block-react": "^1.1",
         "phpunit/phpunit": "~4.8",
-        "react/stream": "^0.6"
+        "react/stream": "^1.0 || ^0.7 || ^0.6"
     },
     "autoload": {
         "psr-4": {

src/Connection.php

@@ -2,7 +2,12 @@
 
 namespace React\Socket;
 
+use Evenement\EventEmitter;
+use React\EventLoop\LoopInterface;
+use React\Stream\DuplexResourceStream;
 use React\Stream\Stream;
+use React\Stream\Util;
+use React\Stream\WritableStreamInterface;
 
 /**
  * The actual connection implementation for ConnectionInterface
@@ -12,7 +17,7 @@
  * @see ConnectionInterface
  * @internal
  */
-class Connection extends Stream implements ConnectionInterface
+class Connection extends EventEmitter implements ConnectionInterface
 {
     /**
      * Internal flag whether encryption has been enabled on this connection
@@ -24,6 +29,84 @@ class Connection extends Stream implements ConnectionInterface
      */
     public $encryptionEnabled = false;
 
+    /** @internal */
+    public $stream;
+
+    private $input;
+
+    public function __construct($resource, LoopInterface $loop)
+    {
+        // PHP < 5.6.8 suffers from a buffer indicator bug on secure TLS connections
+        // as a work-around we always read the complete buffer until its end.
+        // The buffer size is limited due to TCP/IP buffers anyway, so this
+        // should not affect usage otherwise.
+        // See https://bugs.php.net/bug.php?id=65137
+        // https://bugs.php.net/bug.php?id=41631
+        // https://github.com/reactphp/socket-client/issues/24
+        $clearCompleteBuffer = (version_compare(PHP_VERSION, '5.6.8', '<'));
+
+        // @codeCoverageIgnoreStart
+        if (class_exists('React\Stream\Stream')) {
+            // legacy react/stream < 0.7 requires additional buffer property
+            $this->input = new Stream($resource, $loop);
+            if ($clearCompleteBuffer) {
+                $this->input->bufferSize = null;
+            }
+        } else {
+            // preferred react/stream >= 0.7 accepts buffer parameter
+            $this->input = new DuplexResourceStream($resource, $loop, $clearCompleteBuffer ? -1 : null);
+        }
+        // @codeCoverageIgnoreEnd
+
+        $this->stream = $resource;
+
+        Util::forwardEvents($this->input, $this, array('data', 'end', 'error', 'close', 'pipe', 'drain'));
+
+        $this->input->on('close', array($this, 'close'));
+    }
+
+    public function isReadable()
+    {
+        return $this->input->isReadable();
+    }
+
+    public function isWritable()
+    {
+        return $this->input->isWritable();
+    }
+
+    public function pause()
+    {
+        $this->input->pause();
+    }
+
+    public function resume()
+    {
+        $this->input->resume();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        return $this->input->pipe($dest, $options);
+    }
+
+    public function write($data)
+    {
+        return $this->input->write($data);
+    }
+
+    public function end($data = null)
+    {
+        $this->input->end($data);
+    }
+
+    public function close()
+    {
+        $this->input->close();
+        $this->handleClose();
+        $this->removeAllListeners();
+    }
+
     public function handleClose()
     {
         if (!is_resource($this->stream)) {

src/SecureConnector.php

@@ -3,7 +3,6 @@
 namespace React\Socket;
 
 use React\EventLoop\LoopInterface;
-use React\Stream\Stream;
 use React\Promise;
 
 final class SecureConnector implements ConnectorInterface
@@ -41,9 +40,9 @@ public function connect($uri)
         return $this->connector->connect($uri)->then(function (ConnectionInterface $connection) use ($context, $encryption) {
             // (unencrypted) TCP/IP connection succeeded
 
-            if (!$connection instanceof Stream) {
+            if (!$connection instanceof Connection) {
                 $connection->close();
-                throw new \UnexpectedValueException('Connection MUST extend Stream in order to access underlying stream resource');
+                throw new \UnexpectedValueException('Base connector does not use internal Connection class exposing stream resource');
             }
 
             // set required SSL/TLS context options

src/SecureServer.php

@@ -6,7 +6,6 @@
 use React\EventLoop\LoopInterface;
 use React\Socket\TcpServer;
 use React\Socket\ConnectionInterface;
-use React\Stream\Stream;
 
 /**
  * The `SecureServer` class implements the `ServerInterface` and is responsible
@@ -101,9 +100,9 @@ final class SecureServer extends EventEmitter implements ServerInterface
      * Internally, the `SecureServer` has to set the required TLS context options on
      * the underlying stream resources.
      * These resources are not exposed through any of the interfaces defined in this
-     * package, but only through the `React\Stream\Stream` class.
+     * package, but only through the internal `Connection` class.
      * The `TcpServer` class is guaranteed to emit connections that implement
-     * the `ConnectionInterface` and also extend the `Stream` class in order to
+     * the `ConnectionInterface` and uses the internal `Connection` class in order to
      * expose these underlying resources.
      * If you use a custom `ServerInterface` and its `connection` event does not
      * meet this requirement, the `SecureServer` will emit an `error` event and
@@ -163,8 +162,8 @@ public function close()
     /** @internal */
     public function handleConnection(ConnectionInterface $connection)
     {
-        if (!$connection instanceof Stream) {
-            $this->emit('error', array(new \UnexpectedValueException('Connection event MUST emit an instance extending Stream in order to access underlying stream resource')));
+        if (!$connection instanceof Connection) {
+            $this->emit('error', array(new \UnexpectedValueException('Base server does not use internal Connection class exposing stream resource')));
             $connection->end();
             return;
         }

src/StreamEncryption.php

@@ -21,22 +21,11 @@ class StreamEncryption
     private $errstr;
     private $errno;
 
-    private $wrapSecure = false;
-
     public function __construct(LoopInterface $loop, $server = true)
     {
         $this->loop = $loop;
         $this->server = $server;
 
-        // See https://bugs.php.net/bug.php?id=65137
-        // https://bugs.php.net/bug.php?id=41631
-        // https://github.com/reactphp/socket-client/issues/24
-        // On versions affected by this bug we need to fread the stream until we
-        //  get an empty string back because the buffer indicator could be wrong
-        if (version_compare(PHP_VERSION, '5.6.8', '<')) {
-            $this->wrapSecure = true;
-        }
-
         if ($server) {
             $this->method = STREAM_CRYPTO_METHOD_TLS_SERVER;
 
@@ -100,16 +89,11 @@ public function toggle(Connection $stream, $toggle)
             $toggleCrypto();
         }
 
-        $wrap = $this->wrapSecure && $toggle;
         $loop = $this->loop;
 
-        return $deferred->promise()->then(function () use ($stream, $socket, $wrap, $loop, $toggle) {
+        return $deferred->promise()->then(function () use ($stream, $socket, $loop, $toggle) {
             $loop->removeReadStream($socket);
 
-            if ($wrap) {
-                $stream->bufferSize = null;
-            }
-
             $stream->encryptionEnabled = $toggle;
             $stream->resume();
 

tests/FunctionalSecureServerTest.php

@@ -3,7 +3,6 @@
 namespace React\Tests\Socket;
 
 use React\EventLoop\Factory;
-use React\Stream\Stream;
 use React\Socket\SecureServer;
 use React\Socket\ConnectionInterface;
 use React\Socket\TcpServer;
@@ -60,7 +59,7 @@ public function testWritesDataToConnection()
         $promise = $connector->connect($server->getAddress());
 
         $local = Block\await($promise, $loop, self::TIMEOUT);
-        /* @var $local React\Stream\Stream */
+        /* @var $local ConnectionInterface */
 
         $local->on('data', $this->expectCallableOnceWith('foo'));
 
@@ -346,7 +345,7 @@ public function testEmitsErrorIfConnectionIsNotSecureHandshake()
         $connector = new TcpConnector($loop);
         $promise = $connector->connect(str_replace('tls://', '', $server->getAddress()));
 
-        $promise->then(function (Stream $stream) {
+        $promise->then(function (ConnectionInterface $stream) {
             $stream->write("GET / HTTP/1.0\r\n\r\n");
         });
 

tests/IntegrationTest.php

@@ -7,7 +7,6 @@
 use React\Socket\Connector;
 use React\Socket\SecureConnector;
 use React\Socket\TcpConnector;
-use React\Stream\BufferedSink;
 use Clue\React\Block;
 use React\Socket\DnsConnector;
 
@@ -28,7 +27,7 @@ public function gettingStuffFromGoogleShouldWork()
 
         $conn->write("GET / HTTP/1.0\r\n\r\n");
 
-        $response = Block\await(BufferedSink::createPromise($conn), $loop, self::TIMEOUT);
+        $response = $this->buffer($conn, $loop, self::TIMEOUT);
 
         $this->assertRegExp('#^HTTP/1\.0#', $response);
     }
@@ -47,7 +46,7 @@ public function gettingEncryptedStuffFromGoogleShouldWork()
 
         $conn->write("GET / HTTP/1.0\r\n\r\n");
 
-        $response = Block\await(BufferedSink::createPromise($conn), $loop, self::TIMEOUT);
+        $response = $this->buffer($conn, $loop, self::TIMEOUT);
 
         $this->assertRegExp('#^HTTP/1\.0#', $response);
     }
@@ -76,7 +75,7 @@ public function gettingEncryptedStuffFromGoogleShouldWorkIfHostIsResolvedFirst()
 
         $conn->write("GET / HTTP/1.0\r\n\r\n");
 
-        $response = Block\await(BufferedSink::createPromise($conn), $loop, self::TIMEOUT);
+        $response = $this->buffer($conn, $loop, self::TIMEOUT);
 
         $this->assertRegExp('#^HTTP/1\.0#', $response);
     }
@@ -94,7 +93,7 @@ public function gettingPlaintextStuffFromEncryptedGoogleShouldNotWork()
 
         $conn->write("GET / HTTP/1.0\r\n\r\n");
 
-        $response = Block\await(BufferedSink::createPromise($conn), $loop, self::TIMEOUT);
+        $response = $this->buffer($conn, $loop, self::TIMEOUT);
 
         $this->assertNotRegExp('#^HTTP/1\.0#', $response);
     }