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

Compare commits

base: nsq:875cb8b542f4cb22c7e72995c2d05ddd4d1827b1
Branches Tags
nsq:master nsq:renovate/configure nsq:tests nsq:amphp nsq:snappy
..
compare: nsq:snappy
Branches Tags
nsq:renovate/configure nsq:master nsq:tests nsq:amphp nsq:snappy

1 Commits
875cb8b542 ... snappy

Author SHA1 Message Date
Konstantin Grachev
87fe939018 Snappy 2021-01-28 01:12:00 +03:00
34 changed files with 511 additions and 1097 deletions
Expand all files Collapse all files

10
.php_cs.dist
View File

@ -8,16 +8,14 @@ return (new PhpCsFixer\Config())
'@PhpCsFixer:risky' => true,
'@PSR12' => true,
'@PSR12:risky' => true,
'blank_line_before_statement' => [
'statements' => ['continue', 'do', 'die', 'exit', 'goto', 'if', 'return', 'switch', 'throw', 'try'],
],
'declare_strict_types' => true,
'global_namespace_import' => ['import_classes' => false, 'import_constants' => false, 'import_functions' => false],
'php_unit_internal_class' => false,
'php_unit_test_case_static_method_calls'=> ['call_type' => 'self'],
'php_unit_test_class_requires_covers' => false,
'phpdoc_to_comment' => false,
'yoda_style' => true,
'php_unit_test_case_static_method_calls'=> ['call_type' => 'self'],
'blank_line_before_statement' => [
'statements' => ['continue', 'do', 'die', 'exit', 'goto', 'if', 'return', 'switch', 'throw', 'try']
],
])
->setFinder(
PhpCsFixer\Finder::create()

26
README.md
View File

@ -34,15 +34,14 @@ Features
- [ ] Discovery
- [ ] Backoff
- [ ] TLS
- [ ] Deflate
- [X] Snappy
- [ ] Snappy
- [X] Sampling
- [X] AUTH
Usage
-----
### Producer
### Publish
```php
use Nsq\Producer;
@ -62,21 +61,17 @@ $producer->mpub('topic', [
$producer->dpub('topic', 'Deferred message', delay: 5000);
```
### Consumer
### Subscription
```php
use Nsq\Consumer;
use Nsq\Protocol\Message;
use Nsq\Message;
use Nsq\Subscriber;
$consumer = new Consumer(
topic: 'topic',
channel: 'channel',
address: 'tcp://nsqd:4150',
);
// Simple blocking loop based on generator
$generator = $consumer->generator();
$consumer = new Consumer('tcp://nsqd:4150');
$subscriber = new Subscriber($consumer);
$generator = $subscriber->subscribe('topic', 'channel');
foreach ($generator as $message) {
if ($message instanceof Message) {
$payload = $message->body;
@ -91,8 +86,11 @@ foreach ($generator as $message) {
// In case of nothing received during timeout generator will return NULL
// Here we can do something between messages, like pcntl_signal_dispatch()
// We can also communicate with generator through send
// for example:
// Gracefully close connection (loop will be ended)
$generator->send(0);
$generator->send(Subscriber::STOP);
}
```

4
docker-compose.yml
View File

@ -3,8 +3,6 @@ version: '3.7'
services:
nsqd:
image: nsqio/nsq:v1.2.0
labels:
ru.grachevko.dhu: 'nsqd'
command: /nsqd
ports:
- 4150:4150
@ -12,8 +10,6 @@ services:
nsqadmin:
image: nsqio/nsq:v1.2.0
labels:
ru.grachevko.dhu: 'nsqadmin'
command: /nsqadmin --nsqd-http-address=nsqd:4151 --http-address=0.0.0.0:4171
ports:
- 4171:4171

68
src/Buffer.php
View File

@ -1,68 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq;
use PHPinnacle\Buffer\ByteBuffer;
final class Buffer
{
private ByteBuffer $buffer;
public function __construct(string $initial = '')
{
$this->buffer = new ByteBuffer($initial);
}
public function append(string $data): self
{
$this->buffer->append($data);
return $this;
}
public function consumeSize(): int
{
/** @see Bytes::BYTES_SIZE */
return $this->buffer->consumeUint32();
}
public function consumeType(): int
{
/** @see Bytes::BYTES_TYPE */
return $this->buffer->consumeUint32();
}
public function consumeTimestamp(): int
{
/** @see Bytes::BYTES_TIMESTAMP */
return $this->buffer->consumeInt64();
}
public function consumeAttempts(): int
{
/** @see Bytes::BYTES_ATTEMPTS */
return $this->buffer->consumeUint16();
}
public function consumeId(): string
{
return $this->buffer->consume(Bytes::BYTES_ID);
}
public function size(): int
{
return $this->buffer->size();
}
public function bytes(): string
{
return $this->buffer->bytes();
}
public function flush(): string
{
return $this->buffer->flush();
}
}

63
src/Config/ClientConfig.php
View File

@ -5,6 +5,9 @@ declare(strict_types=1);
namespace Nsq\Config;
use Composer\InstalledVersions;
use InvalidArgumentException;
use JsonSerializable;
use function gethostname;
/**
* This class is used for configuring the clients for nsq. Immutable properties must be set when creating the object and
@ -13,34 +16,25 @@ use Composer\InstalledVersions;
*
* @psalm-immutable
*/
final class ClientConfig implements \JsonSerializable
final class ClientConfig implements JsonSerializable
{
/**
* @psalm-suppress ImpureFunctionCall
*/
/** @psalm-suppress ImpureFunctionCall */
public function __construct(
/**
/*
* The secret used for authorization, if the server requires it. This value will be ignored if the server
* does not require authorization.
*/
public ?string $authSecret = null,
/**
* The timeout for establishing a connection in seconds.
*/
// The timeout for establishing a connection in seconds.
public int $connectTimeout = 10,
/**
* An identifier used to disambiguate this client (i.e. something specific to the consumer).
*/
// An identifier used to disambiguate this client (i.e. something specific to the consumer)
public string $clientId = '',
/**
* Enable deflate compression for this connection. A client cannot enable both [snappy] and [deflate].
*/
// Enable deflate compression for this connection. A client cannot enable both [snappy] and [deflate].
public bool $deflate = false,
/**
/*
* Configure the deflate compression level for this connection.
*
* Valid range: `1 <= deflate_level <= configured_max`
@ -49,54 +43,42 @@ final class ClientConfig implements \JsonSerializable
*/
public int $deflateLevel = 6,
/**
/*
* Milliseconds between heartbeats.
*
* Valid range: `1000 <= heartbeat_interval <= configured_max` (`-1` disables heartbeats)
*/
public int $heartbeatInterval = 30000,
/**
* The hostname where the client is deployed.
*/
// The hostname where the client is deployed
public string $hostname = '',
/**
* Configure the server-side message timeout in milliseconds for messages delivered to this client.
*/
// Configure the server-side message timeout in milliseconds for messages delivered to this client.
public int $msgTimeout = 60000,
/**
/*
* The sample rate for incoming data to deliver a percentage of all messages received to this connection.
* This only applies to subscribing connections. The valid range is between 0 and 99, where 0 means that all
* data is sent (this is the default). 1 means that 1% of the data is sent.
*/
public int $sampleRate = 0,
/**
/*
* Boolean used to indicate that the client supports feature negotiation. If the server is capable,
* it will send back a JSON payload of supported features and metadata.
*/
public bool $featureNegotiation = true,
/**
* Enable TLS for this connection.
*/
// Enable TLS for this connection
public bool $tls = false,
/**
* Enable snappy compression for this connection. A client cannot enable both [snappy] and [deflate].
*/
// Enable snappy compression for this connection. A client cannot enable both [snappy] and [deflate].
public bool $snappy = false,
/**
* The read timeout for connection sockets and for awaiting responses from nsq.
*/
// The read timeout for connection sockets and for awaiting responses from nsq.
public int $readTimeout = 5,
/**
* A string identifying the agent for this client in the spirit of HTTP.
*/
// A string identifying the agent for this client in the spirit of HTTP.
public string $userAgent = '',
) {
$this->featureNegotiation = true; // Always enabled
@ -110,7 +92,7 @@ final class ClientConfig implements \JsonSerializable
}
if ($this->snappy && $this->deflate) {
throw new \InvalidArgumentException('Client cannot enable both [snappy] and [deflate]');
throw new InvalidArgumentException('Client cannot enable both [snappy] and [deflate]');
}
}
@ -133,9 +115,4 @@ final class ClientConfig implements \JsonSerializable
'user_agent' => $this->userAgent,
];
}
public function toString(): string
{
return json_encode($this, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT);
}
}

