as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

改名

Browse Source
This commit is contained in:
Mr·x
2021-04-22 14:49:32 +08:00
parent dfcf1bf411
commit ed927e00e0
1 changed files with 16 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
System/Channel.php
+15 -40
View File
@@ -7,6 +7,7 @@ namespace Snowflake;
use Closure; use Closure;
use Exception; use Exception;
use Snowflake\Abstracts\Component; use Snowflake\Abstracts\Component;
use SplQueue;
use Swoole\Coroutine\Channel as CChannel; use Swoole\Coroutine\Channel as CChannel;
@@ -18,58 +19,36 @@ class Channel extends Component
{ {
private ?CChannel $_channel = null;
private array $_channels = []; private array $_channels = [];
/** /**
* @param mixed $value * @param mixed $value
* @param string $name * @param string $name
* @param int $length
* @return mixed * @return mixed
* @throws Exception * @throws Exception
*/ */
public function push(mixed $value, string $name = '', $length = 999): mixed public function push(mixed $value, string $name = ''): mixed
{ {
return true; $channel = $this->channelInit($name);
$channel = $this->channelInit($length, $name);
if ($channel->isFull()) { if ($channel->isFull()) {
return $this->addError('Channel is full.'); return $this->addError('Channel is full.');
} }
return true;
return $channel->push($value); return $channel->push($value);
} }
/** /**
* @param int $length
* @param string $name * @param string $name
* @return bool|CChannel * @return bool|CChannel
*/ */
private function channelInit(int $length, string $name = ''): bool|CChannel private function channelInit(string $name = ''): bool|SplQueue
{ {
if ($length < 1) { if (!isset($this->_channels[$name]) || !($this->_channels[$name] instanceof SplQueue)) {
return false; $this->_channels[$name] = new SplQueue();
}
if (empty($name)) {
if (!($this->_channel instanceof CChannel)
|| $this->_channel->close()) {
$this->_channel = new CChannel($length);
}
return $this->_channel;
} else {
if (!isset($this->_channels[$name]) || !($this->_channels[$name] instanceof CChannel)) {
$this->_channels[$name] = new CChannel($length);
} else if ($this->_channels[$name]->close()) {
$this->_channels[$name] = new CChannel($length);
} }
return $this->_channels[$name]; return $this->_channels[$name];
} }
}
/** /**
@@ -78,11 +57,11 @@ class Channel extends Component
*/ */
public function cleanAll() public function cleanAll()
{ {
return; /** @var SplQueue $channel */
/** @var CChannel $channel */
foreach ($this->_channels as $channel) { foreach ($this->_channels as $channel) {
$channel->close(); while ($channel->count() > 0) {
$channel->dequeue();
}
} }
$this->_channels = []; $this->_channels = [];
} }
@@ -91,24 +70,20 @@ class Channel extends Component
* @param $timeout * @param $timeout
* @param Closure $closure * @param Closure $closure
* @param string $name * @param string $name
* @param int $length
* @return mixed * @return mixed
* @throws Exception * @throws Exception
*/ */
public function pop(string $name, Closure $closure, int|float $timeout = null, int $length = 999): mixed public function pop(string $name, Closure $closure, int|float $timeout = null): mixed
{ {
return call_user_func($closure); if (($channel = $this->channelInit($name)) == false) {
if (($channel = $this->channelInit($length, $name)) == false) {
return $this->addError('Channel is full.'); return $this->addError('Channel is full.');
} }
if (!$channel->isEmpty()) { if (!$channel->isEmpty()) {
return $channel->pop(); return $channel->pop();
} }
$data = null; if ($timeout !== null) {
// if ($timeout !== null) { $data = $channel->pop($timeout);
// $data = $channel->pop($timeout); }
// }
if (empty($data)) { if (empty($data)) {
$data = call_user_func($closure); $data = call_user_func($closure);
} }