modify plugin name
This commit is contained in:
+32
-3
@@ -3,12 +3,14 @@
|
|||||||
namespace Kiri\Rpc;
|
namespace Kiri\Rpc;
|
||||||
|
|
||||||
use Exception;
|
use Exception;
|
||||||
use Kiri\Abstracts\Component;
|
|
||||||
use Kiri\Context;
|
|
||||||
use Kiri\Exception\ConfigException;
|
|
||||||
use Kiri;
|
use Kiri;
|
||||||
|
use Kiri\Abstracts\Component;
|
||||||
|
use Kiri\Annotation\Inject;
|
||||||
|
use Kiri\Events\EventProvider;
|
||||||
|
use Kiri\Exception\ConfigException;
|
||||||
use Kiri\Pool\Alias;
|
use Kiri\Pool\Alias;
|
||||||
use Kiri\Pool\Pool;
|
use Kiri\Pool\Pool;
|
||||||
|
use Kiri\Server\Events\OnBeforeShutdown;
|
||||||
use Swoole\Client;
|
use Swoole\Client;
|
||||||
|
|
||||||
|
|
||||||
@@ -32,6 +34,31 @@ class ClientPool extends Component
|
|||||||
public int $waite;
|
public int $waite;
|
||||||
|
|
||||||
|
|
||||||
|
#[Inject(EventProvider::class)]
|
||||||
|
public EventProvider $provider;
|
||||||
|
|
||||||
|
|
||||||
|
private array $names = [];
|
||||||
|
|
||||||
|
|
||||||
|
public function init()
|
||||||
|
{
|
||||||
|
$this->provider->on(OnBeforeShutdown::class, [$this, 'onBeforeShutdown']);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @return void
|
||||||
|
* @throws Exception
|
||||||
|
*/
|
||||||
|
public function onBeforeShutdown()
|
||||||
|
{
|
||||||
|
foreach ($this->names as $name) {
|
||||||
|
$this->getPool()->clean($name);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param $config
|
* @param $config
|
||||||
* @param callable $callback
|
* @param callable $callback
|
||||||
@@ -45,6 +72,8 @@ class ClientPool extends Component
|
|||||||
|
|
||||||
$pool = $config['pool'] ?? ['min' => 1, 'max' => 100];
|
$pool = $config['pool'] ?? ['min' => 1, 'max' => 100];
|
||||||
|
|
||||||
|
$this->names[] = $coroutineName;
|
||||||
|
|
||||||
return $this->getPool()->get($coroutineName, $callback, $pool['min'] ?? 1);
|
return $this->getPool()->get($coroutineName, $callback, $pool['min'] ?? 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ class JsonRpcPoolTransporter implements RpcClientInterface
|
|||||||
|
|
||||||
$client = $this->getClient();
|
$client = $this->getClient();
|
||||||
|
|
||||||
$response = $this->request($client, $content, false);
|
$response = $this->request($client, $content);
|
||||||
|
|
||||||
$this->pool->push($client, $this->config['Address'], $this->config['Port']);
|
$this->pool->push($client, $this->config['Address'], $this->config['Port']);
|
||||||
|
|
||||||
|
|||||||
@@ -28,9 +28,11 @@ class JsonRpcTransporter implements RpcClientInterface
|
|||||||
{
|
{
|
||||||
$content = $request->getBody()->getContents();
|
$content = $request->getBody()->getContents();
|
||||||
|
|
||||||
$response = $this->request($this->newClient(), $content, true);
|
$body = $this->request($this->newClient(), $content);
|
||||||
|
|
||||||
return (new Response())->withBody(new Stream($response));
|
$response = \Kiri::getDi()->get(ResponseInterface::class);
|
||||||
|
|
||||||
|
return $response->withBody(new Stream($body));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -32,17 +32,12 @@ trait TraitTransporter
|
|||||||
/**
|
/**
|
||||||
* @param Client|Coroutine\Client $client
|
* @param Client|Coroutine\Client $client
|
||||||
* @param $content
|
* @param $content
|
||||||
* @param bool $isClose
|
|
||||||
* @return mixed
|
* @return mixed
|
||||||
*/
|
*/
|
||||||
private function request(Client|Coroutine\Client $client, $content, bool $isClose): mixed
|
private function request(Client|Coroutine\Client $client, $content): mixed
|
||||||
{
|
{
|
||||||
$client->send($content);
|
$client->send($content);
|
||||||
$read = $client->recv();
|
return $client->recv();
|
||||||
if ($isClose) {
|
|
||||||
$client->close();
|
|
||||||
}
|
|
||||||
return $read;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user