Merge pull request #2 from nutgram/support_new_version

Support new version

composer.json

@@ -3,8 +3,8 @@
   "description": "Async running mode for Nutgram",
   "type": "library",
   "require": {
-    "nutgram/nutgram": ">=3.1.4",
-    "spatie/fork": "^1.1"
+    "nutgram/nutgram": "^4.0",
+    "ext-pcntl": "*"
   },
   "license": "MIT",
   "autoload": {
@@ -19,7 +19,7 @@
     }
   ],
   "require-dev": {
-    "orchestra/testbench": "^7.5"
+    "orchestra/testbench": "^7.0|^8.0"
   },
   "extra": {
     "laravel": {

src/ExtendsNutgramProvider.php

@@ -16,8 +16,8 @@ public function register()
     {
         $this->app->extend(Nutgram::class, function (Nutgram $bot, Application $app) {
             if (!$app->runningUnitTests() && $app->runningInConsole()) {
-                $concurrency = $bot->getConfig()['concurrency'] ?? 2;
-                $bot->setRunningMode(new ParallelPolling($concurrency));
+                $mode = new ParallelPolling($bot->getConfig()?->concurrency ?? 2);
+                $bot->setRunningMode($mode);
             }
 
             return $bot;

src/ParallelPolling.php

@@ -4,34 +4,19 @@
 
 use SergiX44\Nutgram\Nutgram;
 use SergiX44\Nutgram\RunningMode\Polling;
-use Spatie\Async\Pool;
-use Spatie\Fork\Fork;
 use Throwable;
 
 class ParallelPolling extends Polling
 {
-    private Fork $pool;
+    private ProcessManager $manager;
 
     public function __construct(int $concurrency = 2)
     {
-        $this->pool = Fork::new()->concurrent($concurrency);
+        $this->manager = new ProcessManager($concurrency);
     }
 
-    protected function fire(Nutgram $bot, array|null $updates): void
+    protected function fire(Nutgram $bot, array $updates = []): void
     {
-        $tasks = [];
-        foreach ($updates as $update) {
-            $tasks[] = static function () use ($bot, $update) {
-                try {
-                    $bot->processUpdate($update);
-                } catch (Throwable $e) {
-                    echo "$e\n";
-                } finally {
-                    $bot->clearData();
-                }
-            };
-        }
-
-        $this->pool->run(...$tasks);
+        $this->manager->pushUpdates($bot, self::$STDERR, $updates);
     }
 }

src/ProcessManager.php

@@ -0,0 +1,73 @@
+<?php
+
+namespace SergiX44\Async;
+
+use RuntimeException;
+use SergiX44\Nutgram\Nutgram;
+use SergiX44\Nutgram\Telegram\Types\Common\Update;
+use Throwable;
+
+class ProcessManager
+{
+
+    private array $runners = [];
+
+    public function __construct(private int $maxWorkers)
+    {
+    }
+
+    public function pushUpdates(Nutgram $bot, mixed $stderr, array $updates)
+    {
+        $pid = pcntl_fork();
+
+        if ($pid == -1) {
+            throw new RuntimeException('Cannot fork!');
+        } elseif ($pid) {
+            $this->runners[$pid] = $pid;
+
+            // remove the stopped runners
+            foreach ($this->runners as $pid) {
+                if (pcntl_waitpid($pid, $status, WNOHANG | WUNTRACED)) {
+                    if (pcntl_wifexited($status)) {
+                        unset($this->runners[$pid]);
+                    }
+                }
+            }
+        } else {
+            $this->forkRunners($bot, $stderr, $updates);
+        }
+    }
+
+    public function forkRunners(Nutgram $bot, mixed $stderr, array $updates)
+    {
+        $updateWorkers = [];
+        foreach ($updates as $update) {
+            if (count($updateWorkers) >= $this->maxWorkers) {
+                $pid = pcntl_wait($status);
+                unset($updateWorkers[$pid]);
+            }
+
+            $pid = pcntl_fork();
+
+            if ($pid == -1) {
+                throw new RuntimeException('Cannot fork!');
+            } elseif ($pid) {
+                $updateWorkers[$pid] = $pid;
+            } else {
+                try {
+                    $bot->processUpdate($update);
+                } catch (Throwable $e) {
+                    fwrite($stderr, "$e\n");
+                } finally {
+                    exit(0);
+                }
+            }
+        }
+
+        foreach ($updateWorkers as $pid) {
+            pcntl_waitpid($pid, $status);
+        }
+        exit(0);
+    }
+
+}