diff --git a/src/Exception/BadApiRequest.php b/src/Exception/BadApiRequest.php new file mode 100644 index 0000000..b9b1b8c --- /dev/null +++ b/src/Exception/BadApiRequest.php @@ -0,0 +1,10 @@ +error(new RequestHasNoResponders()); } elseif ($status === Status::NoMessages) { - $deferred->error(new MessageNotFound()); + // Direct get answers with a bare 404 status instead of an API error, so mimic the API one. + $deferred->error(new MessageNotFound(Status::NoMessages->value, 10037, 'no message found')); } else { $deferred->complete($delivery); } diff --git a/src/JetStream/Api/Result/Error.php b/src/JetStream/Api/Result/Error.php index 33542a2..9f008bf 100644 --- a/src/JetStream/Api/Result/Error.php +++ b/src/JetStream/Api/Result/Error.php @@ -4,11 +4,25 @@ namespace Thesis\Nats\JetStream\Api\Result; +use Thesis\Nats\Exception\BadApiRequest; +use Thesis\Nats\Exception\ConsumerAlreadyExists; use Thesis\Nats\Exception\ConsumerDoesNotExist; +use Thesis\Nats\Exception\ConsumerNameAlreadyInUse; use Thesis\Nats\Exception\ConsumerNotFound; +use Thesis\Nats\Exception\JetStreamApiException; +use Thesis\Nats\Exception\JetStreamNotEnabled; +use Thesis\Nats\Exception\JetStreamNotEnabledForAccount; +use Thesis\Nats\Exception\MaximumConsumersLimitReached; +use Thesis\Nats\Exception\MaximumStreamsLimitReached; use Thesis\Nats\Exception\MessageNotFound; +use Thesis\Nats\Exception\MessageSizeExceedsMaximum; use Thesis\Nats\Exception\StreamDoesNotMatch; +use Thesis\Nats\Exception\StreamNameAlreadyInUse; +use Thesis\Nats\Exception\StreamNameDoesNotMatchSubject; use Thesis\Nats\Exception\StreamNotFound; +use Thesis\Nats\Exception\StreamSealed; +use Thesis\Nats\Exception\StreamStoreFailed; +use Thesis\Nats\Exception\UnknownJetStreamApiError; use Thesis\Nats\Exception\WrongLastMessageId; use Thesis\Nats\Exception\WrongLastSequence; @@ -17,13 +31,34 @@ */ final readonly class Error { - private const int STREAM_NOT_FOUND = 10059; - private const int CONSUMER_NOT_FOUND = 10014; - private const int CONSUMER_DOES_NOT_EXISTS = 10149; - private const int STREAM_DOES_NOT_MATCH = 10060; - private const int WRONG_LAST_MESSAGE_ID = 10070; - private const int WRONG_LAST_SEQUENCE = 10071; - private const int MESSAGE_NOT_FOUND = 10037; + /** + * Codes are taken from https://github.com/nats-io/nats-server/blob/main/server/errors.json. + * + * @var array> + */ + private const array EXCEPTIONS = [ + 10003 => BadApiRequest::class, + 10013 => ConsumerNameAlreadyInUse::class, + 10014 => ConsumerNotFound::class, + 10026 => MaximumConsumersLimitReached::class, + 10027 => MaximumStreamsLimitReached::class, + 10037 => MessageNotFound::class, + 10039 => JetStreamNotEnabledForAccount::class, + 10054 => MessageSizeExceedsMaximum::class, + 10056 => StreamNameDoesNotMatchSubject::class, + 10058 => StreamNameAlreadyInUse::class, + 10059 => StreamNotFound::class, + 10060 => StreamDoesNotMatch::class, + 10070 => WrongLastMessageId::class, + 10071 => WrongLastSequence::class, + 10076 => JetStreamNotEnabled::class, + 10077 => StreamStoreFailed::class, + 10105 => ConsumerAlreadyExists::class, // consumer already exists and is still active + 10109 => StreamSealed::class, + 10148 => ConsumerAlreadyExists::class, + 10149 => ConsumerDoesNotExist::class, + 10164 => WrongLastSequence::class, // the same as 10071, but without the sequence in the description + ]; public function __construct( public int $code, @@ -31,17 +66,10 @@ public function __construct( public string $description, ) {} - public function exception(): \Exception + public function exception(): JetStreamApiException { - return match ($this->errCode) { - self::STREAM_NOT_FOUND => new StreamNotFound($this->description), - self::CONSUMER_NOT_FOUND => new ConsumerNotFound($this->description), - self::CONSUMER_DOES_NOT_EXISTS => new ConsumerDoesNotExist($this->description), - self::STREAM_DOES_NOT_MATCH => new StreamDoesNotMatch($this->description), - self::WRONG_LAST_MESSAGE_ID => new WrongLastMessageId($this->description), - self::WRONG_LAST_SEQUENCE => new WrongLastSequence($this->description), - self::MESSAGE_NOT_FOUND => new MessageNotFound($this->description), - default => new \RuntimeException("Nats error: '{$this->description} ({$this->errCode}) received'."), - }; + $exception = self::EXCEPTIONS[$this->errCode] ?? UnknownJetStreamApiError::class; + + return new $exception($this->code, $this->errCode, $this->description); } } diff --git a/tests/JetStream/Api/Result/ErrorTest.php b/tests/JetStream/Api/Result/ErrorTest.php new file mode 100644 index 0000000..5f953c4 --- /dev/null +++ b/tests/JetStream/Api/Result/ErrorTest.php @@ -0,0 +1,78 @@ + $class + */ + #[DataProvider('provideExceptionCases')] + public function testException(int $code, int $errCode, string $class): void + { + $e = (new Error($code, $errCode, 'description'))->exception(); + + self::assertSame($class, $e::class); + self::assertSame($code, $e->getCode()); + self::assertSame($errCode, $e->errCode); + self::assertSame('description', $e->description); + self::assertSame('description', $e->getMessage()); + } + + /** + * @return iterable}> + */ + public static function provideExceptionCases(): iterable + { + yield 'bad request' => [400, 10003, BadApiRequest::class]; + yield 'consumer name already in use' => [400, 10013, ConsumerNameAlreadyInUse::class]; + yield 'consumer not found' => [404, 10014, ConsumerNotFound::class]; + yield 'maximum consumers limit' => [400, 10026, MaximumConsumersLimitReached::class]; + yield 'maximum streams limit' => [400, 10027, MaximumStreamsLimitReached::class]; + yield 'no message found' => [404, 10037, MessageNotFound::class]; + yield 'not enabled for account' => [503, 10039, JetStreamNotEnabledForAccount::class]; + yield 'message size exceeds maximum' => [400, 10054, MessageSizeExceedsMaximum::class]; + yield 'stream name in subject does not match' => [400, 10056, StreamNameDoesNotMatchSubject::class]; + yield 'stream name already in use' => [400, 10058, StreamNameAlreadyInUse::class]; + yield 'stream not found' => [404, 10059, StreamNotFound::class]; + yield 'expected stream does not match' => [400, 10060, StreamDoesNotMatch::class]; + yield 'wrong last msg id' => [400, 10070, WrongLastMessageId::class]; + yield 'wrong last sequence' => [400, 10071, WrongLastSequence::class]; + yield 'not enabled' => [503, 10076, JetStreamNotEnabled::class]; + yield 'store failed' => [503, 10077, StreamStoreFailed::class]; + yield 'consumer exists and is active' => [400, 10105, ConsumerAlreadyExists::class]; + yield 'sealed' => [400, 10109, StreamSealed::class]; + yield 'consumer already exists' => [400, 10148, ConsumerAlreadyExists::class]; + yield 'consumer does not exist' => [400, 10149, ConsumerDoesNotExist::class]; + yield 'wrong last sequence without seq' => [400, 10164, WrongLastSequence::class]; + yield 'unknown' => [500, 10074, UnknownJetStreamApiError::class]; + } +} diff --git a/tests/JetStreamApiErrorsTest.php b/tests/JetStreamApiErrorsTest.php new file mode 100644 index 0000000..8467a1c --- /dev/null +++ b/tests/JetStreamApiErrorsTest.php @@ -0,0 +1,167 @@ +client()->jetStream(); + + try { + $js->streamInfo(generateUniqueId(10)); + self::fail('Expected the stream to be not found.'); + } catch (StreamNotFound $e) { + self::assertSame(404, $e->getCode()); + self::assertSame(10059, $e->errCode); + self::assertSame('stream not found', $e->description); + self::assertSame('stream not found', $e->getMessage()); + } + } + + public function testUnknownErrorIsUnknownJetStreamApiError(): void + { + $js = $this->client()->jetStream(); + + $stream = $js->createStream(new StreamConfig( + name: generateUniqueId(10), + subjects: [generateUniqueId(10)], + retention: RetentionPolicy::WorkQueue, + )); + + try { + // 10084 has no dedicated exception. + $stream->createConsumer(new ConsumerConfig(durableName: generateUniqueId(10), ackPolicy: AckPolicy::None)); + self::fail('Expected the consumer to be rejected.'); + } catch (JetStreamApiException $e) { + self::assertSame(UnknownJetStreamApiError::class, $e::class); + self::assertSame(400, $e->getCode()); + self::assertSame(10084, $e->errCode); + self::assertStringContainsString('requires explicit ack policy', $e->description); + } finally { + $stream->delete(); + } + } + + public function testStreamNameAlreadyInUse(): void + { + $js = $this->client()->jetStream(); + + $name = generateUniqueId(10); + $stream = $js->createStream(new StreamConfig($name, subjects: [generateUniqueId(10)])); + + try { + self::expectException(StreamNameAlreadyInUse::class); + $js->createStream(new StreamConfig($name, subjects: [generateUniqueId(10)])); + } finally { + $stream->delete(); + } + } + + public function testConsumerAlreadyExists(): void + { + $js = $this->client()->jetStream(); + + $stream = $js->createStream(new StreamConfig(generateUniqueId(10), subjects: [generateUniqueId(10)])); + $consumer = generateUniqueId(10); + $stream->createConsumer(new ConsumerConfig(durableName: $consumer)); + + try { + self::expectException(ConsumerAlreadyExists::class); + $stream->createConsumer(new ConsumerConfig(durableName: $consumer, description: 'changed')); + } finally { + $stream->delete(); + } + } + + public function testStreamSealed(): void + { + $js = $this->client()->jetStream(); + + $subject = generateUniqueId(10); + $config = new StreamConfig(generateUniqueId(10), subjects: [$subject]); + $stream = $js->createStream($config); + $js->publish($subject, new Message('x')); + $js->updateStream($config->seal()); + + try { + self::expectException(StreamSealed::class); + $stream->deleteMessage(1); + } finally { + $stream->delete(); + } + } + + public function testMessageSizeExceedsMaximum(): void + { + $js = $this->client()->jetStream(); + + $subject = generateUniqueId(10); + $stream = $js->createStream(new StreamConfig(generateUniqueId(10), subjects: [$subject], maxMessageSize: 10)); + + try { + self::expectException(MessageSizeExceedsMaximum::class); + $js->publish($subject, new Message(str_repeat('x', 100))); + } finally { + $stream->delete(); + } + } + + public function testStreamStoreFailed(): void + { + $js = $this->client()->jetStream(); + + $subject = generateUniqueId(10); + $stream = $js->createStream(new StreamConfig( + name: generateUniqueId(10), + subjects: [$subject], + discard: DiscardPolicy::New, + maxMessages: 1, + )); + $js->publish($subject, new Message('x')); + + try { + $js->publish($subject, new Message('x')); + self::fail('Expected the stream to reject the message.'); + } catch (StreamStoreFailed $e) { + self::assertSame(10077, $e->errCode); + self::assertSame('maximum messages exceeded', $e->description); + } finally { + $stream->delete(); + } + } + + public function testJetStreamNotEnabledForAccount(): void + { + // The system account has no JetStream. + $js = $this->clientWithConfig(new Config( + urls: $this->config()->urls, + user: 'admin', + password: 'Pswd1', + noResponders: true, + ))->jetStream(); + + self::expectException(JetStreamNotEnabledForAccount::class); + $js->accountInfo(); + } +}