onProgress callback

Author: lajosbencz

src/LajosBencz/StreamParseSql/StreamParseSql.php

@@ -12,6 +12,9 @@ class StreamParseSql
 
     const DEFAULT_CHUNK_SIZE = 4096;
 
+    /** @var callable[] */
+    private $_onProgress = [];
+
     /** @var string */
     protected $_filePath = '';
 
@@ -77,15 +80,18 @@ public function parse(): Generator
         $tokenizer = new Tokenizer($this->getDelimiter());
         $tokens = [];
         $filePath = $this->getFilePath();
+        $size = filesize($filePath);
         $fh = fopen($filePath, 'r');
         if (!is_resource($fh)) {
             throw new Exception\FileReadException($filePath);
         }
         if (!flock($fh, LOCK_SH)) {
             throw new Exception\FileLockException($filePath);
         }
+        $pos = 0;
         while (!feof($fh)) {
             $line = fread($fh, $this->_chunkSize);
+            $pos += min(strlen($line), $this->_chunkSize);
             foreach($tokenizer->append($line, feof($fh)) as $token) {
                 switch($token->type) {
                     case 'COMMENT_SINGLE':
@@ -101,11 +107,19 @@ public function parse(): Generator
                         break;
                 }
             }
+            foreach($this->_onProgress as $callback) {
+                $callback($pos, $size);
+            }
         }
         flock($fh, LOCK_UN);
         fclose($fh);
         if(count($tokens) > 0) {
             yield Token::Assemble(...$tokens);
         }
     }
+
+    public function onProgress(callable $callback): void
+    {
+        $this->_onProgress[] = $callback;
+    }
 }