eee

Abstracts/BaseProcess.php

@@ -20,15 +20,30 @@ abstract class BaseProcess implements OnProcessInterface
     private bool $stop = false;
 
 
+    /**
+     * @var bool
+     */
     protected bool $redirect_stdin_and_stdout = FALSE;
 
 
+    /**
+     * @var int
+     */
     protected int $pipe_type = SOCK_DGRAM;
 
 
+    /**
+     * @var bool
+     */
     protected bool $enable_coroutine = false;
 
 
+    /**
+     * @var bool
+     */
+    protected bool $enable_queue = false;
+
+
     /**
      * @var StdoutLogger
      */
@@ -43,9 +58,21 @@ abstract class BaseProcess implements OnProcessInterface
     public \Kiri\Di\Container $container;
 
 
+    /**
+     * @var string
+     */
     public string $name = '';
 
 
+    /**
+     * @return bool
+     */
+    public function isEnableQueue(): bool
+    {
+        return $this->enable_queue;
+    }
+
+
     /**
      * @return string
      */

Abstracts/ProcessManager.php

@@ -1,193 +0,0 @@
-<?php
-
-namespace Kiri\Server\Abstracts;
-
-use Closure;
-use Exception;
-use Kiri;
-use Kiri\Abstracts\Component;
-use Kiri\Di\Inject\Container;
-use Kiri\Error\StdoutLogger;
-use Kiri\Server\Contract\OnProcessInterface;
-use Psr\Log\LoggerInterface;
-use ReflectionException;
-use Swoole\Process;
-use Kiri\Server\ServerInterface;
-use Kiri\Server\Events\OnServerBeforeStart;
-
-class ProcessManager extends Component
-{
-
-
-    /**
-     * @var StdoutLogger
-     */
-    #[Container(LoggerInterface::class)]
-    public StdoutLogger $logger;
-
-
-    /** @var array<string, BaseProcess> */
-    private array $_process = [];
-
-
-    /**
-     * @return void
-     * @throws Exception
-     */
-    public function init(): void
-    {
-        on(OnServerBeforeStart::class, [$this, 'OnServerBeforeStart']);
-    }
-
-
-    /**
-     * @param OnServerBeforeStart $beforeStart
-     * @return void
-     */
-    public function OnServerBeforeStart(OnServerBeforeStart $beforeStart): void
-    {
-        $server = Kiri::getDi()->get(ServerInterface::class);
-        foreach ($this->_process as $custom) {
-            $server->addProcess(new Process(function (Process $process) use ($custom) {
-                $this->extracted($custom, $process);
-            },
-                $custom->getRedirectStdinAndStdout(),
-                $custom->getPipeType(),
-                $custom->isEnableCoroutine()
-            ));
-        }
-    }
-
-
-    /**
-     * @return Process[]
-     */
-    public function getProcesses(): array
-    {
-        return $this->_process;
-    }
-
-
-    /**
-     * @param string|OnProcessInterface|BaseProcess $custom
-     * @throws
-     */
-    public function add(string|OnProcessInterface|BaseProcess $custom): void
-    {
-        if (is_string($custom)) {
-            $custom = Kiri::getDi()->get($custom);
-        }
-
-        if (isset($this->_process[$custom->getName()])) {
-            throw new Exception('Process(' . $custom->getName() . ') is exists.');
-        }
-
-        $this->_process[$custom->getName()] = $custom;
-    }
-
-
-    /**
-     * @return void
-     */
-    public function shutdown(): void
-    {
-        foreach ($this->_process as $process) {
-            Process::kill($process->pid, 0) && Process::kill($process->pid, 15);
-        }
-    }
-
-
-    /**
-     * @param BaseProcess $customProcess
-     * @return Closure
-     */
-    public function resolve(BaseProcess $customProcess): Closure
-    {
-        return static function (Process $process) use ($customProcess) {
-            $this->extracted($customProcess, $process);
-        };
-    }
-
-
-    /**
-     * @param string|null $name
-     * @param string $tag
-     * @return array|Process|null
-     */
-    public function get(?string $name = null, string $tag = 'default'): array|Process|null
-    {
-        $process = $this->_process[$tag] ?? null;
-        if (empty($process)) {
-            return null;
-        }
-        if (!empty($name)) {
-            if (!isset($process[$name])) {
-                return null;
-            }
-            return $process[$name];
-        }
-        return null;
-    }
-
-
-    /**
-     * @return void
-     */
-    public function stop(): void
-    {
-        foreach ($this->_process as $process) {
-            Process::kill($process->pid, 0) && Process::kill($process->pid, 15);
-        }
-    }
-
-
-    /**
-     * @param array|null $processes
-     * @return void
-     * @throws Exception
-     */
-    public function batch(?array $processes): void
-    {
-        if (empty($processes)) {
-            return;
-        }
-        foreach ($processes as $process) {
-            $this->add($process);
-        }
-    }
-
-
-    /**
-     * @param string $message
-     * @param string $name
-     * @return void
-     */
-    public function push(string $name, string $message): void
-    {
-        if (!isset($this->_process[$name])) {
-            return;
-        }
-        $process = $this->_process[$name];
-        $process->write($message);
-    }
-
-    /**
-     * @param mixed $custom
-     * @param Process $process
-     * @return void
-     * @throws Kiri\Exception\ConfigException
-     * @throws ReflectionException
-     */
-    public function extracted(mixed $custom, Process $process): void
-    {
-        set_env('environmental', Kiri::PROCESS);
-        $system = sprintf('[%s].Custom Process', \config('id', 'system-service'));
-        $this->logger->alert($system . ' ' . $custom->getName() . ' start.');
-        if (Kiri::getPlatform()->isLinux()) {
-            $process->name($system . '[' . $process->pid . '].' . $custom->getName());
-        }
-        $custom->onSigterm()->process($process);
-    }
-
-
-}

