Download the PHP package idct/php-nats-jetstream-client without Composer

IDCT PHP NATS JetStream Client

Async-first NATS and JetStream client for PHP 8.2+ with support for core NATS messaging, JetStream, KeyValue, ObjectStore, and NATS microservices.

The library is built around Amp and provides a typed, high-level API for connection management, publish/subscribe, request/reply, reconnect handling, authentication flows, and JetStream resource management without falling back to blocking I/O.

It is intended for real application use, including service-to-service messaging, event processing, JetStream-backed persistence patterns, and NATS-based microservice discovery.

Installation

Install from Packagist:

Package name: idct/php-nats-jetstream-client

Source repository: https://github.com/ideaconnect/php-nats-jetstream-client

Index

• Installation
• Features
• NATS Server Version Requirements
• PHP Support Policy
• Usage
• Authentication Options
• WebSocket Transport
• Connect and Publish/Subscribe
• Request/Reply
• Request Many (Scatter-Gather)
• Connection Statistics and RTT
• Headers and Server Info
• JetStream Stream and Durable Consumer
• JetStream Stream Update and Consumer Info
• JetStream Pull Consumer (Fetch + ACK)
• JetStream Pull Consumer (NAK, Delayed NAK, TERM, In-Progress)
• Queue Group Subscribe
• Polling Subscribe (SubscriptionQueue)
• JetStream Push Consumer (Durable)
• JetStream Ephemeral Consumers
• Scheduled Publish Example (@at)
• Distributed Counter
• KeyValue Bucket
• Object Store Bucket
• Object Store Streaming to Callback
• Object Store Streaming Upload
• Services Framework
• Services: SCHEMA Discovery
• Graceful Drain
• Ordered Consumer
• Consumer Pause/Resume
• Fetch Batch
• Stream Purge and List
• Consumer List
• Stream Message Get
• JetStream Direct Get
• Atomic Batch Publish
• Credentials File Authentication
• Typed Stream Configuration
• Pull Consumer Batching/Iteration
• Pull Consumer Priority Groups
• Stream Mirroring and Sourcing
• Republish and Subject Transform
• Compatibility Mapping
• Behavior Notes
• Production Notes and Limitations
• Configuration Option Mapping
• Performance Benchmark Recipe
• Testing
• Contributing and contributors
• Current Test Baseline
• License

Features

Current functionality includes:

• Core NATS connect/disconnect with graceful drain
• Publish and subscribe
• Request/reply with timeout and cancellation
• Reconnect with exponential backoff, server rotation, validated subscription replay, and async INFO updates
• Ping/pong heartbeat with maxPingsOut detection
• max_payload enforcement and no_responders negotiation
• Subject validation against NATS naming rules
• JetStream account info
• JetStream stream CRUD (create, update, get, delete, purge, list)
• JetStream consumer CRUD (durable + ephemeral, pull + push, list)
• JetStream pull consumers (fetch next, fetch batch, ACK/NAK/TERM/WPI, delayed NAK)
• JetStream push consumers with heartbeat/flow-control handling
• JetStream ordered consumers with automatic sequence tracking and gap recovery
• JetStream consumer pause/resume
• JetStream publish ACK
• JetStream stream message get by sequence - both the regular STREAM.MSG.GET request and the Direct Get API (directGetStreamMessage() / directGetLastMessageForSubject())
• JetStream atomic (all-or-nothing) batch publish (batch() -> BatchPublisher, ADR-50; requires allow_atomic, NATS 2.12+)
• Scheduled publish (@at support)
• Distributed counter CRDT (atomic incrementCounter() / counterValue(), arbitrary-precision values)
• KeyValue API (bucket lifecycle with history/TTL/storage options, put/get/update/delete/purge, watch, getAll/status)
• ObjectStore API (bucket lifecycle, put/get/delete/list/watch, chunked uploads, streaming upload via putStream(), SHA-256 digest verification)
• Connection flush() (PING/PONG round-trip) to confirm the server has processed prior writes
• Request-many scatter-gather (requestMany(): collect multiple replies bounded by max-count / stall / timeout)
• Connection traffic statistics (statistics() -> ConnectionStats) and round-trip-time measurement (rtt())
• Microservices framework (service registration, PING/INFO/STATS/SCHEMA discovery, grouped endpoints)
• Server authorization methods: token, username/password, JWT + nonce signer, built-in NKey seed signer, credentials file parser
• Standalone NKey authentication (Ed25519 challenge signing without JWT)
• no_echo CONNECT option
• tlsHandshakeFirst TLS option
• Typed JetStream configuration enums (RetentionPolicy, StorageBackend, DiscardPolicy, DeliverPolicy, AckPolicy, ReplayPolicy)
• Max frame size limit in protocol parser (DoS protection)
• Queue-based polling subscribe API (SubscriptionQueue with fetch(), next(), fetchAll())
• Pull-consumer batching/iteration chain API (PullConsumerIterator with setBatching(), setIterations(), handle())
• Stream mirroring and sourcing configuration helpers (StreamSource)
• Republish and subject transform configuration helpers (Republish, SubjectTransform)

