Support all streaming flush modes for Compressor

README.md

@@ -140,6 +140,25 @@ This is particularly useful in a piping context:
 $input->pipe($filterBadWords)->pipe($compressor)->pipe($output);
 ```
 
+This class takes an optional `int $level` parameter that controls the
+compression level from `0` (no compression) to `9` (best compression).
+It defaults to `-1` which uses zlib's default compression level.
+
+This class takes an optional `int $flush` parameter that controls when
+compressed data will be emitted. It defaults to `ZLIB_NO_FLUSH` which
+buffers data internally to achieve the best compression ratio, so compressed
+data is only emitted once enough data is available or the stream ends.
+For streaming protocols like EventSource (SSE), you can use `ZLIB_SYNC_FLUSH`
+to make sure each chunk can be decompressed by the receiving side immediately
+after writing. Each flush adds a few bytes, so this may produce larger output
+when writing many small chunks. This parameter also accepts `ZLIB_PARTIAL_FLUSH`
+and `ZLIB_FULL_FLUSH`, the latter also resets the compression state for each
+chunk, which may considerably increase output size.
+
+```php
+$compressor = new Clue\React\Zlib\Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_SYNC_FLUSH);
+```
+
 For more details, see ReactPHP's
 [`DuplexStreamInterface`](https://github.com/reactphp/stream#duplexstreaminterface).
 

src/Compressor.php

@@ -26,6 +26,25 @@
  * $input->pipe($filterBadWords)->pipe($compressor)->pipe($output);
  * ```
  *
+ * This class takes an optional `int $level` parameter that controls the
+ * compression level from `0` (no compression) to `9` (best compression).
+ * It defaults to `-1` which uses zlib's default compression level.
+ *
+ * This class takes an optional `int $flush` parameter that controls when
+ * compressed data will be emitted. It defaults to `ZLIB_NO_FLUSH` which
+ * buffers data internally to achieve the best compression ratio, so compressed
+ * data is only emitted once enough data is available or the stream ends.
+ * For streaming protocols like EventSource (SSE), you can use `ZLIB_SYNC_FLUSH`
+ * to make sure each chunk can be decompressed by the receiving side immediately
+ * after writing. Each flush adds a few bytes, so this may produce larger output
+ * when writing many small chunks. This parameter also accepts `ZLIB_PARTIAL_FLUSH`
+ * and `ZLIB_FULL_FLUSH`, the latter also resets the compression state for each
+ * chunk, which may considerably increase output size.
+ *
+ * ```php
+ * $compressor = new Clue\React\Zlib\Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_SYNC_FLUSH);
+ * ```
+ *
  * For more details, see ReactPHP's
  * [`DuplexStreamInterface`](https://github.com/reactphp/stream#duplexstreaminterface).
  */
@@ -40,7 +59,7 @@ final class Compressor extends TransformStream
     /**
      * @param int $encoding ZLIB_ENCODING_GZIP, ZLIB_ENCODING_RAW or ZLIB_ENCODING_DEFLATE
      * @param int $level    optional compression level
-     * @param int $flush    optional flush mode (ZLIB_NO_FLUSH, ZLIB_SYNC_FLUSH, ZLIB_FULL_FLUSH, ZLIB_FINISH)
+     * @param int $flush    optional flush mode (ZLIB_NO_FLUSH, ZLIB_PARTIAL_FLUSH, ZLIB_SYNC_FLUSH, ZLIB_FULL_FLUSH)
      */
     public function __construct($encoding, $level = -1, int $flush = ZLIB_NO_FLUSH)
     {
@@ -65,8 +84,8 @@ public function __construct($encoding, $level = -1, int $flush = ZLIB_NO_FLUSH)
             throw new \InvalidArgumentException('Unable to initialize compressor' . $errstr); // @codeCoverageIgnore
         }
 
-        if (!in_array($flush, [ZLIB_NO_FLUSH, ZLIB_SYNC_FLUSH, ZLIB_FULL_FLUSH, ZLIB_FINISH], true)) {
-            throw new \InvalidArgumentException('Argument #3 ($flush) must be one of ZLIB_NO_FLUSH, ZLIB_SYNC_FLUSH, ZLIB_FULL_FLUSH or ZLIB_FINISH');
+        if (!in_array($flush, [ZLIB_NO_FLUSH, ZLIB_PARTIAL_FLUSH, ZLIB_SYNC_FLUSH, ZLIB_FULL_FLUSH], true)) {
+            throw new \InvalidArgumentException('Argument #3 ($flush) must be one of ZLIB_NO_FLUSH, ZLIB_PARTIAL_FLUSH, ZLIB_SYNC_FLUSH or ZLIB_FULL_FLUSH');
         }
 
         $this->context = $context;

tests/CompressorTest.php

@@ -34,6 +34,13 @@ public function testCtorThrowsForInvalidEncodingAndUnsetsUsedErrorHandler()
     public function testCtorThrowsForInvalidFlushMode()
     {
         $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('Argument #3 ($flush) must be one of ZLIB_NO_FLUSH, ZLIB_PARTIAL_FLUSH, ZLIB_SYNC_FLUSH or ZLIB_FULL_FLUSH');
         new Compressor(ZLIB_ENCODING_GZIP, -1, -1);
     }
+
+    public function testCtorThrowsForFinishFlushMode()
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_FINISH);
+    }
 }

