Merge pull request #1 from n-hor/1.0.1

fix common callback behavior

src/Channel.php

@@ -19,8 +19,6 @@ class Channel
 
     public Generator $reader;
 
-    protected string $buffer = '';
-
     public function __construct(
         protected Socket $socket,
         protected int $bufferSize,
@@ -34,7 +32,7 @@ public function __construct(
 
     public function __destruct()
     {
-        socket_close($this->socket);
+        $this->close();
     }
 
     public function read(): Generator
@@ -93,7 +91,7 @@ protected function write(string $data): void
         }
     }
 
-    protected function socketSelect($isRead): bool
+    protected function socketSelect(bool $isRead): bool
     {
         $write = ! $isRead ? [$this->socket] : null;
         $read = $isRead ? [$this->socket] : null;

src/ParallelTasks.php

@@ -93,7 +93,7 @@ protected function getExecutableTask(Closure $task): Closure
     {
         return isset($this->commonBeforeExecutionCallback) ? function () use ($task) {
             ($this->commonBeforeExecutionCallback)();
-            $task();
+            return $task();
         } : $task;
     }
 }

tests/PcntlParallelTest.php

@@ -200,4 +200,56 @@ function () {
     $pool->wait();
     $result = $pool->pullWorkersOutput(); //[6,7,8,9]
     expect(30)->toEqual(array_sum($result));
+    $pool->destroy();
+});
+
+test('common execution callback')->expect(function () {
+
+    $parallelTasks = ParallelTasks::add([
+        function () {
+            global $value;
+            return 1 + $value;
+        },
+        function () {
+            global $value;
+            return 2 + $value;
+        },
+        function () {
+            global $value;
+            return 3 + $value;
+        },
+    ])
+        ->setCommonBeforeExecutionCallback(function () {
+            global $value;
+            $value = 1;
+        })
+        ->run();
+
+    $res = $parallelTasks->waitOutput();
+
+    expect(9)->toEqual(array_sum($res));
+});
+
+test('workers pool common execution callback')->expect(function () {
+    $pool = PersistenceWorkersPool::create(5)
+        ->setCommonBeforeExecutionCallback(function () {
+            global $value;
+            $value = 1;
+        })
+        ->run(function (mixed $job) {
+            global $value;
+            return $job + $value;
+        });
+
+    $tasks = [1,2,3,4,5,6,7,8,9];
+    foreach ($tasks as $task) {
+        $pool->dispatch($task, waitAvailableWorkerTimeout: 2_000_000);
+    }
+
+    $result = $pool->pullWorkersOutput();
+    expect(20)->toEqual(array_sum($result));
+    $pool->wait();
+    $result = $pool->pullWorkersOutput();
+    expect(34)->toEqual(array_sum($result));
+    $pool->destroy();
 });