Scheduling note: scheduled messages use the NATS scheduler headers (ADR-51) and accept @at, @every, 6-field cron, and the predefined aliases (@daily, @hourly, ...). Build expressions with IDCT\NATS\JetStream\Schedule::at(...), Schedule::atTimestamp(...), Schedule::every(...), Schedule::cron(...), or Schedule::predefined(...). The target stream must be created with allow_msg_schedules (NATS 2.12+).

NATS Server Version Requirements

Core NATS (publish/subscribe, request/reply, headers, services) works against any server. JetStream consumer management requires NATS 2.9+: all consumer helpers use the named CONSUMER.CREATE API introduced in 2.9, with no fallback to the legacy DURABLE.CREATE form. Some features depend on even newer NATS server versions; the table below lists the minimum version per feature. Detection is reactive - there is no per-request version probe - and depends on the feature:

• Version-gated stream/consumer config fields (for example creating a stream with allow_atomic or allow_msg_schedules): an older server rejects the unknown field and the request fails fast with an IDCT\NATS\Exception\UnsupportedFeatureException (a subclass of JetStreamException) carrying the feature name, the required version, and the version the server reported.
• Atomic batch publish: a server without batch support acknowledges the batch start/commit as plain publishes; commit() detects that and throws an UnsupportedFeatureException instead of silently storing the batch message-by-message.
• Other data-path calls (for example publish(..., ttl:) or directGetBatch()) against a server or stream without the feature surface a plain JetStreamException built from the server's error response - or the server may silently ignore an unknown header - so consult the table before relying on them against older servers.

You can also query the requirement programmatically: IDCT\NATS\JetStream\FeatureSupport::requiredVersion('allow_atomic') returns "2.12".

PHP Support Policy

This library follows PHP's official release schedule. The minimum required PHP version tracks the releases that still receive official support (active or security), and a version is dropped once it reaches end-of-life - we neither require a PHP version before it is broadly available nor keep supporting one after upstream stops patching it.

• Current minimum: PHP 8.2. CI runs the full test suite on every still-supported minor - currently 8.2, 8.3, 8.4, and 8.5.
• PHP 8.2 reaches end-of-life on 2026-12-31. Support for PHP 8.2 will therefore be dropped by the end of 2026, after which the minimum becomes PHP 8.3. Applications that must stay on PHP 8.2 should pin to the last release made before that change.
• Mutation testing (Infection) requires PHP 8.3+ and so runs only on 8.3+ in CI, but it is a development-only tool - it does not affect the runtime requirement, and the library installs and runs on PHP 8.2.

🚀 This project looks for funding. Love my work? Support it! 💖

• ☕ Buy me a coffee: https://buymeacoffee.com/idct

• 💝 Sponsor: https://github.com/sponsors/ideaconnect

Usage

Every example below also ships as a runnable, self-contained script under examples/README.md.

Authentication Options

📄 Runnable examples: examples/auth-tls.php

Verified by: features/auth/.

NkeySeedSigner derives the public NKey from an encoded seed and emits the base64url Ed25519 nonce signature expected by NATS servers.

NkeySeedSigner requires the PHP sodium extension because NATS NKey authentication uses Ed25519 challenge signing.

WebSocket Transport

📄 Runnable example: examples/websocket-transport.php

By default NatsClient uses the TCP transport (AmpSocketTransport). To connect over WebSocket - e.g. through a NATS gateway that only exposes ws:// / wss:// - construct a WebSocketTransport and inject it. The ws:// / wss:// endpoint goes in servers; wss:// negotiates TLS during connect (using the same tls* options), and the optional webSocketHeaders / webSocketCompression options apply to the upgrade handshake.

ws:// / wss:// URLs are only handled by WebSocketTransport; passing such a URL to the default TCP transport will not work. wss:// requires ext-openssl, and webSocketCompression requires ext-zlib.

Verified by: ClientParityIntegrationTest (testWebSocketTransportCarriesPubSubAndJetStream, testWebSocketCompressionAndCustomHeaders).

Connect and Publish/Subscribe

📄 Runnable example: examples/publish-subscribe.php

Verified by: features/core/connection.feature.

Request/Reply

📄 Runnable example: examples/request-reply.php

_Verified by: features/core/request_reply.feature._

Request Many (Scatter-Gather)

📄 Runnable example: examples/request-many.php

Verified by: NatsConnectionTest (testRequestManyCollectsUpToMaxResponses, testRequestManyStopsOnStallInterval, testRequestManyReturnsEmptyOnNoResponders).

requestMany() sends a single request and collects MULTIPLE replies (scatter-gather), returning a list<NatsMessage>. Collection stops on the first of: $maxResponses replies, a no_responders sentinel (returns []), the per-message $stallMs gap, or $totalTimeoutMs.