tests/GzipCompressorTest.php

@@ -67,30 +67,30 @@ public function testWriteWillOnlyFlushHeaderByDefaultToBufferDataBeforeFlushing(
         $compressor->write('hello');
     }
 
-    public function testWriteWithSyncFlushWillFlushHeaderWithFirstChunkImmediately()
+    public function testWriteWithPartialFlushWillFlushHeaderWithFirstChunkImmediately()
     {
-        $compressor = new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_SYNC_FLUSH);
+        $compressor = new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_PARTIAL_FLUSH);
 
-        $compressor->on('data', $this->expectCallableOnceWith("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00" . $this->os . "\xca\x48\xcd\xc9\xc9\x07\x00\x00\x00\xff\xff"));
+        $compressor->on('data', $this->expectCallableOnceWith("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00" . $this->os . "\xca\x48\xcd\xc9\xc9\x07\x08"));
 
         $compressor->write('hello');
     }
 
-    public function testWriteWithFinishFlushWillFlushEntireGzipHeaderAndFooterWithFirstChunkImmediately()
+    public function testWriteWithSyncFlushWillFlushHeaderWithFirstChunkImmediately()
     {
-        $compressor = new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_FINISH);
+        $compressor = new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_SYNC_FLUSH);
 
-        $compressor->on('data', $this->expectCallableOnceWith("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00" . $this->os . "\xcb\x48\xcd\xc9\xc9\x07\x00\x86\xa6\x10\x36" . "\x05\x00\x00\x00"));
+        $compressor->on('data', $this->expectCallableOnceWith("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00" . $this->os . "\xca\x48\xcd\xc9\xc9\x07\x00\x00\x00\xff\xff"));
 
         $compressor->write('hello');
     }
 
-    public function testWriteAfterFinishFlushWillFlushEntireGzipWithSyncFlushWillFlushEntireGzipHeaderAndFooterAgainImmediately()
+    public function testWriteAfterWriteWithFullFlushWillFlushChunkIndependentOfPreviousChunk()
     {
-        $compressor = new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_FINISH);
+        $compressor = new Compressor(ZLIB_ENCODING_GZIP, -1, ZLIB_FULL_FLUSH);
         $compressor->write('hello');
 
-        $compressor->on('data', $this->expectCallableOnceWith("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00" . $this->os . "\xcb\x48\xcd\xc9\xc9\x07\x00\x86\xa6\x10\x36" . "\x05\x00\x00\x00"));
+        $compressor->on('data', $this->expectCallableOnceWith("\xca\x48\xcd\xc9\xc9\x07\x00\x00\x00\xff\xff"));
 
         $compressor->write('hello');
     }