Abstracts/TraitServer.php

@@ -18,6 +18,9 @@ trait TraitServer
 {
 
 
+    /**
+     * @var array
+     */
     private array $_process = [];
 
 
@@ -45,7 +48,11 @@ trait TraitServer
             if (isset($this->_process[$name->getName()])) {
                 throw new Exception('Process(' . $name->getName() . ') is exists.');
             }
-            $this->_process[$name->getName()] = $this->genProcess($name);
+            $process = $this->genProcess($name);
+            if ($name->isEnableQueue()) {
+                $process->useQueue();
+            }
+            $this->_process[$name->getName()] = $process;
         }
     }
 
@@ -57,7 +64,7 @@ trait TraitServer
     private function genProcess(BaseProcess $name): Process
     {
         return new Process(function (Process $process) use ($name) {
-            $process->name($name->getName());
+            $process->name('[' . \config('id','system-service') . ']' . $name->getName());
             $name->onSigterm()->process($process);
         },
             $name->getRedirectStdinAndStdout(),
@@ -66,6 +73,16 @@ trait TraitServer
     }
 
 
+    /**
+     * @param string $name
+     * @return Process|null
+     */
+    public function getProcess(string $name): ?Process
+    {
+        return $this->_process[$name] ?? null;
+    }
+
+
     /**
      * @return void
      * @throws
@@ -109,14 +126,14 @@ trait TraitServer
      */
     private function onPcntlSignal($signal, $callback): void
     {
-        pcntl_signal($signal, $callback);
+        \pcntl_signal($signal, $callback);
     }
 
 
     /**
      * @return array
      */
-    public function getProcess(): array
+    public function getProcesses(): array
     {
         return $this->_process;
     }
@@ -182,10 +199,10 @@ trait TraitServer
     {
         return match ($type) {
             Constant::SERVER_TYPE_BASE, Constant::SERVER_TYPE_TCP,
-            Constant::SERVER_TYPE_UDP       => Server::class,
-            Constant::SERVER_TYPE_HTTP      => HServer::class,
+            Constant::SERVER_TYPE_UDP => Server::class,
+            Constant::SERVER_TYPE_HTTP => HServer::class,
             Constant::SERVER_TYPE_WEBSOCKET => WServer::class,
-            default                         => null
+            default => null
         };
     }
 

Handler/OnRequest.php

@@ -67,7 +67,7 @@ class OnRequest implements OnRequestInterface
                                 public DataGrip          $dataGrip)
     {
         $this->responseEmitter = $this->response->emmit;
-        $exception             = \config('request.exception');
+        $exception             = \config('exception.http');
         if (!in_array(ExceptionHandlerInterface::class, class_implements($exception))) {
             $exception = ExceptionHandlerDispatcher::class;
         }

HotReload.php

@@ -63,7 +63,7 @@ class HotReload extends BaseProcess
         if (Context::inCoroutine()) {
             Coroutine::create(fn() => $this->onShutdown(Coroutine::waitSignal(SIGTERM | SIGINT)));
         } else {
-            pcntl_signal(SIGTERM, [$this, 'onStop']);
+            \pcntl_signal(SIGTERM, [$this, 'onStop']);
         }
         return $this;
     }

Task/Task.php

@@ -4,7 +4,14 @@ namespace Kiri\Server\Task;
 
 
 use Kiri;
+use Kiri\Router\Base\ExceptionHandlerDispatcher;
+use Kiri\Router\DataGrip;
+use Kiri\Router\Interface\ExceptionHandlerInterface;
 use Kiri\Server\Constant;
+use Psr\Container\ContainerExceptionInterface;
+use Psr\Container\ContainerInterface;
+use Psr\Container\NotFoundExceptionInterface;
+use Psr\Http\Message\ResponseInterface;
 use Swoole\Server;
 
 /**
@@ -13,6 +20,25 @@ use Swoole\Server;
 class Task
 {
 
+
+    public ExceptionHandlerInterface $exception;
+
+
+    /**
+     * @param ContainerInterface $container
+     * @throws ContainerExceptionInterface
+     * @throws NotFoundExceptionInterface
+     */
+    public function __construct(public ContainerInterface $container)
+    {
+        $exception = \config('exception.task');
+        if (!in_array(ExceptionHandlerInterface::class, class_implements($exception))) {
+            $exception = ExceptionHandlerDispatcher::class;
+        }
+        $this->exception = $this->container->get($exception);
+    }
+
+
     /**
      * @param Server $server
      * @return void
@@ -53,12 +79,16 @@ class Task
      */
     public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): mixed
     {
-        $data = json_decode($data, true);
-        if (is_null($data)) {
-            return null;
+        try {
+            $data = json_decode($data, true);
+            if (is_null($data)) {
+                return null;
+            }
+            $data[0] = Kiri::getDi()->get($data[0]);
+            return call_user_func($data, $task_id, $src_worker_id);
+        } catch (\Throwable $throwable) {
+            return $this->exception->emit($throwable, response());
         }
-        $data[0] = Kiri::getDi()->get($data[0]);
-        return call_user_func($data, $task_id, $src_worker_id);
     }