diff --git a/src/JetStream/Consumer/Consumer.php b/src/JetStream/Consumer/Consumer.php index 5c22178..85ab8f3 100644 --- a/src/JetStream/Consumer/Consumer.php +++ b/src/JetStream/Consumer/Consumer.php @@ -13,13 +13,12 @@ final class Consumer implements ConsumerInterface { - private ?ConsumerInfo $cachedInfo = null; - /** @internal */ public function __construct( private readonly JetStreamContext $js, private readonly string $stream, private readonly ConsumerConfig $config, + private ?ConsumerInfo $cachedInfo = null, ) {} public function fetch(int $batch, ?FetchOptions $opts = null): MessageBatch @@ -223,11 +222,8 @@ public function info(): ConsumerInfo return $this->cachedInfo; } - public function cachedInfo(): ConsumerInfo + public function cachedInfo(): ?ConsumerInfo { - if ($this->cachedInfo === null) { - return $this->info(); - } return $this->cachedInfo; } diff --git a/src/JetStream/Consumer/ConsumerInterface.php b/src/JetStream/Consumer/ConsumerInterface.php index 848ada2..3e6317b 100644 --- a/src/JetStream/Consumer/ConsumerInterface.php +++ b/src/JetStream/Consumer/ConsumerInterface.php @@ -29,5 +29,11 @@ public function next(float $timeout = 5.0): JetStreamMessage; public function info(): ConsumerInfo; - public function cachedInfo(): ConsumerInfo; + /** + * Returns the cached consumer info without making a server request. + * + * Handles created without an initial info snapshot return null until info() + * is called or the handle is otherwise hydrated. + */ + public function cachedInfo(): ?ConsumerInfo; } diff --git a/src/JetStream/Consumer/OrderedConsumer.php b/src/JetStream/Consumer/OrderedConsumer.php index 878d88c..645ca02 100644 --- a/src/JetStream/Consumer/OrderedConsumer.php +++ b/src/JetStream/Consumer/OrderedConsumer.php @@ -105,9 +105,9 @@ public function info(): ConsumerInfo } } - public function cachedInfo(): ConsumerInfo + public function cachedInfo(): ?ConsumerInfo { - return $this->getInner()->cachedInfo(); + return $this->inner?->cachedInfo(); } private function getInner(): Consumer @@ -158,6 +158,7 @@ private function createOrderedConsumer(int $resumeStreamSequence): Consumer $this->js, $this->stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), ); } diff --git a/src/JetStream/Consumer/PushConsumer.php b/src/JetStream/Consumer/PushConsumer.php index b48222b..b176721 100644 --- a/src/JetStream/Consumer/PushConsumer.php +++ b/src/JetStream/Consumer/PushConsumer.php @@ -13,7 +13,6 @@ final class PushConsumer implements PushConsumerInterface { - private ?ConsumerInfo $cachedInfo = null; private ?Subscription $subscription = null; private PushConsumerStatus $status = PushConsumerStatus::Unsubscribed; private ?PushConsumeContext $consumeContext = null; @@ -27,6 +26,7 @@ public function __construct( private readonly JetStreamContext $js, private readonly string $stream, private readonly ConsumerConfig $config, + private ?ConsumerInfo $cachedInfo = null, ) { if ($this->config->deliverSubject === null) { throw new NatsException('Push consumer requires a deliver_subject in configuration'); @@ -95,11 +95,8 @@ public function info(): ConsumerInfo return $this->cachedInfo; } - public function cachedInfo(): ConsumerInfo + public function cachedInfo(): ?ConsumerInfo { - if ($this->cachedInfo === null) { - return $this->info(); - } return $this->cachedInfo; } diff --git a/src/JetStream/Consumer/PushConsumerInterface.php b/src/JetStream/Consumer/PushConsumerInterface.php index c966f66..b5c85c9 100644 --- a/src/JetStream/Consumer/PushConsumerInterface.php +++ b/src/JetStream/Consumer/PushConsumerInterface.php @@ -14,9 +14,12 @@ interface PushConsumerInterface public function info(): ConsumerInfo; /** - * Returns the last cached consumer info without making a server request. + * Returns the cached consumer info without making a server request. + * + * Handles created without an initial info snapshot return null until info() + * is called or the handle is otherwise hydrated. */ - public function cachedInfo(): ConsumerInfo; + public function cachedInfo(): ?ConsumerInfo; /** * Returns the consumer name. diff --git a/src/JetStream/JetStreamContext.php b/src/JetStream/JetStreamContext.php index d0a3ffe..64f14f8 100644 --- a/src/JetStream/JetStreamContext.php +++ b/src/JetStream/JetStreamContext.php @@ -163,7 +163,12 @@ public function createConsumer(string $stream, ConsumerConfig $config): Consumer "CONSUMER.CREATE.{$stream}", ['stream_name' => $stream, 'config' => $config->toArray()], ); - return new Consumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new Consumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } /** @@ -175,7 +180,12 @@ public function createOrUpdateConsumer(string $stream, ConsumerConfig $config): ? "CONSUMER.CREATE.{$stream}.{$config->name}" : "CONSUMER.CREATE.{$stream}"; $data = $this->apiRequest($op, ['stream_name' => $stream, 'config' => $config->toArray()]); - return new Consumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new Consumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } /** @@ -184,7 +194,12 @@ public function createOrUpdateConsumer(string $stream, ConsumerConfig $config): public function consumer(string $stream, string $name): ConsumerInterface { $data = $this->apiRequest("CONSUMER.INFO.{$stream}.{$name}"); - return new Consumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new Consumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function orderedConsumer(string $stream, ?OrderedConsumerConfig $config = null): ConsumerInterface @@ -212,7 +227,12 @@ public function createPushConsumer(string $stream, ConsumerConfig $config): Push $op, ['stream_name' => $stream, 'config' => $config->toArray()], ); - return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new PushConsumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function updatePushConsumer(string $stream, ConsumerConfig $config): PushConsumerInterface @@ -225,7 +245,12 @@ public function updatePushConsumer(string $stream, ConsumerConfig $config): Push "CONSUMER.CREATE.{$stream}.{$consumerName}", ['stream_name' => $stream, 'config' => $config->toArray()], ); - return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new PushConsumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function createOrUpdatePushConsumer(string $stream, ConsumerConfig $config): PushConsumerInterface @@ -235,13 +260,23 @@ public function createOrUpdatePushConsumer(string $stream, ConsumerConfig $confi ? "CONSUMER.CREATE.{$stream}.{$consumerName}" : "CONSUMER.CREATE.{$stream}"; $data = $this->apiRequest($op, ['stream_name' => $stream, 'config' => $config->toArray()]); - return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new PushConsumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function pushConsumer(string $stream, string $name): PushConsumerInterface { $data = $this->apiRequest("CONSUMER.INFO.{$stream}.{$name}"); - return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new PushConsumer( + $this, + $stream, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } // --- Publishing --- diff --git a/src/JetStream/Stream/Stream.php b/src/JetStream/Stream/Stream.php index 87df49a..2fd57c4 100644 --- a/src/JetStream/Stream/Stream.php +++ b/src/JetStream/Stream/Stream.php @@ -7,6 +7,7 @@ use Sofascore\Nats\Internal\TypeCast as T; use Sofascore\Nats\JetStream\Consumer\Consumer; use Sofascore\Nats\JetStream\Consumer\ConsumerConfig; +use Sofascore\Nats\JetStream\Consumer\ConsumerInfo; use Sofascore\Nats\JetStream\Consumer\ConsumerInterface; use Sofascore\Nats\JetStream\Consumer\ConsumerPauseResponse; use Sofascore\Nats\JetStream\Consumer\OrderedConsumer; @@ -34,7 +35,12 @@ public function createConsumer(ConsumerConfig $config): ConsumerInterface "CONSUMER.CREATE.{$this->name}", ['stream_name' => $this->name, 'config' => $config->toArray()], ); - return new Consumer($this->js, $this->name, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new Consumer( + $this->js, + $this->name, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function createOrUpdateConsumer(ConsumerConfig $config): ConsumerInterface @@ -47,7 +53,12 @@ public function createOrUpdateConsumer(ConsumerConfig $config): ConsumerInterfac $operation, ['stream_name' => $this->name, 'config' => $config->toArray()], ); - return new Consumer($this->js, $this->name, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new Consumer( + $this->js, + $this->name, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function updateConsumer(ConsumerConfig $config): ConsumerInterface @@ -58,7 +69,12 @@ public function updateConsumer(ConsumerConfig $config): ConsumerInterface public function consumer(string $name): ConsumerInterface { $data = $this->js->apiRequest("CONSUMER.INFO.{$this->name}.{$name}"); - return new Consumer($this->js, $this->name, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? []))); + return new Consumer( + $this->js, + $this->name, + ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), + ConsumerInfo::fromArray(T::stringKeyArray($data)), + ); } public function orderedConsumer(?OrderedConsumerConfig $config = null): ConsumerInterface diff --git a/src/JetStream/Stream/StreamInterface.php b/src/JetStream/Stream/StreamInterface.php index 7b4cd54..dd8c585 100644 --- a/src/JetStream/Stream/StreamInterface.php +++ b/src/JetStream/Stream/StreamInterface.php @@ -58,5 +58,10 @@ public function purge(?StreamPurgeOptions $options = null): int; public function info(bool $deletedDetails = false, ?string $subjectFilter = null): StreamInfo; + /** + * Returns the cached stream info without making a server request. + * + * Stream handles are created with an initial info snapshot already loaded. + */ public function cachedInfo(): StreamInfo; } diff --git a/tests/Integration/JetStream/ConsumerTest.php b/tests/Integration/JetStream/ConsumerTest.php index 32a5b49..1d8068a 100644 --- a/tests/Integration/JetStream/ConsumerTest.php +++ b/tests/Integration/JetStream/ConsumerTest.php @@ -9,10 +9,12 @@ use Sofascore\Nats\Enum\AckPolicy; use Sofascore\Nats\Enum\DeliveryPolicy; use Sofascore\Nats\Enum\StorageType; +use Sofascore\Nats\JetStream\Consumer\Consumer; use Sofascore\Nats\JetStream\Consumer\ConsumerConfig; use Sofascore\Nats\JetStream\Consumer\MessagePollOptions; use Sofascore\Nats\JetStream\Consumer\FetchOptions; use Sofascore\Nats\JetStream\Consumer\OrderedConsumerConfig; +use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\JetStreamContext; use Sofascore\Nats\JetStream\Stream\StreamConfig; use Sofascore\Nats\JetStream\Stream\StreamInterface; @@ -60,6 +62,8 @@ public function testCreateConsumer(): void deliverPolicy: DeliveryPolicy::All, )); + self::assertSame('test-consumer-1', $consumer->cachedInfo()?->name); + $info = $consumer->info(); self::assertSame('TEST_CONSUMER', $info->stream); self::assertSame('test-consumer-1', $info->name); @@ -331,6 +335,57 @@ public function testConsumerInfo(): void self::assertSame(0, $info->numAckPending); } + public function testCachedInfoHydratesColdConsumerHandleOnce(): void + { + $this->stream->createOrUpdateConsumer(new ConsumerConfig( + name: 'test-cached-cold', + durable: 'test-cached-cold', + ackPolicy: AckPolicy::Explicit, + )); + + $consumer = new Consumer( + $this->js, + 'TEST_CONSUMER', + new ConsumerConfig( + name: 'test-cached-cold', + durable: 'test-cached-cold', + ackPolicy: AckPolicy::Explicit, + ), + ); + + self::assertNull($consumer->cachedInfo()); + + $info = $consumer->info(); + self::assertSame('test-cached-cold', $info->name); + self::assertSame('TEST_CONSUMER', $info->stream); + + $this->stream->deleteConsumer('test-cached-cold'); + + self::assertSame('test-cached-cold', $consumer->cachedInfo()?->name); + + $this->expectException(JetStreamException::class); + $consumer->info(); + } + + public function testStreamConsumerLookupSeedsCachedInfo(): void + { + $this->stream->createOrUpdateConsumer(new ConsumerConfig( + name: 'test-cached-lookup', + durable: 'test-cached-lookup', + ackPolicy: AckPolicy::Explicit, + )); + + $consumer = $this->stream->consumer('test-cached-lookup'); + self::assertSame('test-cached-lookup', $consumer->cachedInfo()?->name); + + $this->stream->deleteConsumer('test-cached-lookup'); + + $info = $consumer->cachedInfo(); + self::assertNotNull($info); + self::assertSame('test-cached-lookup', $info->name); + self::assertSame('TEST_CONSUMER', $info->stream); + } + public function testListConsumers(): void { $this->stream->createOrUpdateConsumer(new ConsumerConfig( diff --git a/tests/Integration/JetStream/OrderedConsumerTest.php b/tests/Integration/JetStream/OrderedConsumerTest.php index 0c76935..9626123 100644 --- a/tests/Integration/JetStream/OrderedConsumerTest.php +++ b/tests/Integration/JetStream/OrderedConsumerTest.php @@ -69,6 +69,31 @@ public function testOrderedConsumerCarriesForwardCursorAcrossRecreation(): void self::assertNotSame($initialName, $consumer->cachedInfo()->name); } + public function testOrderedConsumerCachedInfoIsCacheOnly(): void + { + $byFetch = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + )); + self::assertNull($byFetch->cachedInfo()); + + $first = $byFetch->next(1.0); + self::assertSame(1, $first->metadata()->streamSequence); + + $fetchInfo = $byFetch->cachedInfo(); + self::assertNotNull($fetchInfo); + self::assertSame('TEST_ORDERED_CONSUMER', $fetchInfo->stream); + self::assertSame($fetchInfo->name, $first->metadata()->consumer); + + $byInfo = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + )); + self::assertNull($byInfo->cachedInfo()); + + $info = $byInfo->info(); + self::assertSame($info->name, $byInfo->cachedInfo()?->name); + self::assertSame('TEST_ORDERED_CONSUMER', $info->stream); + } + public function testOrderedConsumerRecreatesWhenSequenceTrackingDetectsLoss(): void { $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( diff --git a/tests/Integration/JetStream/PushConsumerTest.php b/tests/Integration/JetStream/PushConsumerTest.php index 2d456d4..872aa93 100644 --- a/tests/Integration/JetStream/PushConsumerTest.php +++ b/tests/Integration/JetStream/PushConsumerTest.php @@ -10,6 +10,7 @@ use Sofascore\Nats\Inbox; use Sofascore\Nats\JetStream\Consumer\ConsumerConfig; use Sofascore\Nats\JetStream\Consumer\PushConsumeOptions; +use Sofascore\Nats\JetStream\Consumer\PushConsumer; use Sofascore\Nats\JetStream\Consumer\PushConsumerStatus; use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\Error\PushConsumerException; @@ -239,6 +240,8 @@ public function testStreamHandlePushConsumerLookupAndDelete(): void self::assertSame('stream-push', $created->name()); self::assertSame($deliverSubject, $created->deliverSubject()); + self::assertSame('stream-push', $created->cachedInfo()?->name); + self::assertSame($deliverSubject, $created->cachedInfo()?->config->deliverSubject); $fetched = $this->stream->pushConsumer('stream-push'); self::assertSame('stream-push', $fetched->name()); @@ -250,6 +253,64 @@ public function testStreamHandlePushConsumerLookupAndDelete(): void $this->stream->pushConsumer('stream-push'); } + public function testCachedInfoHydratesColdPushConsumerHandleOnce(): void + { + $deliverSubject = Inbox::generate(); + + $this->stream->createPushConsumer(new ConsumerConfig( + name: 'push-cached-cold', + durable: 'push-cached-cold', + deliverSubject: $deliverSubject, + ackPolicy: AckPolicy::Explicit, + )); + + $consumer = new PushConsumer( + $this->js, + 'TEST_PUSH', + new ConsumerConfig( + name: 'push-cached-cold', + durable: 'push-cached-cold', + deliverSubject: $deliverSubject, + ackPolicy: AckPolicy::Explicit, + ), + ); + + self::assertNull($consumer->cachedInfo()); + + $info = $consumer->info(); + self::assertSame('push-cached-cold', $info->name); + self::assertSame($deliverSubject, $info->config->deliverSubject); + + $this->stream->deleteConsumer('push-cached-cold'); + + self::assertSame('push-cached-cold', $consumer->cachedInfo()?->name); + + $this->expectException(JetStreamException::class); + $consumer->info(); + } + + public function testStreamPushConsumerLookupSeedsCachedInfo(): void + { + $deliverSubject = Inbox::generate(); + + $this->stream->createPushConsumer(new ConsumerConfig( + name: 'push-cached-lookup', + durable: 'push-cached-lookup', + deliverSubject: $deliverSubject, + ackPolicy: AckPolicy::Explicit, + )); + + $consumer = $this->stream->pushConsumer('push-cached-lookup'); + self::assertSame('push-cached-lookup', $consumer->cachedInfo()?->name); + + $this->stream->deleteConsumer('push-cached-lookup'); + + $info = $consumer->cachedInfo(); + self::assertNotNull($info); + self::assertSame('push-cached-lookup', $info->name); + self::assertSame($deliverSubject, $info->config->deliverSubject); + } + public function testStreamHandleUpdatePushConsumer(): void { $deliverSubject = Inbox::generate();