Compare commits

...

34 Commits

Author SHA1 Message Date
as2252258 122f37d6ce Revert "改名"
This reverts commit fdf58326
2022-01-14 15:23:29 +08:00
as2252258 9fe40ff78a Revert "改名"
This reverts commit fdf58326
2022-01-14 15:20:57 +08:00
as2252258 ae4d75f310 Revert "改名"
This reverts commit fdf58326
2022-01-14 14:53:38 +08:00
as2252258 707379c0fd Revert "改名"
This reverts commit fdf58326
2022-01-14 14:52:56 +08:00
as2252258 b861e0b17b Revert "改名"
This reverts commit fdf58326
2022-01-14 14:51:35 +08:00
as2252258 7b576476a2 Revert "改名"
This reverts commit fdf58326
2022-01-14 14:41:57 +08:00
as2252258 711a819ebb Revert "改名"
This reverts commit fdf58326
2022-01-14 14:40:54 +08:00
as2252258 971b1f1fb0 Revert "改名"
This reverts commit fdf58326
2022-01-14 14:18:28 +08:00
as2252258 4831bc67f5 Revert "改名"
This reverts commit fdf58326
2022-01-14 11:29:15 +08:00
as2252258 1b8c6ecde0 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:55:02 +08:00
as2252258 a4c78874e4 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:41:15 +08:00
as2252258 b6c4693ef6 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:36:25 +08:00
as2252258 adfa1cf3f2 Revert "改名"
This reverts commit fdf58326
2022-01-12 16:00:11 +08:00
as2252258 6362a5ce91 Revert "改名"
This reverts commit fdf58326
2022-01-12 14:54:51 +08:00
as2252258 edeedd2258 Revert "改名"
This reverts commit fdf58326
2022-01-12 14:46:10 +08:00
as2252258 02d0c9a8fd Revert "改名"
This reverts commit fdf58326
2022-01-12 14:43:28 +08:00
as2252258 d707b7f384 Revert "改名"
This reverts commit fdf58326
2022-01-12 14:10:32 +08:00
as2252258 b6f1d8eaf7 Revert "改名"
This reverts commit fdf58326
2022-01-12 11:52:58 +08:00
as2252258 c67fc70550 Revert "改名"
This reverts commit fdf58326
2022-01-12 11:42:30 +08:00
as2252258 e55be953bf Revert "改名"
This reverts commit fdf58326
2022-01-12 11:41:26 +08:00
as2252258 aca8cac1a4 Revert "改名"
This reverts commit fdf58326
2022-01-12 11:39:16 +08:00
as2252258 e699907970 Revert "改名"
This reverts commit fdf58326
2022-01-12 11:32:36 +08:00
as2252258 97dc2f559b Revert "改名"
This reverts commit fdf58326
2022-01-12 11:29:50 +08:00
as2252258 33a01eda28 Revert "改名"
This reverts commit fdf58326
2022-01-12 11:20:33 +08:00
as2252258 b38704374a Revert "改名"
This reverts commit fdf58326
2022-01-11 17:58:04 +08:00
as2252258 318377212c Revert "改名"
This reverts commit fdf58326
2022-01-11 16:16:04 +08:00
as2252258 ddbdae5013 Revert "改名"
This reverts commit fdf58326
2022-01-11 10:26:34 +08:00
as2252258 6eb7aa1fb0 Revert "改名"
This reverts commit fdf58326
2022-01-10 19:03:55 +08:00
as2252258 078e8338fe Revert "改名"
This reverts commit fdf58326
2022-01-10 18:57:54 +08:00
as2252258 6bcd76333a Revert "改名"
This reverts commit fdf58326
2022-01-10 13:43:01 +08:00
as2252258 b2728947a3 Revert "改名"
This reverts commit fdf58326
2022-01-10 11:39:55 +08:00
as2252258 8ec5b0e8d8 1 2022-01-09 17:56:47 +08:00
as2252258 2939d58585 1 2022-01-09 15:44:54 +08:00
as2252258 146da1d00c 1 2022-01-09 15:41:37 +08:00
49 changed files with 389 additions and 231 deletions
+9 -4
View File
@@ -1,10 +1,11 @@
<?php <?php
namespace Server\Abstracts; namespace Kiri\Server\Abstracts;
use Kiri\Context; use Kiri\Context;
use Server\Contract\OnProcessInterface; use Kiri\Server\Broadcast\OnBroadcastInterface;
use Kiri\Server\Contract\OnProcessInterface;
use Swoole\Coroutine; use Swoole\Coroutine;
use Swoole\Process; use Swoole\Process;
@@ -48,9 +49,9 @@ abstract class BaseProcess implements OnProcessInterface
/** /**
* @return bool * @return bool
*/ */
public function getRedirectStdinAndStdout(): bool public function getRedirectStdinAndStdout(): bool
{ {
return $this->redirect_stdin_and_stdout; return $this->redirect_stdin_and_stdout;
} }
@@ -105,6 +106,10 @@ abstract class BaseProcess implements OnProcessInterface
protected function onShutdown($data): void protected function onShutdown($data): void
{ {
$this->isStop = true; $this->isStop = true;
$value = Context::getContext('waite:process:message');
if (Coroutine::exists($value)) {
Coroutine::cancel($value);
}
} }
+2 -2
View File
@@ -1,14 +1,14 @@
<?php <?php
namespace Server\Abstracts; namespace Kiri\Server\Abstracts;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Exception; use Exception;
use Kiri\Abstracts\Config; use Kiri\Abstracts\Config;
use Kiri\Exception\ConfigException; use Kiri\Exception\ConfigException;
use Kiri\Kiri; use Kiri;
use Psr\Log\LoggerInterface; use Psr\Log\LoggerInterface;
+34
View File
@@ -0,0 +1,34 @@
<?php
namespace Kiri\Server\Broadcast;
use Kiri;
use Kiri\Server\ProcessManager;
use Kiri\Server\SwooleServerInterface;
class Broadcast
{
/**
* @param $message
* @return void
*/
public function broadcast($message)
{
$di = Kiri::getDi();
$di->get(ProcessManager::class)->push($message);
$server = $di->get(SwooleServerInterface::class);
$total = $server->setting['worker_num'] + $server->setting['task_worker_num'];
for ($i = 0; $i < $total; $i++) {
if ($i == env('environmental_workerId')) {
continue;
}
$server->sendMessage(serialize(new Message($message)), $i);
}
}
}
+34
View File
@@ -0,0 +1,34 @@
<?php
namespace Kiri\Server\Broadcast;
use Kiri;
use Kiri\Server\Contract\OnPipeMessageInterface;
use Psr\Log\LoggerInterface;
/**
*
*/
class Message implements OnPipeMessageInterface, OnBroadcastInterface
{
/**
* @param mixed $data
*/
public function __construct(public mixed $data)
{
}
/**
* @return void
*/
public function process(): void
{
$logger = Kiri::getDi()->get(LoggerInterface::class);
$logger->debug(env('environmental') . '::' . env('environmental_workerId', 0) . '::' . $this->data);
}
}
+14
View File
@@ -0,0 +1,14 @@
<?php
namespace Kiri\Server\Broadcast;
interface OnBroadcastInterface
{
/**
* @return void
*/
public function process(): void;
}
+1 -1
View File
@@ -1,7 +1,7 @@
<?php <?php
namespace Server; namespace Kiri\Server;
/** /**
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
interface OnBeforeShutdown interface OnBeforeShutdown
{ {
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Http\Request; use Swoole\Http\Request;
use Swoole\Http\Response; use Swoole\Http\Response;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Kiri\Websocket\WebSocketInterface; use Kiri\Websocket\WebSocketInterface;
use Swoole\WebSocket\Frame; use Swoole\WebSocket\Frame;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Http\Request; use Swoole\Http\Request;
+2 -2
View File
@@ -1,8 +1,8 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Server\Abstracts\Server; use Kiri\Server\Abstracts\Server;
interface OnPacketInterface interface OnPacketInterface
{ {
+1 -1
View File
@@ -1,7 +1,7 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
/** /**
+1 -1
View File
@@ -1,7 +1,7 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Process; use Swoole\Process;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,7 +1,7 @@
<?php <?php
namespace Server\Contract; namespace Kiri\Server\Contract;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
class OnAfterWorkerStart class OnAfterWorkerStart
{ {
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
class OnBeforeWorkerStart class OnBeforeWorkerStart
{ {
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
class OnServerBeforeStart class OnServerBeforeStart
{ {
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server\Events; namespace Kiri\Server\Events;
use Swoole\Server; use Swoole\Server;
+8 -5
View File
@@ -1,12 +1,12 @@
<?php <?php
namespace Server\Handler; namespace Kiri\Server\Handler;
use Kiri\Annotation\Inject;
use Server\Abstracts\Server;
use Exception; use Exception;
use Server\Contract\OnPipeMessageInterface; use Kiri\Annotation\Inject;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use Kiri\Server\Abstracts\Server;
use Kiri\Server\Contract\OnPipeMessageInterface;
/** /**
* *
@@ -15,7 +15,7 @@ class OnPipeMessage extends Server
{ {
/** @var EventDispatch */ /** @var EventDispatch */
#[Inject(EventDispatch::class)] #[Inject(EventDispatch::class)]
public EventDispatch $eventDispatch; public EventDispatch $eventDispatch;
@@ -28,6 +28,9 @@ class OnPipeMessage extends Server
*/ */
public function onPipeMessage(\Swoole\Server $server, int $src_worker_id, mixed $message) public function onPipeMessage(\Swoole\Server $server, int $src_worker_id, mixed $message)
{ {
if (is_string($message)) {
$message = unserialize($message);
}
if (!is_object($message) || !($message instanceof OnPipeMessageInterface)) { if (!is_object($message) || !($message instanceof OnPipeMessageInterface)) {
return; return;
} }
+5 -5
View File
@@ -1,15 +1,15 @@
<?php <?php
namespace Server\Handler; namespace Kiri\Server\Handler;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use Kiri\Exception\ConfigException; use Kiri\Exception\ConfigException;
use ReflectionException; use ReflectionException;
use Server\Abstracts\Server; use Kiri\Server\Abstracts\Server;
use Server\Events\OnBeforeShutdown; use Kiri\Server\Events\OnBeforeShutdown;
use Server\Events\OnShutdown; use Kiri\Server\Events\OnShutdown;
use Server\Events\OnStart; use Kiri\Server\Events\OnStart;
/** /**
+4 -4
View File
@@ -1,14 +1,14 @@
<?php <?php
namespace Server\Handler; namespace Kiri\Server\Handler;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use ReflectionException; use ReflectionException;
use Server\Abstracts\Server; use Kiri\Server\Abstracts\Server;
use Kiri\Exception\ConfigException; use Kiri\Exception\ConfigException;
use Server\Events\OnManagerStart; use Kiri\Server\Events\OnManagerStart;
use Server\Events\OnManagerStop; use Kiri\Server\Events\OnManagerStop;
/** /**
+3 -3
View File
@@ -1,11 +1,11 @@
<?php <?php
namespace Server\Handler; namespace Kiri\Server\Handler;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use Server\Events\OnAfterReload; use Kiri\Server\Events\OnAfterReload;
use Server\Events\OnBeforeReload; use Kiri\Server\Events\OnBeforeReload;
use Swoole\Server; use Swoole\Server;
+27 -14
View File
@@ -1,20 +1,21 @@
<?php <?php
namespace Server\Handler; namespace Kiri\Server\Handler;
use Exception; use Exception;
use Kiri;
use Kiri\Abstracts\Config; use Kiri\Abstracts\Config;
use Kiri\Annotation\Inject;
use Kiri\Core\Help; use Kiri\Core\Help;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use Kiri\Kiri; use Kiri\Message\Handler\Router;
use Kiri\Annotation\Inject; use Kiri\Server\Events\OnAfterWorkerStart;
use Server\Events\OnAfterWorkerStart; use Kiri\Server\Events\OnBeforeWorkerStart;
use Server\Events\OnBeforeWorkerStart; use Kiri\Server\Events\OnTaskerStart as OnTaskStart;
use Server\Events\OnTaskerStart as OnTaskStart; use Kiri\Server\Events\OnWorkerError;
use Server\Events\OnWorkerError; use Kiri\Server\Events\OnWorkerExit;
use Server\Events\OnWorkerExit; use Kiri\Server\Events\OnWorkerStart;
use Server\Events\OnWorkerStart; use Kiri\Server\Events\OnWorkerStop;
use Server\Events\OnWorkerStop;
use Swoole\Server; use Swoole\Server;
use Swoole\Timer; use Swoole\Timer;
@@ -23,7 +24,7 @@ use Swoole\Timer;
* Class OnServerWorker * Class OnServerWorker
* @package Server\Worker * @package Server\Worker
*/ */
class OnServerWorker extends \Server\Abstracts\Server class OnServerWorker extends \Kiri\Server\Abstracts\Server
{ {
@@ -34,6 +35,18 @@ class OnServerWorker extends \Server\Abstracts\Server
public EventDispatch $eventDispatch; public EventDispatch $eventDispatch;
public Router $collector;
/**
* @return void
*/
public function init()
{
$this->collector = Kiri::getDi()->get(Router::class);
}
/** /**
* @param Server $server * @param Server $server
* @param int $workerId * @param int $workerId
@@ -42,9 +55,11 @@ class OnServerWorker extends \Server\Abstracts\Server
public function onWorkerStart(Server $server, int $workerId) public function onWorkerStart(Server $server, int $workerId)
{ {
$this->eventDispatch->dispatch(new OnBeforeWorkerStart($workerId)); $this->eventDispatch->dispatch(new OnBeforeWorkerStart($workerId));
set_env('environmental_workerId', $workerId);
if ($workerId < $server->setting['worker_num']) { if ($workerId < $server->setting['worker_num']) {
$this->eventDispatch->dispatch(new OnWorkerStart($server, $workerId));
$this->setProcessName(sprintf('Worker[%d].%d', $server->worker_pid, $workerId)); $this->setProcessName(sprintf('Worker[%d].%d', $server->worker_pid, $workerId));
$this->collector->scan_build_route();
$this->eventDispatch->dispatch(new OnWorkerStart($server, $workerId));
set_env('environmental', Kiri::WORKER); set_env('environmental', Kiri::WORKER);
} else { } else {
$this->eventDispatch->dispatch(new OnTaskStart($server, $workerId)); $this->eventDispatch->dispatch(new OnTaskStart($server, $workerId));
@@ -74,8 +89,6 @@ class OnServerWorker extends \Server\Abstracts\Server
*/ */
public function onWorkerExit(Server $server, int $workerId) public function onWorkerExit(Server $server, int $workerId)
{ {
set_env('state', 'exit');
$this->eventDispatch->dispatch(new OnWorkerExit($server, $workerId)); $this->eventDispatch->dispatch(new OnWorkerExit($server, $workerId));
} }
+114
View File
@@ -0,0 +1,114 @@
<?php
namespace Kiri\Server;
use Kiri\Abstracts\Config;
use Kiri\Annotation\Inject;
use Kiri\Context;
use Kiri\Exception\ConfigException;
use Kiri;
use Kiri\Server\Abstracts\BaseProcess;
use Kiri\Server\Broadcast\Message;
use Kiri\Server\Contract\OnProcessInterface;
use Psr\Log\LoggerInterface;
use Swoole\Coroutine;
use Swoole\Process;
class ProcessManager
{
/** @var array<string, Process> */
private array $_process = [];
#[Inject(LoggerInterface::class)]
public LoggerInterface $logger;
/**
* @param string|OnProcessInterface|BaseProcess $customProcess
* @return void
* @throws ConfigException
*/
public function add(string|OnProcessInterface|BaseProcess $customProcess)
{
$server = Kiri::getDi()->get(SwooleServerInterface::class);
if (is_string($customProcess)) {
$customProcess = Kiri::getDi()->get($customProcess);
}
$system = sprintf('[%s].process', Config::get('id', 'system-service'));
$this->logger->debug($system . ' ' . $customProcess->getName() . ' start.');
$server->addProcess($process = $this->parse($customProcess, $system));
$this->_process[$customProcess->getName()] = $process;
}
/**
* @param $customProcess
* @param $system
* @return Process
*/
private function parse($customProcess, $system): Process
{
return new Process(function (Process $process) use ($customProcess, $system) {
if (Kiri::getPlatform()->isLinux()) {
$process->name($system . '(' . $customProcess->getName() . ')');
}
set_env('environmental', Kiri::PROCESS);
$channel = Coroutine::create(function () use ($process, $customProcess) {
while (!$customProcess->isStop()) {
$message = $process->read();
if (!empty($message)) {
$message = unserialize($message);
}
if (is_null($message)) {
continue;
}
$customProcess->onBroadcast($message);
}
});
Context::setContext('waite:process:message', $channel);
$customProcess->onSigterm()->process($process);
},
$customProcess->getRedirectStdinAndStdout(),
$customProcess->getPipeType(),
$customProcess->isEnableCoroutine()
);
}
/**
* @param array $processes
* @return void
* @throws ConfigException
*/
public function batch(array $processes)
{
foreach ($processes as $process) {
$this->add($process);
}
}
/**
* @param string $message
* @param string $name
* @return void
*/
public function push(string $message, string $name = '')
{
$processes = $this->_process;
if (!empty($this->_process[$name])) {
$processes = [$this->_process[$name]];
}
foreach ($processes as $process) {
$process->write(serialize(new Message($message)));
}
}
}
+36 -25
View File
@@ -1,18 +1,22 @@
<?php <?php
namespace Server; namespace Kiri\Server;
use Exception; use Exception;
use Http\Handler\Abstracts\HttpService; use JetBrains\PhpStorm\Pure;
use Kiri;
use Kiri\Abstracts\Config; use Kiri\Abstracts\Config;
use Kiri\Annotation\Inject;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use Kiri\Exception\ConfigException; use Kiri\Exception\ConfigException;
use Kiri\Kiri; use Kiri\Message\Handler\Abstracts\HttpService;
use Kiri\Annotation\Inject; use Kiri\Message\Handler\Router;
use Kiri\Server\Events\OnShutdown;
use Psr\Container\ContainerExceptionInterface; use Psr\Container\ContainerExceptionInterface;
use Kiri\Server\Events\OnServerBeforeStart;
use Psr\Container\NotFoundExceptionInterface; use Psr\Container\NotFoundExceptionInterface;
use Server\Events\OnShutdown; use Swoole\Coroutine;
defined('PID_PATH') or define('PID_PATH', APP_PATH . 'storage/server.pid'); defined('PID_PATH') or define('PID_PATH', APP_PATH . 'storage/server.pid');
@@ -38,11 +42,27 @@ class Server extends HttpService
public State $state; public State $state;
public ServerManager $manager;
/** /**
* *
*/ */
public function init() public function init()
{ {
$this->manager = Kiri::getContainer()->get(ServerManager::class);
$enable_coroutine = Config::get('servers.settings.enable_coroutine', false);
Config::set('servers.settings.enable_coroutine', true);
if ($enable_coroutine != true) {
return;
}
Coroutine::set([
'hook_flags' => SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_BLOCKING_FUNCTION,
'enable_deadlock_check' => FALSE,
'exit_condition' => function () {
return Coroutine::stats()['coroutine_num'] === 0;
}
]);
} }
@@ -63,22 +83,23 @@ class Server extends HttpService
* @throws \ReflectionException * @throws \ReflectionException
* @throws Exception * @throws Exception
*/ */
public function start(): string public function start(): mixed
{ {
$this->manager()->initBaseServer(Config::get('server', [], true), $this->daemon); $this->manager->initBaseServer(Config::get('server', [], true), $this->daemon);
$rpcService = Config::get('rpc', []); $rpcService = Config::get('rpc', []);
if (!empty($rpcService)) { if (!empty($rpcService)) {
$this->manager()->addListener($rpcService['type'], $rpcService['host'], $rpcService['port'], $this->manager->addListener($rpcService['type'], $rpcService['host'], $rpcService['port'],
$rpcService['mode'], $rpcService); $rpcService['mode'], $rpcService);
} }
$processes = array_merge($this->process, Config::get('processes', [])); $processes = array_merge($this->process, Config::get('processes', []));
foreach ($processes as $process) {
$this->manager()->addProcess($process);
}
return $this->manager()->getServer()->start(); $this->container->get(ProcessManager::class)->batch($processes);
$this->eventDispatch->dispatch(new OnServerBeforeStart());
return $this->manager->getServer()->start();
} }
@@ -93,7 +114,7 @@ class Server extends HttpService
public function shutdown() public function shutdown()
{ {
$configs = Config::get('server', [], true); $configs = Config::get('server', [], true);
foreach ($this->manager()->sortService($configs['ports'] ?? []) as $config) { foreach ($this->manager->sortService($configs['ports'] ?? []) as $config) {
$this->state->exit($config['port']); $this->state->exit($config['port']);
} }
$this->container->get(EventDispatch::class)->dispatch(new OnShutdown()); $this->container->get(EventDispatch::class)->dispatch(new OnShutdown());
@@ -127,20 +148,10 @@ class Server extends HttpService
/** /**
* @return \Swoole\Http\Server|\Swoole\Server|\Swoole\WebSocket\Server|null * @return \Swoole\Http\Server|\Swoole\Server|\Swoole\WebSocket\Server|null
* @throws \ReflectionException
*/ */
public function getServer(): \Swoole\Http\Server|\Swoole\Server|\Swoole\WebSocket\Server|null #[Pure] public function getServer(): \Swoole\Http\Server|\Swoole\Server|\Swoole\WebSocket\Server|null
{ {
return $this->manager()->getServer(); return $this->manager->getServer();
}
/**
* @return ServerManager
*/
private function manager(): ServerManager
{
return Kiri::getDi()->get(ServerManager::class);
} }
} }
+11 -48
View File
@@ -1,14 +1,14 @@
<?php <?php
declare(strict_types=1); declare(strict_types=1);
namespace Server; namespace Kiri\Server;
use Exception; use Exception;
use Kiri\Abstracts\Config; use Kiri\Abstracts\Config;
use Kiri\Events\EventDispatch; use Kiri\Events\EventDispatch;
use Kiri\Exception\ConfigException; use Kiri\Exception\ConfigException;
use Kiri\Kiri; use Kiri;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Psr\Container\ContainerExceptionInterface; use Psr\Container\ContainerExceptionInterface;
use Psr\Container\NotFoundExceptionInterface; use Psr\Container\NotFoundExceptionInterface;
@@ -63,58 +63,21 @@ class ServerCommand extends Command
{ {
$manager = Kiri::app()->getServer(); $manager = Kiri::app()->getServer();
$manager->setDaemon((int)!is_null($input->getOption('daemon'))); $manager->setDaemon((int)!is_null($input->getOption('daemon')));
if (is_null($input->getArgument('action'))) {
$action = $input->getArgument('action');
if (is_null($action)) {
throw new Exception('I don\'t know what I want to do.'); throw new Exception('I don\'t know what I want to do.');
} }
if (!in_array($input->getArgument('action'), self::ACTIONS)) { if (!in_array($action, self::ACTIONS)) {
throw new Exception('I don\'t know what I want to do.'); throw new Exception('I don\'t know what I want to do.');
} }
if ($input->getArgument('action') == 'restart') { if ($action == 'restart' || $action == 'stop') {
$manager->shutdown(); $manager->shutdown();
if ($action == 'stop') {
return 1;
}
} }
if ($input->getArgument('action') != 'stop') { return (int)$manager->start();
return $this->generate_runtime_builder($manager);
}
$manager->shutdown();
return 0;
}
/**
* @throws ConfigException
*/
private function configure_set()
{
$enable_coroutine = Config::get('servers.settings.enable_coroutine', false);
Config::set('servers.settings.enable_coroutine', true);
if ($enable_coroutine != true) {
return;
}
Coroutine::set([
'hook_flags' => SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_BLOCKING_FUNCTION,
'enable_deadlock_check' => FALSE,
'exit_condition' => function () {
return Coroutine::stats()['coroutine_num'] === 0;
}
]);
}
/**
* @param $manager
* @return int
* @throws ConfigException
* @throws Exception
*/
private function generate_runtime_builder($manager): int
{
$this->configure_set();
Kiri::app()->getRouter()->read_files();
$manager->start();
return 1;
} }
} }
+37 -67
View File
@@ -1,43 +1,34 @@
<?php <?php
namespace Server; namespace Kiri\Server;
use Exception; use Exception;
use Kiri\Abstracts\Component; use Kiri\Abstracts\Component;
use Kiri\Abstracts\Config; use Kiri\Abstracts\Config;
use Kiri\Error\Logger;
use Kiri\Events\EventDispatch;
use Kiri\Exception\ConfigException;
use Kiri\Kiri;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Kiri\Error\Logger;
use Kiri\Exception\ConfigException;
use Kiri\Server\Contract\OnCloseInterface;
use Kiri\Server\Contract\OnConnectInterface;
use Kiri\Server\Contract\OnDisconnectInterface;
use Kiri\Server\Contract\OnHandshakeInterface;
use Kiri\Server\Contract\OnMessageInterface;
use Kiri\Server\Contract\OnPacketInterface;
use Kiri\Server\Contract\OnReceiveInterface;
use Kiri\Server\Handler\OnPipeMessage;
use Kiri\Server\Handler\OnServer;
use Kiri\Server\Handler\OnServerManager;
use Kiri\Server\Handler\OnServerReload;
use Kiri\Server\Handler\OnServerWorker;
use Kiri\Server\Tasker\OnServerTask;
use Kiri\Websocket\WebSocketInterface;
use Psr\Container\ContainerExceptionInterface; use Psr\Container\ContainerExceptionInterface;
use Psr\Container\NotFoundExceptionInterface; use Psr\Container\NotFoundExceptionInterface;
use ReflectionException;
use Server\Abstracts\BaseProcess;
use Server\Contract\OnCloseInterface;
use Server\Contract\OnConnectInterface;
use Server\Contract\OnDisconnectInterface;
use Server\Contract\OnHandshakeInterface;
use Server\Contract\OnMessageInterface;
use Server\Contract\OnPacketInterface;
use Server\Contract\OnProcessInterface;
use Server\Contract\OnReceiveInterface;
use Server\Events\OnServerBeforeStart;
use Server\Handler\OnPipeMessage;
use Server\Handler\OnServer;
use Server\Handler\OnServerManager;
use Server\Handler\OnServerReload;
use Server\Handler\OnServerWorker;
use Server\Tasker\OnServerTask;
use Swoole\Http\Server as HServer; use Swoole\Http\Server as HServer;
use Swoole\Process; use Swoole\Process;
use Swoole\Server; use Swoole\Server;
use Swoole\Server\Port; use Swoole\Server\Port;
use Swoole\WebSocket\Server as WServer; use Swoole\WebSocket\Server as WServer;
use Symfony\Component\Console\Helper\Table;
use Symfony\Component\Console\Helper\TableSeparator;
use Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Output\OutputInterface;
/** /**
@@ -89,6 +80,9 @@ class ServerManager extends Component
]; ];
/**
* @var array|string[]
*/
private array $eventInterface = [ private array $eventInterface = [
OnReceiveInterface::class => 'receive', OnReceiveInterface::class => 'receive',
OnPacketInterface::class => 'packet', OnPacketInterface::class => 'packet',
@@ -102,11 +96,9 @@ class ServerManager extends Component
/** /**
* @return Server|WServer|HServer|null * @return Server|WServer|HServer|null
* @throws ReflectionException
*/ */
public function getServer(): Server|WServer|HServer|null public function getServer(): Server|WServer|HServer|null
{ {
di(EventDispatch::class)->dispatch(new OnServerBeforeStart());
return $this->server; return $this->server;
} }
@@ -151,30 +143,6 @@ class ServerManager extends Component
} }
/**
* @param string|OnProcessInterface|BaseProcess $customProcess
* @throws Exception
*/
public function addProcess(string|OnProcessInterface|BaseProcess $customProcess)
{
if (is_string($customProcess)) {
$customProcess = Kiri::getDi()->get($customProcess);
}
$system = sprintf('[%s].process', Config::get('id', 'system-service'));
$this->logger->debug($system . ' ' . $customProcess->getName() . ' start.');
$this->server->addProcess(new Process(function (Process $process) use ($customProcess, $system) {
if (Kiri::getPlatform()->isLinux()) {
$process->name($system . '(' . $customProcess->getName() . ')');
}
$customProcess->onSigterm()->process($process);
},
$customProcess->getRedirectStdinAndStdout(),
$customProcess->getPipeType(),
$customProcess->isEnableCoroutine()
));
}
/** /**
* @return array<string,Process> * @return array<string,Process>
*/ */
@@ -279,6 +247,22 @@ class ServerManager extends Component
* @throws ContainerExceptionInterface * @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface * @throws NotFoundExceptionInterface
* @throws Exception * @throws Exception
*
*
*
* $data = new Table($this->container->get(OutputInterface::class));
* $data->setHeaders(['key', 'value']);
*
* $array = [];
* foreach ($this->server->setting as $key => $value) {
* $array[] = [$key, $value];
* $array[] = new TableSeparator();
* }
*
* array_pop($array);
*
* $data->setStyle('box-double');
* $data->setRows($array);
*/ */
private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = []) private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = [])
{ {
@@ -291,20 +275,7 @@ class ServerManager extends Component
$this->server = new $match($host, $port, SWOOLE_PROCESS, $mode); $this->server = new $match($host, $port, SWOOLE_PROCESS, $mode);
$this->server->set(array_merge(Config::get('server.settings', []), $settings['settings'])); $this->server->set(array_merge(Config::get('server.settings', []), $settings['settings']));
$data = new Table($this->container->get(OutputInterface::class)); $this->container->setBindings(SwooleServerInterface::class, $this->server);
$data->setHeaders(['key', 'value']);
$array = [];
foreach ($this->server->setting as $key => $value) {
$array[] = [$key, $value];
$array[] = new TableSeparator();
}
array_pop($array);
$data->setStyle('box-double');
$data->setRows($array);
$data->render();
$id = Config::get('id', 'system-service'); $id = Config::get('id', 'system-service');
@@ -324,7 +295,6 @@ class ServerManager extends Component
if (($this->server->setting['task_worker_num'] ?? 0) > 0) { if (($this->server->setting['task_worker_num'] ?? 0) > 0) {
$this->addTaskListener($settings['events']); $this->addTaskListener($settings['events']);
} }
$this->container->setBindings(SwooleServerInterface::class, $this->server);
$this->addServiceEvents(ServerManager::DEFAULT_EVENT, $this->server); $this->addServiceEvents(ServerManager::DEFAULT_EVENT, $this->server);
if (!empty($settings['events']) && is_array($settings['events'])) { if (!empty($settings['events']) && is_array($settings['events'])) {
$this->addServiceEvents($settings['events'], $this->server); $this->addServiceEvents($settings['events'], $this->server);
+2 -4
View File
@@ -1,13 +1,13 @@
<?php <?php
declare(strict_types=1); declare(strict_types=1);
namespace Server; namespace Kiri\Server;
use Exception; use Exception;
use Kiri\Abstracts\Providers; use Kiri\Abstracts\Providers;
use Kiri\Application; use Kiri\Application;
use Kiri\Kiri; use Kiri;
/** /**
* Class DatabasesProviders * Class DatabasesProviders
@@ -23,8 +23,6 @@ class ServerProviders extends Providers
*/ */
public function onImport(Application $application) public function onImport(Application $application)
{ {
$application->set('server', ['class' => Server::class]);
$container = Kiri::getDi(); $container = Kiri::getDi();
$console = $container->get(\Symfony\Component\Console\Application::class); $console = $container->get(\Symfony\Component\Console\Application::class);
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server; namespace Kiri\Server;
use Exception; use Exception;
use Kiri\Abstracts\Component; use Kiri\Abstracts\Component;
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server; namespace Kiri\Server;
use Swoole\Server; use Swoole\Server;
+11 -12
View File
@@ -1,14 +1,15 @@
<?php <?php
namespace Server\Tasker; namespace Kiri\Server\Tasker;
use Exception; use Exception;
use Kiri\Abstracts\Component; use Kiri\Abstracts\Component;
use Kiri\Core\HashMap; use Kiri\Core\HashMap;
use Kiri\Kiri; use Kiri;
use ReflectionException; use ReflectionException;
use Server\Contract\OnTaskInterface; use Kiri\Server\Contract\OnTaskInterface;
use Server\SwooleServerInterface; use Kiri\Server\SwooleServerInterface;
use Swoole\Server;
/** /**
@@ -19,9 +20,9 @@ class AsyncTaskExecute extends Component
/** /**
* @var SwooleServerInterface|null * @var Server|\Swoole\WebSocket\Server|\Swoole\Http\Server
*/ */
public ?SwooleServerInterface $server = null; public mixed $server = null;
private HashMap $hashMap; private HashMap $hashMap;
@@ -49,18 +50,16 @@ class AsyncTaskExecute extends Component
/** /**
* @param OnTaskInterface|string $handler * @param OnTaskInterface|string $handler
* @param array $params * @param array $params
* @param int|null $workerId * @param int $workerId
* @throws Exception * @throws Exception
*/ */
public function execute(OnTaskInterface|string $handler, array $params = [], int $workerId = null) public function execute(OnTaskInterface|string $handler, array $params = [], int $workerId = -1)
{ {
if (!$this->server) { if (!$this->server) {
$this->server = Kiri::getDi()->get(SwooleServerInterface::class); $this->server = Kiri::getDi()->get(SwooleServerInterface::class);
} }
if ($workerId === null || $workerId <= $this->server->setting['worker_num']) { if ($workerId < 0 || $workerId > $this->server->setting['task_worker_num']) {
$workerNum = $this->server->setting['worker_num']; $workerId = random_int(0, $this->server->setting['task_worker_num'] - 1);
$taskerNum = $workerNum + $this->server->setting['task_worker_num'];
$workerId = random_int($workerNum, $taskerNum - 1);
} }
if (is_string($handler)) { if (is_string($handler)) {
$handler = $this->handle($handler, $params); $handler = $this->handle($handler, $params);
+5 -5
View File
@@ -1,13 +1,13 @@
<?php <?php
namespace Server\Tasker; namespace Kiri\Server\Tasker;
use Kiri\Annotation\Inject; use Kiri\Annotation\Inject;
use Kiri\Abstracts\Logger; use Kiri\Abstracts\Logger;
use Kiri\Exception\ConfigException; use Kiri\Exception\ConfigException;
use Server\Contract\OnTaskInterface; use Kiri\Server\Contract\OnTaskInterface;
use Swoole\Server; use Swoole\Server;
@@ -37,7 +37,7 @@ class OnServerTask
} catch (\Throwable $exception) { } catch (\Throwable $exception) {
$data = jTraceEx($exception); $data = jTraceEx($exception);
$this->logger->error('task', [$data]); $this->logger->error('task', [error_trigger_format($exception)]);
} finally { } finally {
$server->finish($data); $server->finish($data);
} }
@@ -56,9 +56,9 @@ class OnServerTask
} catch (\Throwable $exception) { } catch (\Throwable $exception) {
$data = jTraceEx($exception); $data = jTraceEx($exception);
$this->logger->error('task', [$data]); $this->logger->error('task', [error_trigger_format($exception)]);
} finally { } finally {
$server->finish($data); $task->finish($data);
} }
} }
+1 -1
View File
@@ -1,6 +1,6 @@
<?php <?php
namespace Server; namespace Kiri\Server;
trait TraitServer trait TraitServer
{ {
+1 -1
View File
@@ -19,7 +19,7 @@
}, },
"autoload": { "autoload": {
"psr-4": { "psr-4": {
"Server\\": "./" "Kiri\\Server\\": "./"
} }
} }
} }