51
src/Config/ConnectionConfig.php
View File

@ -12,72 +12,47 @@ namespace Nsq\Config;
final class ConnectionConfig
{
public function __construct(
/**
* Whether or not authorization is required by nsqd.
*/
// Whether or not authorization is required by nsqd.
public bool $authRequired,
/**
* Whether deflate compression is enabled for this connection or not.
*/
// Whether deflate compression is enabled for this connection or not.
public bool $deflate,
/**
* The deflate level. This value can be ignored if [deflate] is `false`.
*/
// The deflate level. This value can be ignored if [deflate] is `false`.
public int $deflateLevel,
/**
* The maximum deflate level supported by the server.
*/
// The maximum deflate level supported by the server.
public int $maxDeflateLevel,
/**
* The maximum value for message timeout.
*/
// The maximum value for message timeout.
public int $maxMsgTimeout,
/**
/*
* Each nsqd is configurable with a max-rdy-count. If the consumer sends a RDY count that is outside
* of the acceptable range its connection will be forcefully closed.
*/
public int $maxRdyCount,
/**
* The effective message timeout.
*/
// The effective message timeout.
public int $msgTimeout,
/**
* The size in bytes of the buffer nsqd will use when writing to this client.
*/
// The size in bytes of the buffer nsqd will use when writing to this client.
public int $outputBufferSize,
/**
* The timeout after which any data that nsqd has buffered will be flushed to this client.
*/
// The timeout after which any data that nsqd has buffered will be flushed to this client.
public int $outputBufferTimeout,
/**
/*
* The sample rate for incoming data to deliver a percentage of all messages received to this connection.
* This only applies to subscribing connections. The valid range is between 0 and 99, where 0 means that all
* data is sent (this is the default). 1 means that 1% of the data is sent.
*/
public int $sampleRate,
/**
* Whether snappy compression is enabled for this connection or not.
*/
// Whether snappy compression is enabled for this connection or not.
public bool $snappy,
/**
* Whether TLS is enabled for this connection or not.
*/
// Whether TLS is enabled for this connection or not.
public bool $tls,
/**
* The nsqd version.
*/
// The nsqd version.
public string $version,
) {
}

365
src/Connection.php
View File

@ -7,114 +7,125 @@ namespace Nsq;
use Nsq\Config\ClientConfig;
use Nsq\Config\ConnectionConfig;
use Nsq\Exception\AuthenticationRequired;
use Nsq\Exception\BadResponse;
use Nsq\Exception\ConnectionFail;
use Nsq\Exception\NotConnected;
use Nsq\Exception\NsqError;
use Nsq\Exception\NsqException;
use Nsq\Protocol\Error;
use Nsq\Protocol\Frame;
use Nsq\Protocol\Message;
use Nsq\Protocol\Response;
use Nsq\Socket\DeflateSocket;
use Nsq\Socket\RawSocket;
use Nsq\Socket\SnappySocket;
use Nsq\Exception\UnexpectedResponse;
use Nsq\Reconnect\ExponentialStrategy;
use Nsq\Reconnect\ReconnectStrategy;
use PHPinnacle\Buffer\ByteBuffer;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Socket\Raw\Exception;
use Socket\Raw\Factory;
use Socket\Raw\Socket;
use function addcslashes;
use function hash;
use function http_build_query;
use function implode;
use function json_encode;
use function pack;
use function snappy_compress;
use function unpack;
use const JSON_FORCE_OBJECT;
use const JSON_THROW_ON_ERROR;
use const PHP_EOL;
/**
* @internal
*
* @property ConnectionConfig $connectionConfig
*/
abstract class Connection
{
use LoggerAwareTrait;
protected ClientConfig $clientConfig;
private string $address;
private NsqSocket $socket;
private ?Socket $socket = null;
private ConnectionConfig $connectionConfig;
private ReconnectStrategy $reconnect;
private bool $closed = false;
private ClientConfig $clientConfig;
private ?ConnectionConfig $connectionConfig = null;
public function __construct(
private string $address,
string $address,
ClientConfig $clientConfig = null,
ReconnectStrategy $reconnectStrategy = null,
LoggerInterface $logger = null,
) {
$this->address = $address;
$this->logger = $logger ?? new NullLogger();
$this->reconnect = $reconnectStrategy ?? new ExponentialStrategy(logger: $this->logger);
$this->clientConfig = $clientConfig ?? new ClientConfig();
$socket = new RawSocket($this->address, $this->logger);
$socket->write(' V2');
$this->socket = new NsqSocket($socket);
$this->connectionConfig = ConnectionConfig::fromArray(
$this
->command('IDENTIFY', data: $this->clientConfig->toString())
->readResponse()
->toArray()
);
if ($this->connectionConfig->snappy) {
$this->socket = new NsqSocket(
new SnappySocket(
$socket,
$this->logger,
),
);
$this->checkIsOK();
}
if ($this->connectionConfig->deflate) {
$this->socket = new NsqSocket(
new DeflateSocket(
$socket,
),
);
public function connect(): void
{
$this->reconnect->connect(function (): void {
try {
$this->socket = (new Factory())->createClient($this->address);
}
// @codeCoverageIgnoreStart
catch (Exception $e) {
$this->logger->error('Connecting to {address} failed.', ['address' => $this->address]);
$this->checkIsOK();
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
$this->socket->write(' V2');
$body = json_encode($this->clientConfig, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT);
$response = $this->command('IDENTIFY', data: $body)->response();
$this->connectionConfig = ConnectionConfig::fromArray($response->toArray());
if ($this->connectionConfig->snappy || $this->connectionConfig->deflate) {
$this->response()->okOrFail();
}
if ($this->connectionConfig->authRequired) {
if (null === $this->clientConfig->authSecret) {
throw new AuthenticationRequired();
throw new AuthenticationRequired('NSQ requires authorization, set ClientConfig::$authSecret before connecting');
}
$authResponse = $this
->command('AUTH', data: $this->clientConfig->authSecret)
->readResponse()
->toArray()
;
$authResponse = $this->command('AUTH', data: $this->clientConfig->authSecret)->response()->toArray();
$this->logger->info('Authorization response: '.http_build_query($authResponse));
}
});
}
/**
* Cleanly close your connection (no more messages are sent).
*/
public function close(): void
public function disconnect(): void
{
if ($this->closed) {
if (null === $this->socket) {
return;
}
try {
$this->command('CLS');
$this->socket->write('CLS'.PHP_EOL);
$this->socket->close();
} catch (\Throwable $e) {
}
// @codeCoverageIgnoreStart
catch (Exception $e) {
$this->logger->debug($e->getMessage(), ['exception' => $e]);
}
// @codeCoverageIgnoreEnd
$this->socket = null;
$this->connectionConfig = null;
}
$this->closed = true;
}
public function isClosed(): bool
public function isReady(): bool
{
return $this->closed;
return null !== $this->socket;
}
/**
@ -122,95 +133,211 @@ abstract class Connection
*/
protected function command(string $command, array | string $params = [], string $data = null): self
{
if ($this->closed) {
throw new NotConnected('Connection closed.');
$socket = $this->socket();
$buffer = [] === $params ? $command : implode(' ', [$command, ...((array) $params)]);
$buffer .= PHP_EOL;
if (null !== $data) {
$buffer .= pack('N', \strlen($data));
$buffer .= $data;
}
$command = [] === $params
? $command
: implode(' ', [$command, ...((array) $params)]);
$this->logger->debug('Prepare send uncompressed buffer: {bytes}', ['bytes' => addcslashes($buffer, PHP_EOL)]);
$this->logger->info('Command [{command}] with data [{data}]', ['command' => $command, 'data' => $data ?? 'null']);
if ($this->connectionConfig?->snappy) {
$identifierFrame = [0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59];
$compressedFrame = 0x00;
$uncompressedFrame = 0x01;
$this->socket->write($command, $data);
$chunk = snappy_compress($buffer);
[$chunk, $compressFrame] = match (\strlen($chunk) < \strlen($buffer)) {
true => [$chunk, $compressedFrame],
false => [$buffer, $uncompressedFrame],
};
$size = \strlen($chunk) + 4;
$buffer = new ByteBuffer();
foreach ([...$identifierFrame, $compressFrame, $size, $size >> 8, $size >> 16] as $byte) {
$buffer->appendUint8($byte);
}
$crc32c = hash('crc32c', $data, true);
$crc32c = unpack('V', $crc32c)[1];
$unsignedRightShift = static function ($a, $b) {
if ($b >= 32 || $b < -32) {
$m = (int) ($b / 32);
$b -= ($m * 32);
}
if ($b < 0) {
$b = 32 + $b;
}
if (0 === $b) {
return (($a >> 1) & 0x7fffffff) * 2 + (($a >> $b) & 1);
}
if ($a < 0) {
$a >>= 1;
$a &= 2147483647;
$a |= 0x40000000;
$a >>= ($b - 1);
} else {
$a >>= $b;
}
return $a;
};
$checksum = $unsignedRightShift((($crc32c >> 15) | ($crc32c << 17)) + 0xa282ead8, 0);
$buffer->appendUint32($checksum);
$buffer->append($chunk);
$buffer = $buffer->bytes();
}
$this->logger->debug('Prepare send compressed buffer: {bytes}', ['bytes' => addcslashes($buffer, PHP_EOL)]);
try {
$socket->write($buffer);
}
// @codeCoverageIgnoreStart
catch (Exception $e) {
$this->disconnect();
$this->logger->error($e->getMessage(), ['exception' => $e]);
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
return $this;
}
public function hasMessage(float $timeout): bool
public function hasMessage(float $timeout = 0): bool
{
if ($this->closed) {
throw new NotConnected('Connection closed.');
try {
return false !== $this->socket()->selectRead($timeout);
}
// @codeCoverageIgnoreStart
catch (Exception $e) {
$this->disconnect();
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
}
public function receive(float $timeout = null): ?Response
{
$socket = $this->socket();
$timeout ??= $this->clientConfig->readTimeout;
$deadline = microtime(true) + $timeout;
if (!$this->hasMessage($timeout)) {
return null;
}
try {
return false !== $this->socket->wait($timeout);
} catch (ConnectionFail $e) {
$this->close();
$size = $socket->read(Bytes::BYTES_SIZE);
throw $e;
}
if ('' === $size) {
$this->disconnect();
throw new ConnectionFail('Probably connection lost');
}
protected function readFrame(): Frame
{
if ($this->closed) {
throw new NotConnected('Connection closed.');
}
if ($this->connectionConfig?->snappy) {
$buffer = new ByteBuffer();
$snappyBuffer = new ByteBuffer($size);
while (true) {
$typeByte = \ord($snappyBuffer->consume(1));
$buffer = $this->socket->read();
$size = \ord($snappyBuffer->consume(1)) + (\ord($snappyBuffer->consume(1)) << 8) + (\ord($snappyBuffer->consume(1)) << 16);
$type = match ($typeByte) {
0xff => 'identifier',
0x00 => 'compressed',
0x01 => 'uncompressed',
0xfe => 'padding',
};
$this->logger->debug('Received snappy chunk: {type}, size: {size}', [
'type' => $type,
'size' => $size,
]);
switch ($typeByte) {
case 0xff: // 'identifier',
$socket->read($size);
$snappyBuffer->append($socket->read(4));
continue 2;
case 0x00: // 'compressed',
case 0x01: // 'uncompressed',
$uncompressed = $socket->read($size);
$this->logger->debug('Received uncompressed bytes: {bytes}', ['bytes' => $uncompressed]);
$buffer->append($uncompressed);
$buffer->consume(4); // slice snappy prefix
$buffer->consumeUint32(); // slice size
break 2;
case 0xfe:// 'padding',
}
}
} else {
$this->logger->debug('Size bytes received: "{bytes}"', ['bytes' => $size]);
$buffer = new ByteBuffer($size);
$size = $buffer->consumeUint32();
do {
$chunk = $socket->read($size);
$buffer->append($chunk);
$size -= \strlen($chunk);
} while (0 < $size);
}
$this->logger->debug('Received buffer: '.addcslashes($buffer->bytes(), PHP_EOL));
return match ($type = $buffer->consumeType()) {
0 => new Response($buffer->flush()),
1 => new Error($buffer->flush()),
2 => new Message(
timestamp: $buffer->consumeTimestamp(),
attempts: $buffer->consumeAttempts(),
id: $buffer->consumeId(),
body: $buffer->flush(),
consumer: $this instanceof Consumer ? $this : throw new NsqException('what?'),
),
default => throw new NsqException('Unexpected frame type: '.$type)
};
}
protected function checkIsOK(): void
{
$response = $this->readResponse();
$response = new Response($buffer);
if ($response->isHeartBeat()) {
$this->command('NOP');
$this->checkIsOK();
return $this->receive(
($currentTime = microtime(true)) > $deadline ? 0 : $deadline - $currentTime
);
}
}
// @codeCoverageIgnoreStart
catch (Exception $e) {
$this->disconnect();
return;
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
return $response;
}
if (!$response->isOk()) {
throw new BadResponse($response);
}
$this->logger->info('Ok checked.');
}
private function readResponse(): Response
protected function response(): Response
{
$frame = $this->readFrame();
if ($frame instanceof Response) {
return $frame;
return $this->receive() ?? throw UnexpectedResponse::null();
}
if ($frame instanceof Error) {
if ($frame->type->terminateConnection) {
$this->close();
private function socket(): Socket
{
if (null === $this->socket) {
$this->connect();
}
throw new NsqError($frame);
}
throw new NsqException('Unreachable statement.');
return $this->socket ?? throw new ConnectionFail('This connection is closed, create new one.');
}
}

86
src/Consumer.php
View File

@ -4,82 +4,16 @@ declare(strict_types=1);
namespace Nsq;
use Generator;
use Nsq\Config\ClientConfig;
use Nsq\Exception\NsqError;
use Nsq\Exception\NsqException;
use Nsq\Protocol\Error;
use Nsq\Protocol\Message;
use Nsq\Protocol\Response;
use Psr\Log\LoggerInterface;
final class Consumer extends Connection
{
private int $rdy = 0;
public function __construct(
private string $topic,
private string $channel,
string $address,
ClientConfig $clientConfig = null,
LoggerInterface $logger = null
) {
parent::__construct($address, $clientConfig, $logger);
}
/**
* @psalm-return Generator<int, Message|float|null, int|null, void>
* Subscribe to a topic/channel.
*/
public function generator(): \Generator
public function sub(string $topic, string $channel): void
{
$this->command('SUB', [$this->topic, $this->channel])->checkIsOK();
while (true) {
$this->rdy(1);
$timeout = $this->clientConfig->readTimeout;
do {
$deadline = microtime(true) + $timeout;
$message = $this->hasMessage($timeout) ? $this->readMessage() : null;
$timeout = ($currentTime = microtime(true)) > $deadline ? 0 : $deadline - $currentTime;
} while (0 < $timeout && null === $message);
$command = yield $message;
if (0 === $command) {
break;
}
}
$this->close();
}
public function readMessage(): ?Message
{
$frame = $this->readFrame();
if ($frame instanceof Message) {
return $frame;
}
if ($frame instanceof Response && $frame->isHeartBeat()) {
$this->command('NOP');
return null;
}
if ($frame instanceof Error) {
if ($frame->type->terminateConnection) {
$this->close();
}
throw new NsqError($frame);
}
throw new NsqException('Unreachable statement.');
$this->command('SUB', [$topic, $channel])->response()->okOrFail();
}
/**
@ -98,8 +32,6 @@ final class Consumer extends Connection
/**
* Finish a message (indicate successful processing).
*
* @internal
*/
public function fin(string $id): void
{
@ -109,12 +41,10 @@ final class Consumer extends Connection
}
/**
* Re-queue a message (indicate failure to process) The re-queued message is placed at the tail of the queue,
* equivalent to having just published it, but for various implementation specific reasons that behavior should not
* be explicitly relied upon and may change in the future. Similarly, a message that is in-flight and times out
* behaves identically to an explicit REQ.
*
* @internal
* Re-queue a message (indicate failure to process)
* The re-queued message is placed at the tail of the queue, equivalent to having just published it,
* but for various implementation specific reasons that behavior should not be explicitly relied upon and may change in the future.
* Similarly, a message that is in-flight and times out behaves identically to an explicit REQ.
*/
public function req(string $id, int $timeout): void
{
@ -125,8 +55,6 @@ final class Consumer extends Connection
/**
* Reset the timeout for an in-flight message.
*
* @internal
*/
public function touch(string $id): void
{

8
src/Exception/AuthenticationRequired.php
View File

@ -4,10 +4,8 @@ declare(strict_types=1);
namespace Nsq\Exception;
final class AuthenticationRequired extends NsqException
use RuntimeException;
final class AuthenticationRequired extends RuntimeException implements NsqException
{
public function __construct()
{
parent::__construct('NSQ requires authorization, set ClientConfig::$authSecret before connecting');
}
}

15
src/Exception/BadResponse.php
View File

@ -1,15 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Exception;
use Nsq\Protocol\Response;
final class BadResponse extends NsqException
{
public function __construct(Response $response)
{
parent::__construct($response->msg);
}
}

7
src/Exception/ConnectionFail.php
View File

@ -4,12 +4,15 @@ declare(strict_types=1);
namespace Nsq\Exception;
final class ConnectionFail extends NsqException
use RuntimeException;
use Throwable;
final class ConnectionFail extends RuntimeException implements NsqException
{
/**
* @codeCoverageIgnore
*/
public static function fromThrowable(\Throwable $throwable): self
public static function fromThrowable(Throwable $throwable): self
{
return new self($throwable->getMessage(), (int) $throwable->getCode(), $throwable);
}

5
src/Exception/MessageAlreadyFinished.php
View File

@ -4,9 +4,10 @@ declare(strict_types=1);
namespace Nsq\Exception;
use Nsq\Protocol\Message;
use Nsq\Message;
use RuntimeException;
final class MessageAlreadyFinished extends NsqException
final class MessageAlreadyFinished extends RuntimeException implements NsqException
{
public static function finish(Message $message): self
{

9
src/Exception/NotConnected.php
View File

@ -1,9 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Exception;
final class NotConnected extends NsqException
{
}

8
src/Exception/NsqError.php
View File

@ -4,12 +4,8 @@ declare(strict_types=1);
namespace Nsq\Exception;
use Nsq\Protocol\Error;
use RuntimeException;
final class NsqError extends NsqException
final class NsqError extends RuntimeException implements NsqException
{
public function __construct(Error $error)
{
parent::__construct($error->rawData);
}
}

4
src/Exception/NsqException.php
View File

@ -4,6 +4,8 @@ declare(strict_types=1);
namespace Nsq\Exception;
class NsqException extends \RuntimeException
use Throwable;
interface NsqException extends Throwable
{
}

9
src/Exception/NullReceived.php
View File

@ -1,9 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Exception;
final class NullReceived extends NsqException
{
}

18
src/Exception/UnexpectedResponse.php Normal file
View File

@ -0,0 +1,18 @@
<?php
declare(strict_types=1);
namespace Nsq\Exception;
use RuntimeException;
final class UnexpectedResponse extends RuntimeException implements NsqException
{
/**
* @codeCoverageIgnore
*/
public static function null(): self
{
return new self('Response was expected, but null received.');
}
}

14
src/Protocol/Message.php → src/Message.php
View File

@ -2,13 +2,11 @@
declare(strict_types=1);
namespace Nsq\Protocol;
namespace Nsq;
use Nsq\Bytes;
use Nsq\Consumer;
use Nsq\Exception\MessageAlreadyFinished;
final class Message extends Frame
final class Message
{
/**
* @psalm-readonly
@ -36,14 +34,6 @@ final class Message extends Frame
public function __construct(int $timestamp, int $attempts, string $id, string $body, Consumer $consumer)
{
parent::__construct(
Bytes::BYTES_TYPE
+ Bytes::BYTES_TIMESTAMP
+ Bytes::BYTES_ATTEMPTS
+ Bytes::BYTES_ID
+ \strlen($body)
);
$this->timestamp = $timestamp;
$this->attempts = $attempts;
$this->id = $id;

78
src/NsqSocket.php
View File

@ -1,78 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq;
use Nsq\Exception\ConnectionFail;
use Nsq\Socket\Socket;
use PHPinnacle\Buffer\ByteBuffer;
use Throwable;
final class NsqSocket
{
private Buffer $input;
private ByteBuffer $output;
public function __construct(
private Socket $socket,
) {
$this->input = new Buffer();
$this->output = new ByteBuffer();
}
public function write(string $command, string $data = null): void
{
$this->output->append($command.PHP_EOL);
if (null !== $data) {
$this->output->appendUint32(\strlen($data));
$this->output->append($data);
}
$this->socket->write($this->output->flush());
}
public function wait(float $timeout): bool
{
return $this->socket->selectRead($timeout);
}
public function read(): Buffer
{
$buffer = $this->input;
$size = Bytes::BYTES_SIZE;
do {
$buffer->append(
$this->socket->read($size),
);
$size -= $buffer->size();
} while ($buffer->size() < Bytes::BYTES_SIZE);
if ('' === $buffer->bytes()) {
throw new ConnectionFail('Probably connection closed.');
}
$size = $buffer->consumeSize();
do {
$buffer->append(
$this->socket->read($size - $buffer->size()),
);
} while ($buffer->size() < $size);
return $buffer;
}
public function close(): void
{
try {
$this->socket->close();
} catch (Throwable) {
}
}
}

35
src/Producer.php
View File

@ -4,38 +4,41 @@ declare(strict_types=1);
namespace Nsq;
use PHPinnacle\Buffer\ByteBuffer;
use function array_map;
use function implode;
use function pack;
/**
* @psalm-suppress PropertyNotSetInConstructor
*/
final class Producer extends Connection
{
/**
* @psalm-suppress PossiblyFalseOperand
*/
public function pub(string $topic, string $body): void
{
$this->command('PUB', $topic, $body)->checkIsOK();
$this->command('PUB', $topic, $body)->response()->okOrFail();
}
/**
* @psalm-param array<int, mixed> $bodies
* @psalm-param array<mixed, mixed> $bodies
*
* @psalm-suppress PossiblyFalseOperand
*/
public function mpub(string $topic, array $bodies): void
{
static $buffer;
$buffer ??= new ByteBuffer();
$num = pack('N', \count($bodies));
$buffer->appendUint32(\count($bodies));
$mb = implode('', array_map(static function ($body): string {
return pack('N', \strlen($body)).$body;
}, $bodies));
foreach ($bodies as $body) {
$buffer->appendUint32(\strlen($body));
$buffer->append($body);
}
$this->command('MPUB', $topic, $buffer->flush())->checkIsOK();
$this->command('MPUB', $topic, $num.$mb)->response()->okOrFail();
}
/**
* @psalm-suppress PossiblyFalseOperand
*/
public function dpub(string $topic, string $body, int $delay): void
{
$this->command('DPUB', [$topic, $delay], $body)->checkIsOK();
$this->command('DPUB', [$topic, $delay], $body)->response()->okOrFail();
}
}

22
src/Protocol/Error.php
View File

@ -1,22 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Protocol;
use Nsq\Bytes;
/**
* @psalm-immutable
*/
final class Error extends Frame
{
public ErrorType $type;
public function __construct(public string $rawData)
{
parent::__construct(\strlen($this->rawData) + Bytes::BYTES_TYPE);
$this->type = new ErrorType(explode(' ', $this->rawData)[0]);
}
}

100
src/Protocol/ErrorType.php
View File

@ -1,100 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Protocol;
/**
* @psalm-immutable
*/
final class ErrorType
{
/**
* A generic error type without any more hints.
*/
public const E_INVALID = true;
/**
* This error might be returned during multiple occasions. It can be returned for IDENTIFY, AUTH or MPUB messages.
* It is caused for payloads that do not meet certain requirements. For IDENTIFY and AUTH, this is usually a bug in
* the library and should be reported. For MPUB, this error can occur if the payload is larger than the maximum
* payload size specified in the nsqd config.
*/
public const E_BAD_BODY = true;
/**
* This error indicates that the topic sent to nsqd is not valid.
*/
public const E_BAD_TOPIC = true;
/**
* This error indicates that the channel sent to nsqd is not valid.
*/
public const E_BAD_CHANNEL = true;
/**
* This error is returned by nsqd if the message in the payload of a publishing operation does not meet the
* requirements of the server. This might be caused by too big payloads being sent to nsqd. You should consider
* adding a limit to the payload size or increasing it in the nsqd config.
*/
public const E_BAD_MESSAGE = true;
/**
* This error may happen if a error condition is met after validating the input on the nsqd side. This is usually a
* temporary error and can be caused by topics being added, deleted or cleared.
*/
public const E_PUB_FAILED = true;
/**
* This error may happen if a error condition is met after validating the input on the nsqd side. This is usually a
* temporary error and can be caused by topics being added, deleted or cleared.
*/
public const E_MPUB_FAILED = true;
/**
* This error may happen if a error condition is met after validating the input on the nsqd side. This is usually a
* temporary error and can be caused by topics being added, deleted or cleared.
*/
public const E_DPUB_FAILED = true;
/**
* This error may happen if a error condition is met after validating the input on the nsqd side. This can
* happen in particular for messages that are no longer queued on the server side.
*/
public const E_FIN_FAILED = false;
/**
* This error may happen if a error condition is met after validating the input on the nsqd side. This can
* happen in particular for messages that are no longer queued on the server side.
*/
public const E_REQ_FAILED = false;
/**
* This error may happen if a error condition is met after validating the input on the nsqd side. This can
* happen in particular for messages that are no longer queued on the server side.
*/
public const E_TOUCH_FAILED = false;
/**
* This error indicates that the authorization of the client failed on the server side. This might be related
* to connection issues to the authorization server. Depending on the authorization server implementation, this
* might also indicate that the given auth secret in the [ClientConfig] is not known on the server or the server
* denied authentication with the current connection properties (i.e. TLS status and IP).
*/
public const E_AUTH_FAILED = true;
/**
* This error happens if something breaks on the nsqd side while performing the authorization. This might be
* caused by bugs in nsqd, the authorization server or network issues.
*/
public const E_AUTH_ERROR = true;
/**
* This error is sent by nsqd if the client attempts an authentication, but the server does not support it. This
* should never happen using this library as authorization requests are only sent if the server supports it.
* It is safe to expect that this error is never thrown.
*/
public const E_AUTH_DISABLED = true;
/**
* This error indicates that the client related to the authorization secret set in the [ClientConfig] is not
* allowed to do the operation it tried to do.
*/
public const E_UNAUTHORIZED = true;
/**
* A boolean indicating whether or not an [Error] with this type terminates the connection or not.
*/
public bool $terminateConnection;
public function __construct(public string $type)
{
$this->terminateConnection = \constant('self::'.$this->type) ?? self::E_INVALID;
}
}

16
src/Protocol/Frame.php
View File

@ -1,16 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Protocol;
abstract class Frame
{
public function __construct(
/**
* @psalm-readonly
*/
public int $length,
) {
}
}

39
src/Protocol/Response.php
View File

@ -1,39 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Protocol;
use Nsq\Bytes;
/**
* @psalm-immutable
*/
final class Response extends Frame
{
public const OK = 'OK';
public const HEARTBEAT = '_heartbeat_';
public function __construct(public string $msg)
{
parent::__construct(\strlen($this->msg) + Bytes::BYTES_TYPE);
}
public function isOk(): bool
{
return self::OK === $this->msg;
}
public function isHeartBeat(): bool
{
return self::HEARTBEAT === $this->msg;
}
/**
* @return array<mixed, mixed>
*/
public function toArray(): array
{
return json_decode($this->msg, true, flags: JSON_THROW_ON_ERROR);
}
}

3
src/Reconnect/ExponentialStrategy.php
View File

@ -8,6 +8,7 @@ use Nsq\Exception\ConnectionFail;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Throwable;
final class ExponentialStrategy implements ReconnectStrategy
{
@ -46,7 +47,7 @@ final class ExponentialStrategy implements ReconnectStrategy
try {
$callable();
} catch (\Throwable $e) {
} catch (Throwable $e) {
$nextDelay = 0 === $this->delay ? $this->minDelay : $this->delay * 2;
$this->delay = $nextDelay > $this->maxDelay ? $this->maxDelay : $nextDelay;
$this->nextTryAfter = $currentTime + $this->delay;

87
src/Response.php Normal file
View File

@ -0,0 +1,87 @@
<?php
declare(strict_types=1);
namespace Nsq;
use Nsq\Exception\NsqError;
use Nsq\Exception\UnexpectedResponse;
use PHPinnacle\Buffer\ByteBuffer;
use function json_decode;
use function sprintf;
use const JSON_THROW_ON_ERROR;
final class Response
{
private const OK = 'OK';
private const HEARTBEAT = '_heartbeat_';
private const TYPE_RESPONSE = 0;
private const TYPE_ERROR = 1;
private const TYPE_MESSAGE = 2;
private int $type;
private ByteBuffer $buffer;
public function __construct(ByteBuffer $buffer)
{
$this->type = $buffer->consumeUint32();
$this->buffer = $buffer;
}
public function okOrFail(): void
{
if (self::TYPE_ERROR === $this->type) {
throw new NsqError($this->buffer->bytes());
}
if (self::TYPE_RESPONSE !== $this->type) {
// @codeCoverageIgnoreStart
throw new UnexpectedResponse(sprintf('"%s" type expected, but "%s" received.', self::TYPE_RESPONSE, $this->type));
// @codeCoverageIgnoreEnd
}
if (self::OK !== $this->buffer->bytes()) {
// @codeCoverageIgnoreStart
throw new UnexpectedResponse(sprintf('OK response expected, but "%s" received.', $this->buffer->bytes()));
// @codeCoverageIgnoreEnd
}
}
public function isHeartBeat(): bool
{
return self::TYPE_RESPONSE === $this->type && self::HEARTBEAT === $this->buffer->bytes();
}
/**
* @phpstan-ignore-next-line
*/
public function toArray(): array
{
if (self::TYPE_RESPONSE !== $this->type) {
// @codeCoverageIgnoreStart
throw new UnexpectedResponse(sprintf('"%s" type expected, but "%s" received.', self::TYPE_RESPONSE, $this->type));
// @codeCoverageIgnoreEnd
}
return json_decode($this->buffer->bytes(), true, flags: JSON_THROW_ON_ERROR);
}
public function toMessage(Consumer $reader): Message
{
if (self::TYPE_MESSAGE !== $this->type) {
// @codeCoverageIgnoreStart
throw new UnexpectedResponse(sprintf('Expecting "%s" type, but NSQ return: "%s"', self::TYPE_MESSAGE, $this->type));
// @codeCoverageIgnoreEnd
}
$buffer = new ByteBuffer($this->buffer->bytes());
$timestamp = $buffer->consumeInt64();
$attempts = $buffer->consumeUint16();
$id = $buffer->consume(Bytes::BYTES_ID);
$body = $buffer->flush();
return new Message($timestamp, $attempts, $id, $body, $reader);
}
}

45
src/Socket/DeflateSocket.php
View File

@ -1,45 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Socket;
final class DeflateSocket implements Socket
{
public function __construct(
private Socket $socket,
) {
}
/**
* {@inheritDoc}
*/
public function write(string $data): void
{
throw new \LogicException('not implemented.');
}
/**
* {@inheritDoc}
*/
public function read(int $length): string
{
throw new \LogicException('not implemented.');
}
/**
* {@inheritDoc}
*/
public function close(): void
{
throw new \LogicException('not implemented.');
}
/**
* {@inheritDoc}
*/
public function selectRead(float $timeout): bool
{
return $this->socket->selectRead($timeout);
}
}

81
src/Socket/RawSocket.php
View File

@ -1,81 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Socket;
use Nsq\Exception\ConnectionFail;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Socket\Raw\Exception;
use Socket\Raw\Factory;
use Socket\Raw\Socket as ClueSocket;
use Throwable;
final class RawSocket implements Socket
{
private ClueSocket $socket;
private LoggerInterface $logger;
public function __construct(string $address, LoggerInterface $logger = null)
{
$this->socket = (new Factory())->createClient($address);
$this->logger = $logger ?? new NullLogger();
}
/**
* {@inheritDoc}
*/
public function selectRead(float $timeout): bool
{
try {
return false !== $this->socket->selectRead($timeout);
} // @codeCoverageIgnoreStart
catch (Exception $e) {
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
}
/**
* {@inheritDoc}
*/
public function close(): void
{
try {
$this->socket->close();
} catch (Throwable) {
}
}
/**
* {@inheritDoc}
*/
public function write(string $data): void
{
try {
$this->socket->write($data);
} // @codeCoverageIgnoreStart
catch (Exception $e) {
$this->logger->error($e->getMessage(), ['exception' => $e]);
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
}
/**
* {@inheritDoc}
*/
public function read(int $length): string
{
try {
return $this->socket->read($length);
} // @codeCoverageIgnoreStart
catch (Exception $e) {
throw ConnectionFail::fromThrowable($e);
}
// @codeCoverageIgnoreEnd
}
}

162
src/Socket/SnappySocket.php
View File

@ -1,162 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Socket;
use PHPinnacle\Buffer\ByteBuffer;
use Psr\Log\LoggerInterface;
final class SnappySocket implements Socket
{
private ByteBuffer $output;
private ByteBuffer $input;
public function __construct(
private Socket $socket,
private LoggerInterface $logger,
) {
if (
!\function_exists('snappy_compress')
|| !\function_exists('snappy_uncompress')
|| !\extension_loaded('snappy')
) {
throw new \LogicException('Snappy extension not installed.');
}
$this->output = new ByteBuffer();
$this->input = new ByteBuffer();
}
/**
* {@inheritDoc}
*/
public function write(string $data): void
{
$identifierFrame = [0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59];
$compressedFrame = 0x00;
$uncompressedFrame = 0x01; // 11
$maxChunkLength = 65536;
$byteBuffer = new ByteBuffer();
foreach ($identifierFrame as $bite) {
$byteBuffer->appendUint8($bite);
}
foreach (str_split($data, $maxChunkLength) as $chunk) {
$compressedChunk = snappy_compress($chunk);
[$chunk, $chunkType] = \strlen($compressedChunk) <= 0.875 * \strlen($data)
? [$compressedChunk, $compressedFrame]
: [$data, $uncompressedFrame];
/** @var string $checksum */
$checksum = hash('crc32c', $data, true);
/** @phpstan-ignore-next-line */
$checksum = unpack('N', $checksum)[1];
$maskedChecksum = (($checksum >> 15) | ($checksum << 17)) + 0xa282ead8 & 0xffffffff;
$size = (\strlen($chunk) + 4) << 8;
$byteBuffer->append(pack('V', $chunkType + $size));
$byteBuffer->append(pack('V', $maskedChecksum));
$byteBuffer->append($chunk);
}
$this->socket->write($byteBuffer->flush());
}
/**
* {@inheritDoc}
*/
public function read(int $length): string
{
$output = $this->output;
$input = $this->input;
$this->logger->debug('Snappy requested {length} bytes.', ['length' => $length]);
while ($output->size() < $length) {
$this->logger->debug('Snappy enter loop');
/** @phpstan-ignore-next-line */
$chunkType = unpack('V', $this->socket->read(4))[1];
$size = $chunkType >> 8;
$chunkType &= 0xff;
$this->logger->debug('Snappy receive chunk [{chunk}], size [{size}]', [
'chunk' => $chunkType,
'size' => $size,
]);
do {
$input->append(
$this->socket->read($size),
);
$size -= $input->size();
} while ($input->size() < $size);
switch ($chunkType) {
case 0xff:
$this->logger->debug('Snappy identifier chunk');
$input->discard(6); // discard identifier body
break;
case 0x00: // 'compressed',
$this->logger->debug('Snappy compressed chunk');
$data = $input
->discard(4) // discard checksum
->flush()
;
$this->logger->debug('Snappy compressed data [{data}]', ['data' => $data]);
$output->append(snappy_uncompress($data));
break;
case 0x01: // 'uncompressed',
$this->logger->debug('Snappy uncompressed chunk');
$data = $input
->discard(4) // discard checksum
->flush()
;
$this->logger->debug('Snappy uncompressed data [{data}]', ['data' => $data]);
$output->append($data);
break;
case 0xfe:// 'padding',
$this->logger->debug('Snappy padding chunk');
break;
}
}
$this->logger->debug('Snappy return message [{message}]', ['message' => $output->read($length)]);
return $output->consume($length);
}
/**
* {@inheritDoc}
*/
public function close(): void
{
$this->socket->close();
}
/**
* {@inheritDoc}
*/
public function selectRead(float $timeout): bool
{
return !$this->input->empty() || $this->socket->selectRead($timeout);
}
}

27
src/Socket/Socket.php
View File

@ -1,27 +0,0 @@
<?php
declare(strict_types=1);
namespace Nsq\Socket;
use Nsq\Exception\ConnectionFail;
interface Socket
{
/**
* @throws ConnectionFail
*/
public function write(string $data): void;
/**
* @throws ConnectionFail
*/
public function read(int $length): string;
/**
* @throws ConnectionFail
*/
public function selectRead(float $timeout): bool;
public function close(): void;
}

39
src/Subscriber.php Normal file
View File

@ -0,0 +1,39 @@
<?php
declare(strict_types=1);
namespace Nsq;
use Generator;
final class Subscriber
{
public const STOP = 0;
private Consumer $reader;
public function __construct(Consumer $reader)
{
$this->reader = $reader;
}
/**
* @psalm-return Generator<int, Message|float|null, int|float|null, void>
*/
public function subscribe(string $topic, string $channel): Generator
{
$this->reader->sub($topic, $channel);
while (true) {
$this->reader->rdy(1);
$command = yield $this->reader->receive()?->toMessage($this->reader);
if (self::STOP === $command) {
break;
}
}
$this->reader->disconnect();
}
}

2
tests/MessageTest.php
View File

@ -4,7 +4,7 @@ declare(strict_types=1);
use Nsq\Consumer;
use Nsq\Exception\MessageAlreadyFinished;
use Nsq\Protocol\Message;
use Nsq\Message;
use PHPUnit\Framework\TestCase;
final class MessageTest extends TestCase

46
tests/NsqTest.php
View File

@ -4,28 +4,28 @@ declare(strict_types=1);
use Nsq\Config\ClientConfig;
use Nsq\Consumer;
use Nsq\Message;
use Nsq\Producer;
use Nsq\Protocol\Message;
use Nsq\Subscriber;
use Nyholm\NSA;
use PHPUnit\Framework\TestCase;
final class NsqTest extends TestCase
{
/**
* @dataProvider configs
*/
public function test(ClientConfig $clientConfig): void
public function test(): void
{
$producer = new Producer('tcp://localhost:4150');
$producer->pub(__FUNCTION__, __FUNCTION__);
$consumer = new Consumer(
topic: 'test',
channel: 'test',
address: 'tcp://localhost:4150',
clientConfig: $clientConfig,
clientConfig: new ClientConfig(
heartbeatInterval: 1000,
readTimeout: 1,
),
);
$generator = $consumer->generator();
$subscriber = new Subscriber($consumer);
$generator = $subscriber->subscribe(__FUNCTION__, __FUNCTION__);
/** @var null|Message $message */
$message = $generator->current();
@ -85,30 +85,8 @@ final class NsqTest extends TestCase
$message->touch();
$message->finish();
self::assertFalse($consumer->isClosed());
$generator->send(0);
self::assertTrue($consumer->isClosed());
}
/**
* @return Generator<string, array<int, ClientConfig>>
*/
public function configs(): Generator
{
yield 'default' => [
new ClientConfig(
heartbeatInterval: 3000,
snappy: false,
readTimeout: 1,
),
];
yield 'snappy' => [
new ClientConfig(
heartbeatInterval: 3000,
snappy: true,
readTimeout: 1,
),
];
self::assertTrue($consumer->isReady());
$generator->send(Subscriber::STOP);
self::assertFalse($consumer->isReady());
}
}

30
tests/Protocol/ErrorTypeTest.php
View File

@ -1,30 +0,0 @@
<?php
declare(strict_types=1);
namespace Protocol;
use Nsq\Protocol\ErrorType;
use PHPUnit\Framework\TestCase;
final class ErrorTypeTest extends TestCase
{
/**
* @dataProvider data
*/
public function testConstructor(string $type, bool $isConnectionTerminated): void
{
$errorType = new ErrorType($type);
self::assertSame($isConnectionTerminated, $errorType->terminateConnection);
}
/**
* @return \Generator<string, array<int, bool|string>>
*/
public function data(): \Generator
{
foreach ((new \ReflectionClass(ErrorType::class))->getConstants() as $constant => $isTerminated) {
yield $constant => [$constant, $isTerminated];
}
}
}