nsq/nsqphp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Receive Reader as dependency in Subscriber instead of extending it

Browse Source
This commit is contained in:
Konstantin Grachev
2021-01-23 01:24:35 +03:00
parent 6509753bd9
commit 7873a9f010
4 changed files with 25 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/Connection.php
View File

@ -122,7 +122,7 @@ abstract class Connection
return $this; return $this;
} }
protected function receive(float $timeout = 0): ?Response public function receive(float $timeout = 0): ?Response
{ {
$socket = $this->socket(); $socket = $this->socket();

5
src/Reader.php
View File

@ -50,4 +50,9 @@ class Reader extends Connection
{ {
$this->send('TOUCH '.$id.PHP_EOL); $this->send('TOUCH '.$id.PHP_EOL);
} }
public function nop(): void
{
$this->send('NOP'.PHP_EOL);
}
} }

21
src/Subscriber.php
View File

@ -9,24 +9,31 @@ use function get_debug_type;
use function microtime; use function microtime;
use function sprintf; use function sprintf;
final class Subscriber extends Reader final class Subscriber
{ {
public const STOP = 0; public const STOP = 0;
public const CHANGE_TIMEOUT = 1; public const CHANGE_TIMEOUT = 1;
private Reader $reader;
public function __construct(Reader $reader)
{
$this->reader = $reader;
}
/** /**
* @psalm-return Generator<int, Envelope|null, int|float|null, void> * @psalm-return Generator<int, Envelope|null, int|float|null, void>
*/ */
public function subscribe(string $topic, string $channel, float $timeout = 0): Generator public function subscribe(string $topic, string $channel, float $timeout = 0): Generator
{ {
$this->sub($topic, $channel); $this->reader->sub($topic, $channel);
while (true) { while (true) {
$this->rdy(1); $this->reader->rdy(1);
$message = $this->consume($timeout); $message = $this->consume($timeout);
$command = yield null === $message ? null : new Envelope($message, $this); $command = yield null === $message ? null : new Envelope($message, $this->reader);
if (self::STOP === $command) { if (self::STOP === $command) {
break; break;
@ -43,20 +50,20 @@ final class Subscriber extends Reader
} }
} }
$this->disconnect(); $this->reader->disconnect();
} }
private function consume(float $timeout): ?Message private function consume(float $timeout): ?Message
{ {
$deadline = microtime(true) + $timeout; $deadline = microtime(true) + $timeout;
$response = $this->receive($timeout); $response = $this->reader->receive($timeout);
if (null === $response) { if (null === $response) {
return null; return null;
} }
if ($response->isHeartBeat()) { if ($response->isHeartBeat()) {
$this->send('NOP'.PHP_EOL); $this->reader->nop();
return $this->consume( return $this->consume(
($currentTime = microtime(true)) > $deadline ? 0 : $deadline - $currentTime ($currentTime = microtime(true)) > $deadline ? 0 : $deadline - $currentTime

8
tests/NsqTest.php
View File

@ -3,6 +3,7 @@
declare(strict_types=1); declare(strict_types=1);
use Nsq\Envelope; use Nsq\Envelope;
use Nsq\Reader;
use Nsq\Subscriber; use Nsq\Subscriber;
use Nsq\Writer; use Nsq\Writer;
use Nsq\Exception; use Nsq\Exception;
@ -15,7 +16,8 @@ final class NsqTest extends TestCase
$writer = new Writer('tcp://localhost:4150'); $writer = new Writer('tcp://localhost:4150');
$writer->pub(__FUNCTION__, __FUNCTION__); $writer->pub(__FUNCTION__, __FUNCTION__);
$subscriber = new Subscriber('tcp://localhost:4150'); $reader = new Reader('tcp://localhost:4150');
$subscriber = new Subscriber($reader);
$generator = $subscriber->subscribe(__FUNCTION__, __FUNCTION__, 1); $generator = $subscriber->subscribe(__FUNCTION__, __FUNCTION__, 1);
/** @var null|Envelope $envelope */ /** @var null|Envelope $envelope */
@ -70,9 +72,9 @@ final class NsqTest extends TestCase
static::assertSame('Deferred message.', $envelope->message->body); static::assertSame('Deferred message.', $envelope->message->body);
$envelope->finish(); $envelope->finish();
static::assertFalse($subscriber->isClosed()); static::assertFalse($reader->isClosed());
$generator->send(Subscriber::STOP); $generator->send(Subscriber::STOP);
static::assertTrue($subscriber->isClosed()); static::assertTrue($reader->isClosed());
} }
/** /**