fix: remove JetStream async publish surface

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

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();

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',

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<PubAckFuture> */
-    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

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,
     ) {