Connection Statistics and RTT

📄 Runnable example: examples/connection-stats-rtt.php

Verified by: NatsConnectionTest (testConnectionAccessorsAndStatistics, testRttMeasuresPingPong).

statistics() returns an immutable ConnectionStats snapshot of traffic counters (inMsgs, outMsgs, inBytes, outBytes, reconnects). rtt() measures the round-trip time to the server with a PING/PONG exchange and resolves to a float in seconds.

Headers and Server Info

📄 Runnable example: examples/headers-and-server-info.php

_Verified by: features/core/headers_queueing.feature._

JetStream Stream and Durable Consumer

📄 Runnable example: examples/jetstream-stream-and-consumer.php

_Verified by: features/jetstream-core/stream_lifecycle.feature._

JetStream Stream Update and Consumer Info

📄 Runnable example: examples/stream-update-and-consumer-info.php

Verified by: features/jetstream-core/management.feature.

Idempotent upserts are available when you do not want to branch on "exists vs. not": createOrUpdateStream() creates the stream or falls back to updating it when the name is already in use, and addOrUpdateConsumer() creates or updates a durable consumer. Both mirror nats.go / nats.java CreateOrUpdateStream / CreateOrUpdateConsumer.

Verified by: JetStreamContextTest (testCreateOrUpdateStreamFallsBackToUpdate, testAddOrUpdateConsumerDelegatesToCreateConsumer).

JetStream Pull Consumer (Fetch + ACK)

📄 Runnable example: examples/pull-consumer-fetch-ack.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

When a pull request ends with a terminal JetStream status frame and no user message is delivered, fetchNext() / fetchBatch() raise JetStreamException with the server status code and description, for example JetStream pull request ended with status 404: No Messages.

For exactly-once-style processing use ackSync() instead of ack(): it sends the +ACK as a request and waits for the server to confirm the acknowledgement was durably recorded (double-ack). It throws if the delivered message carries no reply subject, and throws TimeoutException if no confirmation arrives within the optional timeout.

Verified by: JetStreamContextTest (testAckSyncSendsAckAsRequestAndAwaitsConfirmation, testAckSyncThrowsForEmptyReplySubject).

JetStream Pull Consumer (NAK, Delayed NAK, TERM, In-Progress)

📄 Runnable example: examples/pull-consumer-nak-term.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

Queue Group Subscribe

📄 Runnable example: examples/queue-group-subscribe.php

_Verified by: features/core/headers_queueing.feature._

Polling Subscribe (SubscriptionQueue)

📄 Runnable example: examples/polling-subscribe.php

_Verified by: features/core/headers_queueing.feature._

JetStream Push Consumer (Durable)

📄 Runnable example: examples/push-consumer-durable.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

JetStream Ephemeral Consumers

📄 Runnable example: examples/ephemeral-consumers.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

Scheduled Publish Example (@at)

📄 Runnable example: examples/scheduled-publish.php

_Verified by: features/jetstream-data/scheduled_publish.feature._

Prerequisites: the backing stream must be created with allow_msg_schedules: true, and because this example sets scheduleTtl, also allow_msg_ttl: true. The stream's subject list must cover both the schedule subject and the target subject. Without these flags the server rejects the publish with message schedules is disabled or per-message TTL is disabled.

Distributed Counter

📄 Runnable example: examples/distributed-counter.php

Verified by: JetStreamContextTest (testIncrementCounter, testIncrementCounterPreservesBigValue, testIncrementCounterRejectsMalformedDelta, testCounterValue, testCounterValueMissingReturnsZero, testCounterValueRethrowsNon404Exception, testIncrementCounterWithMalformedResponsePayload, testIncrementCounterWithApiErrorInResponse, testIncrementCounterWithIntegerValField, testIncrementCounterWithMissingValFieldThrows).

Distributed counters are an atomic, conflict-free (CRDT) increment subject backed by a JetStream stream. incrementCounter() applies a signed delta and returns the new total; counterValue() reads the current total via Direct Get (returning "0" when nothing is stored yet).

Requires NATS server 2.12+, and the backing stream must be created with allow_msg_counter: true (and allow_direct: true, since counterValue() reads via Direct Get). On a pre-2.12 server, createStream() with the counter flag fails fast with an UnsupportedFeatureException (the server rejects the unknown config field). A stream created without the flag causes the increment request to be rejected, surfaced as a JetStreamException.

The delta is passed as an integer string (e.g. "+5", "-3", "10"); a malformed delta is rejected before dispatch. Counter totals are likewise returned as strings so values beyond PHP_INT_MAX are preserved exactly rather than truncated to a float.

KeyValue Bucket

📄 Runnable example: examples/keyvalue-bucket.php

_Verified by: features/jetstream-data/key_value.feature._

To read a specific historical revision (stream sequence) of a key, use getRevision(). It returns null when nothing is stored at that sequence or when the sequence belongs to a different key, and throws for a non-positive revision.

