Change Compressor and Decompressor to use more efficient streaming compression context (requires PHP 7+)

.travis.yml

@@ -5,9 +5,6 @@ dist: trusty
 
 matrix:
   include:
-    - php: 5.4
-    - php: 5.5
-    - php: 5.6
     - php: 7.0
     - php: 7.1
     - php: 7.2

README.md

@@ -15,7 +15,6 @@ supporting compression and decompression of GZIP, ZLIB and raw DEFLATE formats.
 * [Usage](#usage)
     * [Compressor](#compressor)
     * [Decompressor](#decompressor)
-    * [Inconsistencies](#inconsistencies)
 * [Install](#install)
 * [Tests](#tests)
 * [License](#license)
@@ -162,17 +161,6 @@ $input->pipe($decompressor)->pipe($filterBadWords)->pipe($output);
 For more details, see ReactPHP's
 [`DuplexStreamInterface`](https://github.com/reactphp/stream#duplexstreaminterface).
 
-### Inconsistencies
-
-The stream compression filters are not exactly the most commonly used features of PHP.
-As such, we've spotted some inconsistencies (or *bugs*) in different PHP versions.
-These inconsistencies exist in the underlying PHP engines and there's little we can do about this in this library.
-
-* All PHP versions: Decompressing invalid data does not emit any data (and does not raise an error)
-
-Our test suite contains several test cases that exhibit these issues.
-If you feel some test case is missing or outdated, we're happy to accept PRs! :)
-
 ## Install
 
 The recommended way to install this library is [through Composer](https://getcomposer.org).
@@ -188,9 +176,7 @@ $ composer require clue/zlib-react:^0.2.2
 See also the [CHANGELOG](CHANGELOG.md) for details about version upgrades.
 
 This project aims to run on any platform and thus does not require any PHP
-extensions besides `ext-zlib` and supports running on legacy PHP 5.4 through current
-PHP 7+.
-It's *highly recommended to use PHP 7+* for this project.
+extensions besides `ext-zlib` and supports running on current PHP 7+.
 
 The `ext-zlib` extension is required for handling the underlying data compression
 and decompression.
@@ -200,8 +186,8 @@ builds by default. If you're building PHP from source, you may have to
 [manually enable](https://www.php.net/manual/en/zlib.installation.php) it.
 
 We're committed to providing a smooth upgrade path for legacy setups.
-If you need to support legacy PHP 5.3 and legacy HHVM, you may want to check out
-the legacy `v0.2.x` release branch.
+If you need to support legacy PHP versions and legacy HHVM, you may want to
+check out the legacy `v0.2.x` release branch.
 This legacy release branch also provides an installation candidate that does not
 require `ext-zlib` during installation but uses runtime checks instead.
 

composer.json

@@ -11,9 +11,8 @@
         }
     ],
     "require": {
-        "php": ">=5.4",
+        "php": ">=7.0",
         "ext-zlib": "*",
-        "clue/stream-filter": "~1.3",
         "react/stream": "^1.0 || ^0.7 || ^0.6"
     },
     "require-dev": {

src/Compressor.php

@@ -31,18 +31,44 @@
  * For more details, see ReactPHP's
  * [`DuplexStreamInterface`](https://github.com/reactphp/stream#duplexstreaminterface).
  */
-final class Compressor extends ZlibFilterStream
+final class Compressor extends TransformStream
 {
+    /** @var ?resource */
+    private $context;
+
     /**
      * @param int $encoding ZLIB_ENCODING_GZIP, ZLIB_ENCODING_RAW or ZLIB_ENCODING_DEFLATE
      * @param int $level    optional compression level
      */
     public function __construct($encoding, $level = -1)
     {
-        parent::__construct(
-            Filter\fun('zlib.deflate', array('window' => $encoding, 'level' => $level))
-        );
+        $context = @deflate_init($encoding, ['level' => $level]);
+        if ($context === false) {
+            throw new \InvalidArgumentException('Unable to initialize compressor' . strstr(error_get_last()['message'], ':'));
+        }
+
+        $this->context = $context;
+    }
+
+    protected function transformData($chunk)
+    {
+        $ret = deflate_add($this->context, $chunk, ZLIB_NO_FLUSH);
+
+        if ($ret !== '') {
+            $this->emit('data', [$ret]);
+        }
+    }
+
+    protected function transformEnd($chunk)
+    {
+        $ret = deflate_add($this->context, $chunk, ZLIB_FINISH);
+        $this->context = null;
+
+        if ($ret !== '') {
+            $this->emit('data', [$ret]);
+        }
 
-        $this->emptyWrite = $encoding;
+        $this->emit('end');
+        $this->close();
     }
 }

src/Decompressor.php

@@ -31,15 +31,50 @@
  * For more details, see ReactPHP's
  * [`DuplexStreamInterface`](https://github.com/reactphp/stream#duplexstreaminterface).
  */
-final class Decompressor extends ZlibFilterStream
+final class Decompressor extends TransformStream
 {
+    /** @var ?resource */
+    private $context;
+
     /**
      * @param int $encoding ZLIB_ENCODING_GZIP, ZLIB_ENCODING_RAW or ZLIB_ENCODING_DEFLATE
      */
     public function __construct($encoding)
     {
-        parent::__construct(
-            Filter\fun('zlib.inflate', array('window' => $encoding))
-        );
+        $context = @inflate_init($encoding);
+        if ($context === false) {
+            throw new \InvalidArgumentException('Unable to initialize decompressor' . strstr(error_get_last()['message'], ':'));
+        }
+
+        $this->context = $context;
+    }
+
+    protected function transformData($chunk)
+    {
+        $ret = @inflate_add($this->context, $chunk);
+        if ($ret === false) {
+            throw new \RuntimeException('Unable to decompress' . strstr(error_get_last()['message'], ':'));
+        }
+
+        if ($ret !== '') {
+            $this->emit('data', [$ret]);
+        }
+    }
+
+    protected function transformEnd($chunk)
+    {
+        $ret = @inflate_add($this->context, $chunk, ZLIB_FINISH);
+        $this->context = null;
+
+        if ($ret === false) {
+            throw new \RuntimeException('Unable to decompress' . strstr(error_get_last()['message'], ':'));
+        }
+
+        if ($ret !== '') {
+            $this->emit('data', [$ret]);
+        }
+
+        $this->emit('end');
+        $this->close();
     }
 }

src/TransformStream.php

@@ -35,7 +35,8 @@ public function write($data)
 
             return true;
         } catch (Exception $e) {
-            $this->forwardError($e);
+            $this->emit('error', [$e]);
+            $this->close();
             return false;
         }
     }
@@ -53,7 +54,8 @@ public function end($data = null)
             }
             $this->transformEnd($data);
         } catch (Exception $e) {
-            $this->forwardError($e);
+            $this->emit('error', [$e]);
+            $this->close();
         }
     }
 
@@ -107,109 +109,49 @@ public function pipe(WritableStreamInterface $dest, array $options = array())
         return $dest;
     }
 
-    /**
-     * Forwards a single "data" event to the reading side of the stream
-     *
-     * This will emit an "data" event.
-     *
-     * If the stream is not readable, then this is a NO-OP.
-     *
-     * @param string $data
-     */
-    protected function forwardData($data)
-    {
-        if (!$this->readable) {
-            return;
-        }
-        $this->emit('data', array($data));
-    }
-
-    /**
-     * Forwards an "end" event to the reading side of the stream
-     *
-     * This will emit an "end" event and will then close this stream.
-     *
-     * If the stream is not readable, then this is a NO-OP.
-     *
-     * @uses self::close()
-     */
-    protected function forwardEnd()
-    {
-        if (!$this->readable) {
-            return;
-        }
-        $this->readable = false;
-        $this->writable = false;
-
-        $this->emit('end');
-        $this->close();
-    }
-
-    /**
-     * Forwards the given $error message to the reading side of the stream
-     *
-     * This will emit an "error" event and will then close this stream.
-     *
-     * If the stream is not readable, then this is a NO-OP.
-     *
-     * @param Exception $error
-     * @uses self::close()
-     */
-    protected function forwardError(Exception $error)
-    {
-        if (!$this->readable) {
-            return;
-        }
-        $this->readable = false;
-        $this->writable = false;
-
-        $this->emit('error', array($error));
-        $this->close();
-    }
-
     /**
      * can be overwritten in order to implement custom transformation behavior
      *
-     * This gets passed a single chunk of $data and should invoke `forwardData()`
+     * This gets passed a single chunk of $data and should emit a `data` event
      * with the filtered result.
      *
-     * If the given data chunk is not valid, then you should invoke `forwardError()`
-     * or throw an Exception.
+     * If the given data chunk is not valid, then you should throw an Exception
+     * which will automatically be turned into an `error` event.
      *
-     * If you do not overwrite this method, then its default implementation simply
-     * invokes `forwardData()` on the unmodified input data chunk.
+     * If you do not overwrite this method, then its default implementation
+     * simply emits a `data` event with the unmodified input data chunk.
      *
      * @param string $data
-     * @see self::forwardData()
      */
     protected function transformData($data)
     {
-        $this->forwardData($data);
+        $this->emit('data', [$data]);
     }
 
     /**
      * can be overwritten in order to implement custom stream ending behavior
      *
-     * This may get passed a single final chunk of $data and should invoke `forwardEnd()`.
+     * This may get passed a single final chunk of $data and should emit an
+     * `end` event and close the stream.
      *
-     * If the given data chunk is not valid, then you should invoke `forwardError()`
-     * or throw an Exception.
+     * If the given data chunk is not valid, then you should throw an Exception
+     * which will automatically be turned into an `error` event.
      *
      * If you do not overwrite this method, then its default implementation simply
      * invokes `transformData()` on the unmodified input data chunk (if any),
-     * which in turn defaults to invoking `forwardData()` and then finally
-     * invokes `forwardEnd()`.
+     * which in turn defaults to emitting a `data` event and then finally
+     * emits an `end` event and closes the stream.
      *
      * @param string $data
      * @see self::transformData()
-     * @see self::forwardData()
-     * @see self::forwardEnd()
      */
     protected function transformEnd($data)
     {
         if ($data !== '') {
             $this->transformData($data);
         }
-        $this->forwardEnd();
+
+        $this->emit('end');
+        $this->close();
     }
 }

tests/CompressorTest.php

@@ -0,0 +1,14 @@
+<?php
+
+use Clue\React\Zlib\Compressor;
+
+class CompressorTest extends TestCase
+{
+    /**
+     * @expectedException InvalidArgumentException
+     */
+    public function testCtorThrowsForInvalidEncoding()
+    {
+        new Compressor(0);
+    }
+}