danog/ipc
1
0
Fork 0
mirror of https://github.com/danog/ipc.git synced 2024-11-26 20:15:05 +01:00
Code Issues Projects Releases Wiki Activity

Make closure a bit more reliable

Browse Source
This commit is contained in:
Daniil Gentili 2020-09-24 21:00:49 +02:00
parent 1a30e5492a
commit 1313eb755a
Signed by: danog
GPG Key ID: 8C1BE3B34B230CA7
2 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
lib/Sync/ChannelledSocket.php
View File

@ -28,6 +28,9 @@ final class ChannelledSocket implements Channel
/** @var ResourceOutputStream */
private $write;
/** @var bool */
private $closed = false;
/** @var int */
private $state = self::ESTABLISHED;
@ -56,7 +59,7 @@ final class ChannelledSocket implements Channel
*/
public function receive(): Promise
{
if (!$this->channel) {
if ($this->closed) {
return new Success();
}
return call(function (): \Generator {
@ -85,24 +88,22 @@ final class ChannelledSocket implements Channel
*/
public function disconnect(): Promise
{
if (!$this->channel) {
if ($this->closed) {
return new Success();
}
$channel = $this->channel;
$this->channel = null;
return call(function () use ($channel): \Generator {
yield $channel->send(new ChannelCloseReq);
$this->closed = true;
return call(function (): \Generator {
yield $this->channel->send(new ChannelCloseReq);
if ($this->reading) {
$this->closePromise = new Deferred;
}
do {
$data = yield ($this->closePromise ? $this->closePromise->promise() : $channel->receive());
$data = yield ($this->closePromise ? $this->closePromise->promise() : $this->channel->receive());
if ($this->closePromise) {
$this->closePromise = null;
}
if ($data instanceof ChannelCloseReq) {
yield $channel->send(new ChannelCloseAck);
yield $this->channel->send(new ChannelCloseAck);
$this->state |= self::GOT_FIN_MASK;
} elseif ($data instanceof ChannelCloseAck) {
$this->state |= self::GOT_ACK_MASK;
@ -119,7 +120,7 @@ final class ChannelledSocket implements Channel
*/
public function send($data): Promise
{
if (!$this->channel) {
if ($this->closed) {
throw new ChannelException('The channel was already closed!');
}
return $this->channel->send($data);

1
lib/functions.php
View File

@ -18,6 +18,7 @@ function listen(string $uri): IpcServer
{
return new IpcServer($uri);
}
/**
* Connect to IPC server.
*