Verified by: KeyValueBucketTest (testGetRevisionReturnsEntryAtSequence, testGetRevisionReturnsNullForDifferentKey).

Object Store Bucket

📄 Runnable example: examples/object-store-bucket.php

_Verified by: features/jetstream-data/object_store.feature._

Buckets and objects support extra management operations. seal() makes a bucket permanently read-only (irreversible). addLink() / addBucketLink() create link objects pointing at another object or a whole bucket. updateMeta() renames an object and/or replaces its metadata WITHOUT re-uploading its bytes (the stored chunks are kept by NUID).

Verified by: ObjectStoreBucketTest (testSeal, testAddLink, testAddBucketLink, testUpdateMetaRenamesPreservingNuid, testUpdateMetaReplacesMetadataInPlace).

Object Store Streaming to Callback

📄 Runnable example: examples/object-store-streaming-to-callback.php

_Verified by: features/jetstream-data/object_store.feature._

Object Store Streaming Upload

📄 Runnable example: examples/object-store-streaming-upload.php

Verified by: JetStreamIntegrationTest (testJetStreamObjectStorePutStreamRoundTrip).

Services Framework

📄 Runnable example: examples/services-framework.php

Verified by: features/services/.

Services expose runtime helpers: statsSnapshot() returns the current io.nats.micro.v1.stats_response array (per-endpoint num_requests / num_errors / last_error / processing_time / average_processing_time), reset() zeroes those counters, and withRequestValidator() enables opt-in per-request validation for endpoints declared with a schema. The validator receives the message and the endpoint schema and returns null to accept or a string rejection reason (which becomes a VALIDATION_ERROR error reply).

Verified by: ServiceTest (testStatsIncludeDetailedMetrics, testResetClearsStats, testRequestValidatorCanRejectRequests).

Services: SCHEMA Discovery

📄 Runnable example: examples/services-schema-discovery.php

_Verified by: features/services/service_discovery.feature._

Graceful Drain

📄 Runnable example: examples/graceful-drain.php

_Verified by: features/resilience/client_resilience.feature._

Note that disconnect() and a plain unsubscribe($sid) discard any locally queued, undelivered messages (nats.go Close()/Unsubscribe() parity) - messages already received from the server but not yet dispatched to a handler are dropped without warning. Use drain() (whole connection) or drainSubscription($sid) (single subscription) as the lossless teardown paths: both deliver the buffered backlog first.

Ordered Consumer

📄 Runnable example: examples/ordered-consumer.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

Consumer Pause/Resume

📄 Runnable example: examples/consumer-pause-resume.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

Fetch Batch

📄 Runnable example: examples/fetch-batch.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

Notes:

1. A partial batch is valid. If the server delivers some messages and then ends the pull with a terminal status, the delivered messages are returned.
2. A terminal status only becomes an exception when no user message was delivered for that pull request.

Stream Purge and List

📄 Runnable example: examples/stream-purge-and-list.php

Verified by: features/jetstream-core/management.feature.

Consumer List

📄 Runnable example: examples/consumer-list.php

Verified by: features/jetstream-core/management.feature.

Stream Message Get

📄 Runnable example: examples/stream-message-get.php

Verified by: JetStreamIntegrationTest (testJetStreamGetStreamMessage, testJetStreamGetStreamMessagePreservesZeroAndHeaders).

getStreamMessage() fetches a stored message by sequence using the standard JetStream $JS.API.STREAM.MSG.GET API. The returned NatsMessage preserves the stored subject, payload (including a body that is exactly "0"), and any stored headers on rawHeaders.

Stored messages can be removed by sequence with deleteMessage(). By default this is a fast delete (no_erase: the sequence is unlinked but the bytes are left in place). Pass secureErase: true for a secure delete that overwrites the message data before removal (slower; mirrors nats.go SecureDeleteMsg).

Verified by: JetStreamContextTest::testDeleteMessageFastAndSecure.

JetStream Direct Get

📄 Runnable example: examples/jetstream-direct-get.php

directGetStreamMessage() and directGetLastMessageForSubject() use the JetStream Direct Get API ($JS.API.DIRECT.GET), which requires the stream to be created with allow_direct: true. Unlike getStreamMessage() (regular $JS.API.STREAM.MSG.GET, served by the stream leader), Direct Get can be answered by any replica. The server returns the stored message directly: the payload is the raw body and the stream/sequence/subject/timestamp travel as Nats-* headers (preserved on rawHeaders). A miss is raised as a JetStreamException (for example 404 Message Not Found).

Verified by: JetStreamIntegrationTest::testJetStreamDirectGetStreamMessage.

Batched / multi Direct Get

