Strictly follow stream semantics

Author: clue

README.md

@@ -29,7 +29,7 @@ $stream = new ReadableResourceStream(fopen('archive.tar', 'r'), $loop);
 
 $decoder = new Decoder();
 
-$decoder->on('entry', function ($header, ReadableStreamInterface $file) {
+$decoder->on('entry', function (array $header, React\Stream\ReadableStreamInterface $file) {
     echo 'File ' . $header['filename'];
     echo ' (' . $header['size'] . ' bytes):' . PHP_EOL;
 

composer.json

@@ -12,13 +12,11 @@
     ],
     "require": {
         "php": ">=5.3",
-        "react/stream": "^1 || ^0.7 || ^0.6"
+        "react/stream": "^1.0 || ^0.7 || ^0.6"
     },
     "require-dev": {
         "clue/hexdump": "~0.2.0",
-        "react/event-loop": "^1",
-        "react/promise": "~2.0|~1.0",
-        "react/promise-stream": "^1",
+        "react/event-loop": "^1.0 || ^0.5 || ^0.4 || ^0.3",
         "phpunit/phpunit": "^7.0 || ^6.0 || ^5.0 || ^4.8.35"
     },
     "autoload": {

examples/dump.php

@@ -2,37 +2,35 @@
 
 use Clue\Hexdump\Hexdump;
 use Clue\React\Tar\Decoder;
-use React\EventLoop\StreamSelectLoop;
+use React\EventLoop\Factory;
 use React\Stream\ReadableResourceStream;
-use React\Promise\Stream;
+use React\Stream\ReadableStreamInterface;
 
 require __DIR__ . '/../vendor/autoload.php';
 
 $in = isset($argv[1]) ? $argv[1] : (__DIR__ . '/../tests/fixtures/alice-bob.tar');
 echo 'Reading file "' . $in . '" (pass as argument to example)' . PHP_EOL;
 
-// using the default loop does *not* work for file I/O
-//$loop = Factory::create();
-$loop = new StreamSelectLoop();
-
+$loop = Factory::create();
 $stream = new ReadableResourceStream(fopen($in, 'r'), $loop);
 
 $decoder = new Decoder();
-$decoder->on('entry', function ($header, $file) {
+$decoder->on('entry', function (array $header, ReadableStreamInterface $file) {
     static $i = 0;
     echo 'FILE #' . ++$i . PHP_EOL;
 
-
     echo 'Received entry headers:' . PHP_EOL;
     var_dump($header);
 
-    Stream\buffer($file)->then(function ($contents) {
+    $contents = '';
+    $file->on('data', function ($chunk) use (&$contents) {
+        $contents .= $chunk;
+    });
+    $file->on('close', function () use (&$contents) {
         echo 'Received entry contents (' . strlen($contents) . ' bytes)' . PHP_EOL;
 
         $d = new Hexdump();
         echo $d->dump($contents) . PHP_EOL . PHP_EOL;
-    }, function ($error) {
-        echo 'ERROR: ' . $error . PHP_EOL;
     });
 });
 $decoder->on('error', function ($error) {

src/Decoder.php

@@ -3,7 +3,6 @@
 namespace Clue\React\Tar;
 
 use Evenement\EventEmitter;
-use Exception;
 use React\Stream\ThroughStream;
 use React\Stream\WritableStreamInterface;
 use RuntimeException;
@@ -15,8 +14,8 @@
  * introduced by POSIX IEEE P1003.1. In the future, it should support more of
  * the less common alternative formats.
  *
- * @event entry(array $header, \React\Stream\ReadableStreamInterface $stream, Decoder $thisDecoder)
- * @event error(Exception $e, Decoder $thisDecoder)
+ * @event entry(array $header, \React\Stream\ReadableStreamInterface $stream)
+ * @event error(Exception $e)
  * @event close()
  */
 class Decoder extends EventEmitter implements WritableStreamInterface
@@ -37,14 +36,14 @@ public function __construct()
 
         if (PHP_VERSION < 5.5) {
             // PHP 5.5 replaced 'a' with 'Z' (read X bytes and removing trailing NULL bytes)
-            $this->format = str_replace('Z', 'a', $this->format);
+            $this->format = str_replace('Z', 'a', $this->format); // @codeCoverageIgnore
         }
     }
 
     public function write($data)
     {
         if (!$this->writable) {
-            return;
+            return false;
         }
 
         // incomplete entry => read until end of entry before expecting next header
@@ -53,7 +52,7 @@ public function write($data)
 
             // entry still incomplete => wait for next chunk
             if ($this->streaming !== null) {
-                return;
+                return true;
             }
         }
 
@@ -63,7 +62,7 @@ public function write($data)
 
             // padding still remaining => wait for next chunk
             if ($this->padding !== 0) {
-                return;
+                return true;
             }
         }
 
@@ -80,32 +79,32 @@ public function write($data)
             }
             try {
                 $header = $this->readHeader($header);
-            } catch (Exception $e) {
+            } catch (RuntimeException $e) {
                 // clean up before throwing
                 $this->buffer = '';
                 $this->writable = false;
 
-                $this->emit('error', array($e, $this));
+                $this->emit('error', array($e));
                 $this->close();
-                return;
+                return false;
             }
 
             $this->streaming = new ThroughStream();
             $this->remaining = $header['size'];
             $this->padding   = $header['padding'];
 
-            $this->emit('entry', array($header, $this->streaming, $this));
+            $this->emit('entry', array($header, $this->streaming));
 
             if ($this->remaining === 0) {
-                $this->streaming->close();
+                $this->streaming->end();
                 $this->streaming = null;
             } else {
                 $this->buffer = $this->consumeEntry($this->buffer);
             }
 
             // incomplete entry => do not read next header
             if ($this->streaming !== null) {
-                return;
+                return true;
             }
 
             if ($this->padding !== 0) {
@@ -114,9 +113,11 @@ public function write($data)
 
             // incomplete padding => do not read next header
             if ($this->padding !== 0) {
-                return;
+                return true;
             }
         }
+
+        return true;
     }
 
     public function end($data = null)
@@ -125,6 +126,22 @@ public function end($data = null)
             $this->write($data);
         }
 
+        if ($this->streaming !== null) {
+            // input stream ended but we were still streaming an entry => emit error about incomplete entry
+            $this->streaming->emit('error', array(new \RuntimeException('TAR input stream ended unexpectedly')));
+            $this->streaming->close();
+            $this->streaming = null;
+
+            // add some dummy data to also trigger error on decoder stream
+            $this->buffer = '.';
+        }
+
+        if ($this->buffer !== '') {
+            // incomplete entry in buffer
+            $this->emit('error', array(new \RuntimeException('Stream ended with incomplete entry')));
+            $this->buffer = '';
+        }
+
         $this->writable = false;
         $this->close();
     }
@@ -137,25 +154,18 @@ public function close()
 
         $this->closing = true;
         $this->writable = false;
+        $this->buffer = '';
 
         if ($this->streaming !== null) {
-            // input stream ended but we were still streaming an entry => emit error about incomplete entry
-            $this->streaming->emit('error', array());
+            // input stream ended but we were still streaming an entry => forcefully close without error
             $this->streaming->close();
             $this->streaming = null;
-
-            $this->emit('error', array());
-        }
-
-        if ($this->buffer !== '') {
-            // incomplete entry in buffer
-            $this->emit('error', array());
-            $this->buffer = '';
         }
 
         // ignore whether we're still expecting NUL-padding
 
-        $this->emit('close', array($this));
+        $this->emit('close');
+        $this->removeAllListeners();
     }
 
     public function isWritable()
@@ -174,11 +184,11 @@ private function consumeEntry($buffer)
         $this->remaining -= $len;
 
         // emit chunk of data
-        $this->streaming->emit('data', array($data, $this->streaming));
+        $this->streaming->write($data);
 
         // nothing remaining => entry stream finished
         if ($this->remaining === 0) {
-            $this->streaming->close();
+            $this->streaming->end();
             $this->streaming = null;
         }