Adding EVENT_* constants to reader

Implementing separate parsing of header-row
Implementing row-counter

Author: Rakdar

examples/country-codes-reader.php

@@ -12,8 +12,13 @@
 include '../vendor/autoload.php';
 
 $loop = Factory::create();
-$input = new Reader(new ReadableResourceStream(fopen('country-codes.csv', 'r'), $loop));
+
+$start = microtime();
+$inputFd = fopen('country-codes.csv', 'rn');
+$end = microtime();
+$input = new Reader(new ReadableResourceStream($inputFd, $loop));
 $input->setDelimiter(",");
+$input->setParseHeader(false);
 
 $input->on('data', function ($field) {
     echo $field[10] . PHP_EOL;

examples/writer.php

@@ -5,17 +5,26 @@
 use React\EventLoop\Timer\Timer;
 use React\Stream\WritableResourceStream;
 
+chdir(__DIR__);
 require '../vendor/autoload.php';
 
+// Preperation
 $loop = Factory::create();
-
-$writer = new Writer(new WritableResourceStream(fopen('php://stdout', 'w'), $loop));
+$writer = new Writer(
+    new WritableResourceStream(
+        fopen('php://stdout', 'wn'),
+        $loop
+    )
+);
 $int = (object)0;
 
+// Adding a timer to periodically print out some data
 $loop->addPeriodicTimer(1, function ($timer) use ($writer, $int) {
     /** @var Timer $timer */
     $writer->write([2 * $int->scalar, "fds " . 3 * $int->scalar . " a", 4 * $int->scalar]);
     $int->scalar++;
+
+    // Clean Stopping after printing 10 times
     if ($int->scalar > 10) {
         $writer->close();
         $timer->cancel();

src/Reader.php

@@ -6,25 +6,35 @@
 use Evenement\EventEmitterTrait;
 use React\Stream\ReadableStreamInterface;
 use React\Stream\Util;
-use React\Stream\WritableStreamInterface;
 
 class Reader implements EventEmitterInterface
 {
     use EventEmitterTrait;
 
+    const EVENT_HEADER = 'header';
+    const EVENT_DATA = 'data';
+
     /** @var ReadableStreamInterface $stream */
     protected $stream;
 
     /** @var resource $buffer */
     protected $buffer;
     protected $paused = false;
+
+    /** @var null|array $header */
+    protected $header;
+    protected $parseHeader = true;
+    protected $headerParsed = false;
+
     protected $delimiter = ",";
     protected $enclosure = "\"";
     protected $escape = "\\";
 
+    protected $rowsParsed = 0;
+
     /**
      * Reader constructor.
-     * @param WritableStreamInterface $stream
+     * @param ReadableStreamInterface $stream
      */
     public function __construct(ReadableStreamInterface $stream)
     {
@@ -56,7 +66,17 @@ public function parseBuffer()
                 $this->stream->isReadable() === false
             ) {
                 $start = ftell($this->buffer);
-                $this->emit("data", [$field]);
+                ++$this->rowsParsed;
+                if (
+                    $this->headerParsed === false &&
+                    $this->parseHeader === true
+                ) {
+                    $this->header = $field;
+                    $this->headerParsed = true;
+                    $this->emit("header", [$field]);
+                } else {
+                    $this->emit("data", [$field]);
+                }
             }
         }
 
@@ -66,6 +86,11 @@ public function parseBuffer()
         fputs($this->buffer, $dataRemainig);
     }
 
+    public function setParseHeader($parseHeader)
+    {
+        $this->parseHeader = (bool)$parseHeader;
+    }
+
     public function isPaused()
     {
         return $this->paused;
@@ -96,6 +121,16 @@ public function close()
         $this->stream->close();
     }
 
+    public function getHeader()
+    {
+        return $this->header;
+    }
+
+    public function getRowsParsed()
+    {
+        return $this->rowsParsed;
+    }
+
     public function setDelimiter($delimiter)
     {
         $this->delimiter = mb_substr($delimiter, 0, 1);