diff --git a/README.md b/README.md index c727043..cb3d6c3 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,7 @@ For full runnable scripts grouped by feature, see the example files below. That | **Event Handlers** | onConnect, onDisconnect, onReconnect, onClose, onError, onLameDuckMode, onDiscoveredServers | | **JetStream Streams** | Create, createOrUpdate, update, delete, purge, info (with filters), list, getMessage, directGet, secureDelete, mirrors, sources, subject transforms | | **JetStream Consumers** | Pull (fetch/fetchBytes/fetchNoWait/next/poll/consumeLoop), push, ordered, durable, pause/resume, priority groups | -| **JetStream Publish** | Sync/async (with pending tracking), deduplication (msgId), per-message TTL, optimistic concurrency, retry, stallWait | +| **JetStream Publish** | Synchronous publish acknowledgements, deduplication (msgId), per-message TTL, optimistic concurrency, retry, stallWait | | **Key-Value Store** | Put, get, getRevision, create (CAS + TTL), update (CAS), delete, purge (+ TTL), purgeDeletes, history, watch, watchFiltered, listKeys, listKeysFiltered, bucket CRUD | | **Object Store** | Put (bytes/string/file), get, getInfo, delete, list (+ showDeleted), addLink, addBucketLink, seal, watch, chunked transfer, metadata, compression | | **Micro Services** | Endpoints, groups, discovery ($SRV.PING/INFO/STATS), statistics, pending limits, handler classes | @@ -62,7 +62,6 @@ For full runnable scripts grouped by feature, see the example files below. That - **`readonly` classes** for all immutable data (configs, entries, info, metadata) - **Backed string enums** for all policy types — JSON serialize/deserialize friendly -- **Future-based** async publish acknowledgements (`publishAsync` / `PubAckFuture`) - **Generator/IteratorAggregate** for streaming, watch APIs, and pagination - **Context-based** consumption APIs (`consume`, `messages`, `watch`) - **`stream_socket_client`** for TCP — zero dependencies, select-driven reads, TLS @@ -92,10 +91,10 @@ For full runnable scripts grouped by feature, see the example files below. That |---------|-------------| | [01-streams.php](docs/examples/jetstream/01-streams.php) | Create, info, getMessage, purge, list, delete | | [02-consumers.php](docs/examples/jetstream/02-consumers.php) | Pull fetch/next/messages/consume, ConsumeContext, ordered, ACK variants | -| [03-publish.php](docs/examples/jetstream/03-publish.php) | Deduplication, optimistic concurrency, async publish | +| [03-publish.php](docs/examples/jetstream/03-publish.php) | Deduplication, optimistic concurrency, sync publish | | [04-streams-advanced.php](docs/examples/jetstream/04-streams-advanced.php) | CreateOrUpdate, streamNameBySubject, secureDelete, mirrors, sources, subject transforms | | [05-consumers-advanced.php](docs/examples/jetstream/05-consumers-advanced.php) | Pause/resume, push consumer CRUD/consume, backoff, fetchBytes, headersOnly | -| [06-publish-advanced.php](docs/examples/jetstream/06-publish-advanced.php) | Async lifecycle, msgTtl, expectLastMsgId, stallWait, publishMessage | +| [06-publish-advanced.php](docs/examples/jetstream/06-publish-advanced.php) | msgTtl, expectLastMsgId, stallWait, publishMessage | | [07-direct-get.php](docs/examples/jetstream/07-direct-get.php) | Direct message access, account info, fast lookups | ### Key-Value @@ -405,15 +404,10 @@ $js->publish('orders.new', $data, PublishOptions::msgId('order-1')); $js->publish('orders.new', $data, PublishOptions::expectLastSequence(42)); $js->publish('orders.new', $data, PublishOptions::expectStream('ORDERS')); -// Async -$future = $js->publishAsync('orders.new', $data); -$conn->process(0.5); -$ack = $future->ok(); +// Synchronous publish acknowledgements +$ack = $js->publish('orders.new', $data); ``` -`publishAsync()` returns immediately and the future completes when the connection -processes incoming replies via `process()`, `wait()`, or your own application loop. - ### Key-Value Store ```php @@ -549,7 +543,7 @@ src/ ├── JetStream/ # Streams, consumers, publish, iterators │ ├── Stream/ # StreamConfig, StreamInfo, Stream │ ├── Consumer/ # ConsumerConfig, fetch/consume/messages -│ ├── Publish/ # PubAck, PubAckFuture, PublishOptions +│ ├── Publish/ # PubAck, PublishOptions │ ├── Message/ # JetStreamMessage (ack/nak/term) │ └── Iterator/ # Paginated list iterators ├── KeyValue/ # KV store, watch, history diff --git a/docs/examples/jetstream/03-publish.php b/docs/examples/jetstream/03-publish.php index f416abb..985cb6c 100644 --- a/docs/examples/jetstream/03-publish.php +++ b/docs/examples/jetstream/03-publish.php @@ -3,7 +3,7 @@ /** * JetStream Publish - advanced publishing options. * - * Deduplication, optimistic concurrency, async publish, retry. + * Deduplication, optimistic concurrency, retry. */ require_once __DIR__ . '/../../../vendor/autoload.php'; @@ -74,25 +74,6 @@ ); echo "Multi-option publish: seq={$ack->sequence}\n"; -// --- Async publish --- -// publishAsync() only sends the request and returns a future. -// The future completes once the connection processes incoming replies. -// No extra runtime is required here - calling process() is enough. - -$future1 = $js->publishAsync('orders.new', '{"id": 5}'); -$future2 = $js->publishAsync('orders.new', '{"id": 6}'); -$future3 = $js->publishAsync('orders.new', '{"id": 7}'); - -// Do something else while the publish ACKs are coming back... -// In a larger app this can be your main loop or wait(). -$conn->process(0.5); - -// Check results -if ($future1->isComplete()) { - $ack = $future1->ok(); - echo "Async 1: seq={$ack->sequence}\n"; -} - // Cleanup $js->deleteStream('ORDERS'); $conn->close(); diff --git a/docs/examples/jetstream/06-publish-advanced.php b/docs/examples/jetstream/06-publish-advanced.php index 08e370d..528fdeb 100644 --- a/docs/examples/jetstream/06-publish-advanced.php +++ b/docs/examples/jetstream/06-publish-advanced.php @@ -3,8 +3,7 @@ /** * Advanced JetStream publishing options. * - * Demonstrates: publishAsync lifecycle (publishAsyncPending, publishAsyncComplete, - * cleanupPublisher), expectLastMsgId, expectLastSequenceForSubject, + * Demonstrates: expectLastMsgId, expectLastSequenceForSubject, * stallWait, msgTtl (per-message TTL), publishMessage with headers. */ @@ -154,54 +153,7 @@ ); echo "publishMessage with msg ID: seq={$ack->sequence}\n"; -// --- 9. Async publish lifecycle --- -// publishAsync() is "send now, observe ACK later". -// ACKs arrive only while the connection is being processed. -// You do not need any extra runtime here; process() is enough. - -// Send multiple messages asynchronously -$futures = []; -for ($i = 1; $i <= 10; $i++) { - $futures[] = $js->publishAsync( - 'events.batch', - json_encode(['batch_item' => $i]), - PublishOptions::msgId("batch-{$i}"), - ); -} - -// Check how many are pending -echo "Async pending: {$js->publishAsyncPending()}\n"; -echo "Async complete: " . ($js->publishAsyncComplete() ? 'yes' : 'no') . "\n"; - -// Process responses -// In a real app this can be part of your regular connection loop. -$conn->flush(); -$conn->process(2.0); - -// Check results -$successCount = 0; -$errorCount = 0; -foreach ($futures as $future) { - if ($future->isComplete()) { - $err = $future->error(); - if ($err !== null) { - echo "Async error: {$err->getMessage()}\n"; - $errorCount++; - } else { - $ack = $future->ok(); - $successCount++; - } - } -} - -echo "Async results: {$successCount} ok, {$errorCount} errors\n"; -echo "All complete: " . ($js->publishAsyncComplete() ? 'yes' : 'no') . "\n"; - -// Clean up completed futures from memory -$js->cleanupPublisher(); -echo "Publisher cleanup done, pending: {$js->publishAsyncPending()}\n"; - -// --- 10. Combining multiple PublishOptions --- +// --- 9. Combining multiple PublishOptions --- $ack = $js->publish( 'events.combined', diff --git a/src/JetStream/JetStreamContext.php b/src/JetStream/JetStreamContext.php index 64f14f8..7c083d4 100644 --- a/src/JetStream/JetStreamContext.php +++ b/src/JetStream/JetStreamContext.php @@ -18,7 +18,6 @@ use Sofascore\Nats\JetStream\Consumer\PushConsumerInterface; use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\Publish\PubAck; -use Sofascore\Nats\JetStream\Publish\PubAckFuture; use Sofascore\Nats\JetStream\Publish\PublishOptions; use Sofascore\Nats\JetStream\Stream\Stream; use Sofascore\Nats\JetStream\Stream\StreamConfig; @@ -29,9 +28,6 @@ final class JetStreamContext { - /** @var list */ - private array $pendingFutures = []; - /** @internal */ public function __construct( private readonly Connection $conn, @@ -354,119 +350,6 @@ public function publishMessage(Message $msg, PublishOptions ...$opts): PubAck return PubAck::fromArray($response); } - public function publishAsync(string $subject, string $data = '', PublishOptions ...$opts): PubAckFuture - { - $headers = new Headers(); - PublishOptions::applyToHeaders($headers, array_values($opts)); - - $msg = $headers->count() > 0 - ? new Message(subject: $subject, data: $data, headers: $headers) - : new Message(subject: $subject, data: $data); - - $future = new PubAckFuture($msg); - $this->pendingFutures[] = $future; - - $inbox = $this->conn->newInbox(); - $sub = $this->conn->subscribe($inbox, function (Message $reply) use ($future, &$sub): void { - try { - $response = T::stringKeyArray(json_decode($reply->data, true, 512, JSON_THROW_ON_ERROR)); - if (isset($response['error'])) { - $future->reject(JetStreamException::fromApiResponse($response)); - } else { - $future->resolve(PubAck::fromArray($response)); - } - } catch (\Throwable $e) { - $future->reject($e); - } - if ($sub !== null) { $sub->unsubscribe(); } - }); - $sub->autoUnsubscribe(1); - - $this->conn->publish($subject, $data, $inbox); - - return $future; - } - - /** - * Publish a Message object asynchronously with headers support. - * Matches Go client PublishMsgAsync. - */ - public function publishAsyncMessage(Message $msg, PublishOptions ...$opts): PubAckFuture - { - $headers = $msg->headers !== null ? clone $msg->headers : new Headers(); - PublishOptions::applyToHeaders($headers, array_values($opts)); - - $publishMsg = $headers->count() > 0 - ? new Message(subject: $msg->subject, data: $msg->data, headers: $headers) - : new Message(subject: $msg->subject, data: $msg->data); - - $future = new PubAckFuture($publishMsg); - $this->pendingFutures[] = $future; - - $inbox = $this->conn->newInbox(); - $sub = $this->conn->subscribe($inbox, function (Message $reply) use ($future, &$sub): void { - try { - $response = T::stringKeyArray(json_decode($reply->data, true, 512, JSON_THROW_ON_ERROR)); - if (isset($response['error'])) { - $future->reject(JetStreamException::fromApiResponse($response)); - } else { - $future->resolve(PubAck::fromArray($response)); - } - } catch (\Throwable $e) { - $future->reject($e); - } - if ($sub !== null) { $sub->unsubscribe(); } - }); - $sub->autoUnsubscribe(1); - - $this->conn->publishMessage(new Message( - subject: $msg->subject, - data: $msg->data, - replyTo: $inbox, - headers: $headers->count() > 0 ? $headers : null, - )); - - return $future; - } - - /** - * Returns the number of pending (incomplete) async publish operations. - */ - public function publishAsyncPending(): int - { - $pending = 0; - foreach ($this->pendingFutures as $future) { - if (!$future->isComplete()) { - $pending++; - } - } - return $pending; - } - - /** - * Returns true if all tracked async publish operations have completed. - */ - public function publishAsyncComplete(): bool - { - foreach ($this->pendingFutures as $future) { - if (!$future->isComplete()) { - return false; - } - } - return true; - } - - /** - * Remove completed futures from internal tracking to free memory. - */ - public function cleanupPublisher(): void - { - $this->pendingFutures = array_values(array_filter( - $this->pendingFutures, - static fn(PubAckFuture $f) => !$f->isComplete(), - )); - } - // --- KeyValue --- public function createKeyValue(\Sofascore\Nats\KeyValue\KeyValueConfig $config): \Sofascore\Nats\KeyValue\KeyValue diff --git a/src/JetStream/JetStreamOptions.php b/src/JetStream/JetStreamOptions.php index 8b0900b..81f690d 100644 --- a/src/JetStream/JetStreamOptions.php +++ b/src/JetStream/JetStreamOptions.php @@ -10,8 +10,6 @@ public function __construct( public ?string $apiPrefix = null, public ?string $domain = null, public float $timeout = 5.0, - public int $publishAsyncMaxPending = 4096, - public ?\Closure $publishAsyncErrHandler = null, public int $apiRetries = 0, public float $apiRetryWait = 0.25, ) { diff --git a/src/JetStream/Publish/PubAckFuture.php b/src/JetStream/Publish/PubAckFuture.php deleted file mode 100644 index 50d30a4..0000000 --- a/src/JetStream/Publish/PubAckFuture.php +++ /dev/null @@ -1,58 +0,0 @@ -result !== null) { - return $this->result; - } - if ($this->error !== null) { - throw $this->error; - } - throw new \LogicException('PubAckFuture is not complete; process the connection before calling ok()'); - } - - public function error(): ?\Throwable - { - return $this->error; - } - - public function message(): Message - { - return $this->originalMessage; - } - - public function isComplete(): bool - { - return $this->complete; - } - - /** @internal */ - public function resolve(PubAck $ack): void - { - $this->result = $ack; - $this->complete = true; - } - - /** @internal */ - public function reject(\Throwable $error): void - { - $this->error = $error; - $this->complete = true; - } -} diff --git a/tests/Integration/JetStream/PublishAsyncMessageTest.php b/tests/Integration/JetStream/PublishAsyncMessageTest.php deleted file mode 100644 index 19c39da..0000000 --- a/tests/Integration/JetStream/PublishAsyncMessageTest.php +++ /dev/null @@ -1,84 +0,0 @@ -conn = Connection::connect(self::natsUrl()); - $this->js = $this->conn->jetStream(); - try { $this->js->deleteStream('TEST_ASYNC_MSG'); } catch (\Throwable) {} - } - - protected function tearDown(): void - { - try { $this->js->deleteStream('TEST_ASYNC_MSG'); } catch (\Throwable) {} - $this->conn->close(); - } - - public function testPublishAsyncMessageWithHeaders(): void - { - $this->js->createStream(new StreamConfig( - name: 'TEST_ASYNC_MSG', - subjects: ['asyncmsg.>'], - )); - - $msg = new Message( - subject: 'asyncmsg.test', - data: 'async-payload', - headers: new Headers(['X-Custom' => 'value123']), - ); - - $future = $this->js->publishAsyncMessage($msg); - - // Process to get the ack - $deadline = microtime(true) + 5.0; - while (!$future->isComplete() && microtime(true) < $deadline) { - $this->conn->process(0.05); - } - - self::assertTrue($future->isComplete(), 'Future should be complete'); - $ack = $future->ok(); - self::assertSame('TEST_ASYNC_MSG', $ack->stream); - self::assertGreaterThan(0, $ack->sequence); - } - - public function testPublishAsyncMessageWithoutHeaders(): void - { - $this->js->createStream(new StreamConfig( - name: 'TEST_ASYNC_MSG', - subjects: ['asyncmsg.>'], - )); - - $msg = new Message( - subject: 'asyncmsg.plain', - data: 'no-headers', - ); - - $future = $this->js->publishAsyncMessage($msg); - - $deadline = microtime(true) + 5.0; - while (!$future->isComplete() && microtime(true) < $deadline) { - $this->conn->process(0.05); - } - - self::assertTrue($future->isComplete()); - self::assertGreaterThan(0, $future->ok()->sequence); - } -} diff --git a/tests/Integration/JetStream/PublishExtendedTest.php b/tests/Integration/JetStream/PublishExtendedTest.php index 507f3cc..f40fe3a 100644 --- a/tests/Integration/JetStream/PublishExtendedTest.php +++ b/tests/Integration/JetStream/PublishExtendedTest.php @@ -43,43 +43,6 @@ protected function tearDown(): void $this->conn->close(); } - public function testPublishAsyncPending(): void - { - $prefix = strtolower($this->streamName); - - // Publish several messages asynchronously - $futures = []; - for ($i = 0; $i < 5; $i++) { - $futures[] = $this->js->publishAsync("{$prefix}.async", "msg{$i}"); - } - - // At least some should be pending before processing - $initialPending = $this->js->publishAsyncPending(); - self::assertGreaterThanOrEqual(0, $initialPending); - self::assertFalse($this->js->publishAsyncComplete() && $initialPending > 0); - - // Process the event loop to receive acks - for ($i = 0; $i < 20; $i++) { - $this->conn->process(0.1); - if ($this->js->publishAsyncComplete()) { - break; - } - } - - self::assertTrue($this->js->publishAsyncComplete()); - self::assertSame(0, $this->js->publishAsyncPending()); - - // Verify all futures resolved successfully - foreach ($futures as $future) { - self::assertTrue($future->isComplete()); - $ack = $future->ok(); - self::assertSame($this->streamName, $ack->stream); - } - - // Cleanup internal tracking - $this->js->cleanupPublisher(); - } - public function testExpectLastMsgId(): void { $prefix = strtolower($this->streamName); diff --git a/tests/Integration/JetStream/PublishTest.php b/tests/Integration/JetStream/PublishTest.php index bc8a7b8..9d6fa66 100644 --- a/tests/Integration/JetStream/PublishTest.php +++ b/tests/Integration/JetStream/PublishTest.php @@ -118,21 +118,6 @@ public function testPublishMessage(): void self::assertGreaterThanOrEqual(1, $ack->sequence); } - public function testPublishAsync(): void - { - $future = $this->js->publishAsync('test.pub.async', 'async data'); - - self::assertFalse($future->isComplete()); - - // Drive the event loop to receive the ack - $this->conn->process(1.0); - - if ($future->isComplete()) { - $ack = $future->ok(); - self::assertSame('TEST_PUB', $ack->stream); - } - } - public function testPublishEmptyPayload(): void { $ack = $this->js->publish('test.pub.empty', ''); diff --git a/tests/Unit/JetStream/Publish/PubAckFutureTest.php b/tests/Unit/JetStream/Publish/PubAckFutureTest.php deleted file mode 100644 index a4cf3d7..0000000 --- a/tests/Unit/JetStream/Publish/PubAckFutureTest.php +++ /dev/null @@ -1,34 +0,0 @@ -expectException(\LogicException::class); - $this->expectExceptionMessage('PubAckFuture is not complete'); - - $future->ok(); - } - - public function testOkReturnsResolvedAck(): void - { - $future = new PubAckFuture(new Message(subject: 'test.subject')); - $future->resolve(new PubAck(stream: 'TEST', sequence: 42)); - - $ack = $future->ok(); - - self::assertSame('TEST', $ack->stream); - self::assertSame(42, $ack->sequence); - } -}