Common parts for processes mooved out to ProcessTrait

README.md

@@ -43,15 +43,14 @@ use Amp\Http\Server\HttpErrorException;
 use Amp\Http\Server\Request;
 use Amp\Http\Server\RequestHandler\ClosureRequestHandler;
 use Amp\Http\Server\Response;
-use Luzrain\PHPStreamServer\Internal\WorkerProcess;
 use Luzrain\PHPStreamServer\Plugin\HttpServer\HttpServer;
 use Luzrain\PHPStreamServer\Plugin\HttpServer\Listen;
 use Luzrain\PHPStreamServer\Server;
-use Luzrain\PHPStreamServer\WorkerProcessDefinition;
+use Luzrain\PHPStreamServer\WorkerProcess;
 
 $server = new Server();
 
-$server->addWorkersProcess(new WorkerProcessDefinition(
+$server->addWorkersProcess(new WorkerProcess(
     name: 'HTTP Server',
     onStart: function (WorkerProcess $worker) {
         $requestHandler = new ClosureRequestHandler(function (Request $request) : Response {
@@ -62,7 +61,7 @@ $server->addWorkersProcess(new WorkerProcessDefinition(
             };
         });
 
-        $worker->startPlugin(plugin: new HttpServer(
+        $worker->startPlugin(new HttpServer(
             listen: new Listen(listen: '0.0.0.0:8087'),
             requestHandler: $requestHandler,
         ));

src/Internal/MasterProcess.php

@@ -12,10 +12,10 @@
 use Luzrain\PHPStreamServer\Internal\Scheduler\Scheduler;
 use Luzrain\PHPStreamServer\Internal\ServerStatus\ServerStatus;
 use Luzrain\PHPStreamServer\Internal\Supervisor\Supervisor;
-use Luzrain\PHPStreamServer\PeriodicProcessDefinition;
+use Luzrain\PHPStreamServer\PeriodicProcess;
 use Luzrain\PHPStreamServer\Plugin\Module;
 use Luzrain\PHPStreamServer\Server;
-use Luzrain\PHPStreamServer\WorkerProcessDefinition;
+use Luzrain\PHPStreamServer\WorkerProcess;
 use Psr\Log\LoggerInterface;
 use Revolt\EventLoop;
 use Revolt\EventLoop\Driver\StreamSelectDriver;
@@ -78,19 +78,17 @@ public function __construct(
         $this->serverStatus = new ServerStatus();
     }
 
-    public function addWorkerProcess(WorkerProcessDefinition ...$workers): void
+    public function addWorkerProcess(WorkerProcess ...$workers): void
     {
-        foreach ($workers as $workerDefinition) {
-            $worker = WorkerProcess::createFromDefinition($workerDefinition);
+        foreach ($workers as $worker) {
             $this->supervisor->registerWorkerProcess($worker);
             $this->serverStatus->addWorkerProcess($worker);
         }
     }
 
-    public function addPeriodicProcess(PeriodicProcessDefinition ...$workers): void
+    public function addPeriodicProcess(PeriodicProcess...$workers): void
     {
-        foreach ($workers as $workerDefinition) {
-            $worker = PeriodicProcess::createFromDefinition($workerDefinition);
+        foreach ($workers as $worker) {
             $this->scheduler->addWorker($worker);
             $this->serverStatus->addPeriodicProcess($worker);
         }

src/Internal/ProcessTrait.php

@@ -0,0 +1,110 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Luzrain\PHPStreamServer\Internal;
+
+use Amp\Future;
+use Luzrain\PHPStreamServer\Exception\UserChangeException;
+use Luzrain\PHPStreamServer\Internal\MessageBus\Message;
+use Luzrain\PHPStreamServer\ProcessInterface;
+use Psr\Log\LoggerInterface;
+use Revolt\EventLoop;
+
+/**
+ * @psalm-require-implements ProcessInterface
+ * @psalm-require-implements RunnableProcess
+ */
+trait ProcessTrait
+{
+    private readonly string $name;
+    private readonly int $id;
+    private readonly int $pid;
+    private string|null $user = null;
+    private string|null $group = null;
+    private int $exitCode = 0;
+    private LoggerInterface $logger;
+    private readonly string $socketFile;
+
+    /**
+     * @internal
+     */
+    public function run(WorkerContext $workerContext): int
+    {
+        $this->logger = $workerContext->logger;
+        $this->socketFile = $workerContext->socketFile;
+        $this->setUserAndGroup();
+        $this->initWorker();
+        EventLoop::run();
+
+        return $this->exitCode;
+    }
+
+    public function getName(): string
+    {
+        return $this->name;
+    }
+
+    public function getId(): int
+    {
+        return $this->id;
+    }
+
+    public function getPid(): int
+    {
+        return $this->pid;
+    }
+
+    public function getLogger(): LoggerInterface
+    {
+        return $this->logger;
+    }
+
+    private function setUserAndGroup(): void
+    {
+        try {
+            Functions::setUserAndGroup($this->user, $this->group);
+        } catch (UserChangeException $e) {
+            $refl = new \ReflectionClass($this::class);
+            $this->logger->warning($e->getMessage(), [$refl->getShortName() => $this->getName()]);
+            $this->user = Functions::getCurrentUser();
+        }
+    }
+
+    public function detach(): void
+    {
+        $identifiers = EventLoop::getDriver()->getIdentifiers();
+        \array_walk($identifiers, EventLoop::getDriver()->cancel(...));
+        EventLoop::getDriver()->stop();
+        unset($this->logger);
+        unset($this->socketFile);
+    }
+
+    public function exec(string $path, array $args = []): never
+    {
+        $this->detach();
+        $envVars = [...\getenv(), ...$_ENV];
+        \pcntl_exec($path, $args, $envVars);
+        exit(0);
+    }
+
+    public function getUser(): string
+    {
+        return $this->user ?? Functions::getCurrentUser();
+    }
+
+    public function getGroup(): string
+    {
+        return $this->group ?? Functions::getCurrentGroup();
+    }
+
+    /**
+     * @template T
+     * @param Message<T> $message
+     * @return Future<T>
+     */
+    public function dispatch(Message $message): Future
+    {
+        return $this->messageBus->dispatch($message);
+    }
+}