PHPackages                             idct/php-nats-jetstream-client - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [Utility & Helpers](/categories/utility)
4. /
5. idct/php-nats-jetstream-client

ActiveLibrary[Utility & Helpers](/categories/utility)

idct/php-nats-jetstream-client
==============================

Async-first NATS + JetStream client for PHP 8.2+

v2.4.1(1mo ago)68.8k1[3 issues](https://github.com/ideaconnect/php-nats-jetstream-client/issues)1BSD-3-ClausePHPPHP ^8.2CI passing

Since Mar 11Pushed 3w agoCompare

[ Source](https://github.com/ideaconnect/php-nats-jetstream-client)[ Packagist](https://packagist.org/packages/idct/php-nats-jetstream-client)[ RSS](/packages/idct-php-nats-jetstream-client/feed)WikiDiscussions main Synced 4w ago

READMEChangelog (10)Dependencies (21)Versions (19)Used By (1)

IDCT PHP NATS JetStream Client
==============================

[](#idct-php-nats-jetstream-client)

[![codecov](https://camo.githubusercontent.com/9e368fd678067b79b930b9a1ddc42690521b871b62d314f0e8b1f86034ccf17a/68747470733a2f2f636f6465636f762e696f2f67682f69646561636f6e6e6563742f7068702d6e6174732d6a657473747265616d2d636c69656e742f67726170682f62616467652e7376673f746f6b656e3d41383136663445586f6e)](https://codecov.io/gh/ideaconnect/php-nats-jetstream-client)[![CI](https://github.com/ideaconnect/php-nats-jetstream-client/actions/workflows/ci.yml/badge.svg)](https://github.com/ideaconnect/php-nats-jetstream-client/actions/workflows/ci.yml)[![Made in the EU](https://raw.githubusercontent.com/ideaconnect/made-in-the-eu/main/software-badge/made-in-the-eu.svg)](https://github.com/ideaconnect/made-in-the-eu)

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

[](#installation)

Install from Packagist:

```
composer require idct/php-nats-jetstream-client
```

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

Source repository:

Index
-----

[](#index)

- [Installation](#installation)
- [Features](#features)
- [NATS Server Version Requirements](#nats-server-version-requirements)
- [PHP Support Policy](#php-support-policy)
- [Usage](#usage)
- [Authentication Options](#authentication-options)
- [WebSocket Transport](#websocket-transport)
- [Connect and Publish/Subscribe](#connect-and-publishsubscribe)
- [Request/Reply](#requestreply)
- [Request Many (Scatter-Gather)](#request-many-scatter-gather)
- [Connection Statistics and RTT](#connection-statistics-and-rtt)
- [Headers and Server Info](#headers-and-server-info)
- [JetStream Stream and Durable Consumer](#jetstream-stream-and-durable-consumer)
- [JetStream Stream Update and Consumer Info](#jetstream-stream-update-and-consumer-info)
- [JetStream Pull Consumer (Fetch + ACK)](#jetstream-pull-consumer-fetch--ack)
- [JetStream Pull Consumer (NAK, Delayed NAK, TERM, In-Progress)](#jetstream-pull-consumer-nak-delayed-nak-term-in-progress)
- [Queue Group Subscribe](#queue-group-subscribe)
- [Polling Subscribe (SubscriptionQueue)](#polling-subscribe-subscriptionqueue)
- [JetStream Push Consumer (Durable)](#jetstream-push-consumer-durable)
- [JetStream Ephemeral Consumers](#jetstream-ephemeral-consumers)
- [Scheduled Publish Example (`@at`)](#scheduled-publish-example-at)
- [Distributed Counter](#distributed-counter)
- [KeyValue Bucket](#keyvalue-bucket)
- [Object Store Bucket](#object-store-bucket)
- [Object Store Streaming to Callback](#object-store-streaming-to-callback)
- [Object Store Streaming Upload](#object-store-streaming-upload)
- [Services Framework](#services-framework)
- [Services: SCHEMA Discovery](#services-schema-discovery)
- [Graceful Drain](#graceful-drain)
- [Ordered Consumer](#ordered-consumer)
- [Consumer Pause/Resume](#consumer-pauseresume)
- [Fetch Batch](#fetch-batch)
- [Stream Purge and List](#stream-purge-and-list)
- [Consumer List](#consumer-list)
- [Stream Message Get](#stream-message-get)
- [JetStream Direct Get](#jetstream-direct-get)
- [Atomic Batch Publish](#atomic-batch-publish)
- [Credentials File Authentication](#credentials-file-authentication)
- [Typed Stream Configuration](#typed-stream-configuration)
- [Pull Consumer Batching/Iteration](#pull-consumer-batchingiteration)
- [Pull Consumer Priority Groups](#pull-consumer-priority-groups)
- [Stream Mirroring and Sourcing](#stream-mirroring-and-sourcing)
- [Republish and Subject Transform](#republish-and-subject-transform)
- [Compatibility Mapping](#compatibility-mapping)
- [Behavior Notes](#behavior-notes)
- [Production Notes and Limitations](#production-notes-and-limitations)
- [Configuration Option Mapping](#configuration-option-mapping)
- [Performance Benchmark Recipe](#performance-benchmark-recipe)
- [Testing](#testing)
- [Contributing and contributors](#contributing-and-contributors)
- [Current Test Baseline](#current-test-baseline)
- [License](#license)

Features
--------

[](#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
--------------------------------

[](#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.

FeatureAPIMin NATSServer config / headerNamed consumer create (all consumer helpers)`createConsumer()`, `createEphemeralConsumer()`, ordered/push/pull subscribe helpers2.9n/aMulti-subject consumer filters`createConsumer(..., ['filter_subjects' => [...]])`2.10`filter_subjects`Per-message / KV TTL`publish(..., ttl:)`, `KeyValueBucket::put(..., ttl:)`, `delete/purge(..., tombstoneTtl:)`2.11`allow_msg_ttl`, `Nats-TTL`Subject delete markersKV/Object Store `watch()`/`get()` (handled automatically)2.11`subject_delete_marker_ttl`, `Nats-Marker-Reason`Pull priority groups`fetchBatch(..., $pull)`, `PullConsumerIterator::setGroup/setPriority/...`, `unpinConsumer()`2.11 (`prioritized` 2.12)`priority_groups`/`priority_policy`, `Nats-Pin-Id`Batched / multi Direct Get`directGetBatch()`, `directGetLastForSubjects()`2.11`allow_direct`Scheduled publishing (`@every`/cron/aliases)`publishScheduled()`, `Schedule::every/cron/predefined`2.12`allow_msg_schedules`, `Nats-Schedule*`Atomic batch publish`batch()` -> `BatchPublisher`2.12`allow_atomic`, `Nats-Batch-*`Distributed counter CRDT`incrementCounter()`, `counterValue()`2.12`allow_msg_counter`, `Nats-Incr`Publish de-duplication`publish(..., msgId:)`2.2`Nats-Msg-Id````
use IDCT\NATS\Exception\UnsupportedFeatureException;

try {
    $js->batch()->add('orders.created', $payload)->commit()->await();
} catch (UnsupportedFeatureException $e) {
    // e.g. "Atomic batch publish requires NATS server 2.12+ (connected server 2.10.5 treated the batch as plain publishes)"
    echo "{$e->feature} needs NATS {$e->requiredVersion}; server is {$e->serverVersion}\n";
}
```

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

PHP Support Policy
------------------

[](#php-support-policy)

This library follows [PHP's official release schedule](https://www.php.net/supported-versions.php). 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! 💖
-------------------------------------------------------------

[](#-this-project-looks-for-funding-love-my-work-support-it-)

- ☕ **Buy me a coffee**:
- 💝 **Sponsor**:

Usage
-----

[](#usage)

> Every example below also ships as a runnable, self-contained script under [`examples/`](examples/) (one file per example, verified against dockerized NATS via [`scripts/run-examples.sh`](scripts/run-examples.sh)). See [examples/README.md](examples/README.md).

### Authentication Options

[](#authentication-options)

> 📄 **Runnable examples:** [`examples/auth-token.php`](examples/auth-token.php), [`examples/auth-userpass.php`](examples/auth-userpass.php), [`examples/auth-jwt-nkey.php`](examples/auth-jwt-nkey.php), [`examples/auth-standalone-nkey.php`](examples/auth-standalone-nkey.php), [`examples/auth-tls.php`](examples/auth-tls.php)

*Verified by: [NkeySeedSignerTest](tests/Unit/NkeySeedSignerTest.php), [NatsConnectionTest::testConnectIncludesJwtSignatureFromInfoNonce](tests/Unit/NatsConnectionTest.php); [NatsClientIntegrationTest](tests/Integration/NatsClientIntegrationTest.php) (`testTokenAuthSuccessAndFailure`, `testUserPasswordAuthSuccessAndFailure`, `testJwtNonceAuthenticationFlow`, `testStandaloneNkeyAuthenticationFlow`, `testTlsHandshakeFirstConnection`); [features/auth/](features/auth/).*

```