For multi-message reads the client adds two batched Direct Get helpers (ADR-31), which require NATS server 2.11+ in addition to the stream's allow_direct. Each issues a single request whose replies stream to a private inbox, terminated by a 204 end-of-batch marker (or a final message carrying Nats-Num-Pending: 0); the wait is bounded by $expiresMs (default 5000) so a silent server cannot hang the call. directGetBatch() returns a list<NatsMessage>; an error status (e.g. 408) is raised as a JetStreamException.

• directGetLastForSubjects(string $stream, array $subjects, int $expiresMs = 5000) fetches the latest message for each named subject in one round trip (multi_last). It expects exact subjects: a subject containing * or > is rejected with a JetStreamException. An empty $subjects array returns [] without contacting the server.
• directGetBatch(string $stream, array $body, int $expiresMs = 5000) issues a raw batched request; $body accepts keys such as batch, seq, up_to_seq and multi_last. $expiresMs must be greater than zero.

Verified by: JetStreamIntegrationTest::testJetStreamBatchedDirectGet.

Atomic Batch Publish

📄 Runnable example: examples/atomic-batch-publish.php

Verified by: JetStreamIntegrationTest::testJetStreamAtomicBatchPublish.

$js->batch() opens an atomic (all-or-nothing) JetStream publish batch (ADR-50). Messages are staged with the fluent add($subject, $payload, $headers = []) and sent together on commit(), which returns a Future<PubAck>. Every message carries a shared Nats-Batch-Id and an incrementing Nats-Batch-Sequence; the final message carries Nats-Batch-Commit: 1, on which the server atomically commits the whole batch and replies with a single PubAck whose batchCount is the committed count and whose batchId echoes the batch id. If any consistency check fails the entire batch is aborted and nothing is stored - the failure surfaces as a JetStreamException.

This is a NATS 2.12+ feature: the target stream must be created with allow_atomic enabled. On a pre-2.12 server createStream(..., ['allow_atomic' => true]) fails fast with an UnsupportedFeatureException (see NATS Server Version Requirements); if the stream exists without allow_atomic, commit() surfaces a JetStreamException (e.g. "atomic publish not enabled"). Pass your own batch id (1..64 characters) to batch(), or omit it to have one generated. A single batch is capped at BatchPublisher::MAX_MESSAGES (1000) messages.

Credentials File Authentication

📄 Runnable example: examples/auth-credentials-file.php

_Verified by: features/auth/jwt_and_nkey_auth.feature._

Typed Stream Configuration

📄 Runnable example: examples/typed-stream-configuration.php

Verified by: the JetStreamContextTest.

Stream and consumer configuration supports typed enums for type-safe options:

Pull Consumer Batching/Iteration

📄 Runnable example: examples/pull-consumer-batching-iteration.php

_Verified by: features/jetstream-core/consumer_helpers.feature._

The fluent PullConsumerIterator wraps fetchBatch() with configurable batch size, iterations, and a handler callback:

Pull Consumer Priority Groups

📄 Runnable example: examples/pull-consumer-priority-groups.php

Verified by: JetStreamContextTest::testPinIdOf.

ADR-42 priority groups let several pull clients share one consumer while the server steers delivery. Create the consumer with priority_groups (a non-empty array of 1..16-character group names) and a priority_policy - one of overflow, pinned_client, or prioritized. The priority-group fields require NATS server 2.11+; the prioritized policy requires 2.12+ (an older server rejects them). See NATS Server Version Requirements.

Then pull under a group with the PullConsumerIterator setters: setGroup() (required for any priority policy), and as the policy needs them setPriority() (0-9, for prioritized), setMinPending() / setMinAckPending() (for overflow), plus the general setMaxBytes() and setNoWait(). Under the pinned_client policy the server pins one client at a time: the iterator automatically captures the Nats-Pin-Id from the first delivered message and resends it on subsequent pulls, and transparently re-pins if the pin goes stale (the server returns a 423 status). Call JetStreamContext::pinIdOf($msg) to read a message's pin id, and JetStreamContext::unpinConsumer($stream, $consumer, $group) to release the active pin so another client can take over.

Stream Mirroring and Sourcing

📄 Runnable example: examples/stream-mirroring-and-sourcing.php

_Verified by: features/jetstream-core/config_helpers.feature (live source filtering + mirror replication)._

Use StreamSource to build mirror/source configuration arrays:

Use those arrays in createStream() or updateStream() options. Source configurations work with the current high-level API and are covered against the live fixture stack. Mirror-only stream configs also work through createStream() when you pass an empty subjects list together with the mirror configuration.

Republish and Subject Transform

📄 Runnable example: examples/republish-and-subject-transform.php

_Verified by: features/jetstream-core/config_helpers.feature (live republish + subject transform)._

Configure republish rules and subject transforms on streams:

Compatibility Mapping

This repository tracks parity against the basis-company nats.php README examples while exposing an Amp-first API.

Behavior Notes

processIncoming()

Verified by: NatsConnectionTest (testProcessIncomingDispatchesMsgToSubscriber, testProcessIncomingUpdatesServerInfoFromAsyncInfoFrame).

processIncoming() reads a single transport chunk, parses all complete frames from it, and dispatches them to subscription callbacks. It is non-blocking - if no data is available, it returns immediately with a frame count of 0. Because one read returns only a single chunk (and TCP may coalesce several protocol messages into one chunk), call it in a loop to process all available messages:

The client also applies asynchronous INFO updates received after connect, so serverInfo() can change during the lifetime of an open connection when the server advertises updated capabilities such as max_payload or cluster topology details.

Heartbeat and Request Timeouts

Verified by: NatsClientIntegrationTest (testIdleConnectionStaysOpenViaHeartbeat, testRequestTimeoutDoesNotPoisonConnection, testMaxPingsOutTriggersReconnect).

The heartbeat timer answers its own PONG: after sending a PING it performs a short, bounded read to consume the reply (unless an application processIncoming() read is already in flight). Liveness detection therefore does not depend on the application continuously calling processIncoming(), so an otherwise idle connection (for example a pure publisher) is not closed by spurious maxPingsOut detection. Only an actual server PONG resets the outstanding-ping counter - other inbound frames (data, INFO, PING) do not - so a server that keeps sending data but stops answering PINGs is still detected via maxPingsOut.

Request and pull-fetch timeouts cancel the underlying socket read rather than leaving it pending. A timed-out request() (or fetchBatch()/fetchNext()) cleanly releases the read, so it cannot orphan an in-flight read or trigger a spurious reconnect on the next operation.

Reconnect Behavior

Verified by: NatsClientIntegrationTest (testReconnectAfterTransportLossReplaysSubscriptions, testReconnectBackoffDelayProgression, testReconnectAttemptsExhaustedReturnsClosed).

When a connection drops and reconnectEnabled is true:

1. Exponential backoff: delay is computed as reconnectDelayMs * 2^(attempt - 1), capped at reconnectMaxDelayMs, with random jitter up to reconnectJitterMs.
2. Server rotation: the client cycles through configured servers in order.
3. Subscription replay: all active subscriptions are replayed (SUB commands resent) after reconnect.
4. Replay validation: reconnect does not treat replayed subscriptions as successful if the server immediately answers with a fatal -ERR during replay. In that case reconnect keeps retrying until a healthy server accepts the replay or attempts are exhausted.
5. Published messages during reconnect are buffered and replayed: while a reconnect is in flight, publishes are held in an outbound buffer (up to reconnectBufferSize, default 8 MiB, matching nats.go) and flushed in order on a successful reconnect. A publish is rejected (throws) only when the buffer is full, when reconnectBufferSize is 0 (buffering disabled), or when the connection is closed / not reconnecting. Verified by: NatsConnectionTest (testPublishBuffersDuringReconnectAndFlushesOnReconnect, testPublishWithHeadersBuffersDuringReconnectAndRecordsOutbound, testPublishBufferOverflowThrowsDuringReconnect).

Recoverable server -ERR frames such as Invalid Subject or Permissions Violation for Publish/Subscription to ... do not automatically close an already-open connection. Fatal connection-level errors still do.

The initial handshake is bounded by connectTimeoutMs, not by a fixed number of transport reads. During bootstrap the client will also answer server PING frames and process async INFO updates while waiting for the initial PONG.

Ordered Consumer Gap Recovery

Verified by: JetStreamIntegrationTest::testJetStreamOrderedConsumerWithFilteredSubjectAfterPriorMessages.

subscribeOrderedConsumer() tracks the JetStream consumer delivery sequence (which is contiguous per delivery, even for a filtered consumer over a stream that also carries non-matching subjects). If a push is missed - the consumer sequence skips - the consumer is transparently deleted and recreated starting just after the last in-order message; the out-of-order message that exposed the gap is discarded (not forwarded), and the recreated consumer replays the missing range in order. Delivery to your callback therefore stays in order and gap-free, with no duplicates and no recreate storm. If the restart point has been pruned/expired, recovery resumes from the next available message. If the recreate itself fails (for example the stream was deleted or a leadership change is in progress), the error is contained to this ordered consumer rather than disrupting delivery to other subscriptions on the connection.

A separate idle-heartbeat watchdog covers the case the sequence-gap logic cannot: because the gap check only runs when a frame arrives, a consumer that is reaped (for example after an inactive_threshold lapse or a server-side ordered-consumer restart) and then stops delivering entirely would otherwise leave a live subscription to a deliver inbox nothing will ever publish to again - no data, no heartbeat, no error, forever. The ordered consumer requests a periodic idle_heartbeat, and if no frame (data, heartbeat, or flow-control) arrives for two intervals the watchdog transparently recreates it from the last in-order point - the same recovery the gap path uses - firing at most once per silence episode. Tune the interval with the optional idleHeartbeatNs argument to subscribeOrderedConsumer(). A caller-owned push consumer created with idle_heartbeat is watched too, but because the library cannot recreate a consumer it does not own, a stall there surfaces a descriptive "not active" error through the errorListener instead. The same watchdog protects KV watchers, which now request a default idle_heartbeat (tunable via KeyWatchOptions).

Production Notes and Limitations

• Runtime requirements. PHP 8.2+ on the async runtime amphp/amp ^3.1 and amphp/socket ^2.3 (which requires ext-openssl for TLS). ext-sodium is additionally required for NKey / JWT authentication, and ext-zlib for WebSocket permessage-deflate compression; both are optional otherwise (declared under suggest). The library is async-first; it does not require Swoole/ReactPHP or ext-sockets.
• Concurrency model. Message delivery, request replies, and JetStream pull/push consumption are driven by reads. An application must run a processIncoming() loop (directly, or indirectly via helpers such as request(), flush(), SubscriptionQueue::next(), or the consumer iterators, which pump it for you) for callbacks to fire. An idle, publisher-only connection stays alive on its own because the heartbeat timer self-reads PONGs - see Heartbeat and Request Timeouts.
• One connection per fiber/process boundary. A NatsConnection serializes its writes and owns a single socket read; share a connection within a coroutine tree, not across independent concurrent readers.
• Interoperability. KeyValue and Object Store buckets use the official NATS layouts (KV_/OBJ_ streams, base64url object-name encoding, SHA-256=-prefixed base64url digests), so buckets written by this client are readable by the nats CLI and other official clients, and vice-versa.
• Observability. Pass a PSR-3 LoggerInterface via new NatsOptions(logger: $logger) to capture lifecycle events (connect, disconnect, reconnect, close, server discovery, lame-duck), per-attempt reconnect/backoff, and async errors. It defaults to a NullLogger. For structured, programmatic hooks (metrics, alerting, circuit breakers) without parsing log strings, pass typed closures instead: connectionListener: Closure(ConnectionEvent $event, ?Throwable $error): void is invoked on every connection-lifecycle transition, and errorListener: Closure(Throwable $error): void on async errors. Exceptions thrown by a listener are swallowed so a faulty hook cannot disrupt the connection. Verified by: NatsConnectionTest::testLoggerCapturesLifecycleEvents.
• Server version requirements. Newer features (per-message TTL, atomic batch publish, scheduled publish, priority groups, counters, batched Direct Get) require recent NATS servers - see NATS Server Version Requirements.
• Not yet implemented. A dedicated high-throughput fast-ingest batch publisher (#12) is tracked but blocked on an upstream reference; standard JetStream publish with in-flight pipelining is available today and is sufficient for most workloads.

Configuration Option Mapping

NatsOptions fields and defaults (every default below is asserted by NatsOptionsTest::testDefaultsMatchDocumentedValues):

Performance Benchmark Recipe

Quick local publish/request benchmark (single process).

The responder is pumped by a single long-lived background loop rather than one processIncoming() call per request. processIncoming() consumes one transport chunk, and TCP can coalesce several protocol messages into one chunk, so a one-call-per-request responder desynchronizes and stalls. A continuous read loop (cancelled when the run finishes) avoids that:

A ready-to-run version of this recipe ships at scripts/benchmark.php (it also measures fire-and-forget publish throughput):

Sample baseline

Single process, single connection, loopback. Numbers are environment-specific (CPU, loopback vs network, server config) and are a relative baseline, not a guarantee.

Measured with PHP 8.5, nats-server 2.12.9, 5,000 iterations, 16-byte payloads, on a WSL2 loopback. Request/reply is latency-bound (each request awaits its reply on one connection); both publish and request loops await every call, so these reflect serialized per-call throughput rather than peak pipelined rates.

Testing

Typical local workflow:

Additional useful commands:

composer infection runs Infection mutation testing against the unit testsuite (no Docker; needs a coverage driver - Xdebug or PCOV). It is a quality gate on top of line coverage: it makes small changes ("mutants") to src/ and fails if the tests don't catch them. The suite scores ~93% Covered MSI and CI enforces a strict 90% floor via the dedicated mutation job. The strictness is overridable for local exploration, e.g. INFECTION_MIN_MSI=0 composer infection -- --show-mutations, and INFECTION_DIFF_BASE=origin/main composer infection mutates only your changed lines.

Infection requires PHP 8.3+ and is intentionally not in require-dev (so the library stays installable on PHP 8.2). The CI mutation job installs it on the fly; to run mutation testing locally, add it first: composer require --dev infection/infection:^0.33.

composer test:e2e is the preferred compose-backed validation path. It checks the committed JWT fixtures, starts the local NATS stack, waits for readiness, runs unit tests, runs integration tests, runs the Behat feature suite, and tears the stack down again.

composer test:bdd runs only the Behat feature suite against the same local Docker Compose fixtures. Use BEHAT_SUITE=core composer test:bdd to run a narrower slice while iterating locally, or BEHAT_SUITE=core composer test:e2e to keep the rest of the e2e flow and narrow only the Behat stage.

Base integration endpoint:

• NATS_URL (default: nats://127.0.0.1:14222)

When you run docker compose up -d in this repository, additional local auth fixtures are available by default:

• token auth: nats://127.0.0.1:14223 with token local-test-token
• username/password auth: nats://127.0.0.1:14224 with local-user / local-pass
• TLS handshake-first auth: tls://127.0.0.1:14225 using the generated files under build/tls/
• JWT auth: nats://127.0.0.1:14227 using the generated operator/account resolver chain under build/nats/jwt/
• standalone NKey auth: nats://127.0.0.1:14226 with seed SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY

The integration tests use those defaults automatically. Override them with environment variables when you want to target external infrastructure instead.

To regenerate the committed local JWT fixture artifacts and resolver config intentionally, run:

If the local nats-jwt compose service is already running, the regeneration script recreates it so the server picks up the new operator/account resolver state immediately.

To verify the committed JWT fixture is in sync with the regeneration script, run:

• NATS_TLS_URL: TLS-enabled server URL used by testTlsHandshakeFirstConnection
• NATS_TLS_CA_FILE: optional CA bundle path for TLS verification
• NATS_TLS_CERT_FILE: optional client certificate path for TLS/mTLS
• NATS_TLS_KEY_FILE: optional client private key path for TLS/mTLS
• NATS_TLS_SKIP_VERIFY: set to 1 to disable peer verification in the TLS integration test
• NATS_TOKEN_URL: token-auth server URL used by testTokenAuthSuccessAndFailure
• NATS_TOKEN: valid token for the token-auth endpoint
• NATS_TOKEN_INVALID: invalid token used for the negative token-auth path
• NATS_USERPASS_URL: username/password-auth server URL used by testUserPasswordAuthSuccessAndFailure
• NATS_USERNAME: valid username for the user/password endpoint
• NATS_PASSWORD: valid password for the user/password endpoint
• NATS_BAD_PASSWORD: invalid password used for the negative user/password path
• NATS_JWT_URL: JWT-auth server URL used by testJwtNonceAuthenticationFlow (default: nats://127.0.0.1:14227)
• NATS_JWT: user JWT presented in the CONNECT payload (defaults to build/nats/jwt/user.jwt)
• NATS_JWT_NKEY_SEED: encoded user seed used by NkeySeedSigner to derive the public NKey and sign the server nonce (defaults to build/nats/jwt/user.seed)
• NATS_NKEY_URL: standalone NKey-auth server URL used by testStandaloneNkeyAuthenticationFlow
• NATS_NKEY_SEED: encoded user seed used by NkeySeedSigner for standalone NKey challenge-response auth

Example overrides for external infrastructure:

Focused auth/TLS integration run:

To do a quick local flake check against the compose-backed environment, run:

The CI workflow also exposes a manual workflow_dispatch soak job named integration-soak. When triggered from GitHub Actions, it runs scripts/repeat-integration.sh with a configurable repeat count on PHP 8.5.

Contributing and contributors

Contributions should keep changes focused and paired with the narrowest useful verification.

• Add or update tests when behavior changes.
• Prefer composer test:unit for small local changes and composer test:e2e for auth, transport, protocol, JetStream, or integration-fixture changes.
• Do not hand-edit generated JWT fixture files under build/nats/jwt/; regenerate them with composer fixture:jwt.
• Run composer stan for code changes and composer fix when style adjustments are needed.
• Review AGENTS.md for repository structure, standards, and continuation guidance before larger changes.

Many thanks for all the contributions:

• @amirdzhanyane-bit Protocol messages fragmentation issue

Current Test Baseline

• Unit tests cover protocol encoding/parsing, handshake/state transitions, subscriptions, backpressure policies, request/reply flows, reconnect/server-rotation behavior, and exponential backoff.
• Unit tests also cover JetStream account info, stream and consumer CRUD, publish acknowledgments, API error mapping, fetch batch, ordered consumers, consumer pause/resume, KV bucket options, ObjectStore chunking and digest verification.
• Unit tests cover microservices framework including PING/INFO/STATS/SCHEMA discovery and grouped endpoint hierarchy.
• Integration tests cover live connect/disconnect, publish-subscribe roundtrip, request-reply, connection rotation fallback, JetStream stream/consumer lifecycle with publish-ack flow, KV operations, ObjectStore operations, and service discovery.
• Integration tests also cover local token auth, username/password auth, TLS handshake-first auth including strict peer-validation, hostname mismatch, and missing-client-cert failures, resolver-backed JWT auth, and standalone NKey auth.
• Static analysis runs with PHPStan level 8.
• Combined unit + integration line coverage is ~98%, enforced at a 97% floor in CI (the build fails below it).
• Every test is catalogued with a one-line description in TESTS.md.

License

This project is licensed under the BSD 3-Clause License - see the LICENSE file for the full text. Copyright © IDCT - Bartosz Pachołek.