PHPackages                             recisio/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. [HTTP & Networking](/categories/http)
4. /
5. recisio/php-nats-jetstream-client

ActiveLibrary[HTTP & Networking](/categories/http)

recisio/php-nats-jetstream-client
=================================

Async-first NATS + JetStream client for PHP 8.1+

v2.4.0(1mo ago)08781BSD-3-ClausePHPPHP ^8.1CI passing

Since May 12Pushed 1mo agoCompare

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

READMEChangelogDependencies (14)Versions (4)Used By (1)

Recisio PHP NATS JetStream Client
=================================

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

This repository is a `recisio`-maintained fork intended to keep the client usable on PHP 8.1.

[![codecov](https://camo.githubusercontent.com/4e355594be4fac40e5d244f96f7811106c60c76ee75a13e65431ea5cae0d76e9/68747470733a2f2f636f6465636f762e696f2f67682f7265636973696f2f7068702d6e6174732d6a657473747265616d2d636c69656e742f67726170682f62616467652e7376673f746f6b656e3d41383136663445586f6e)](https://codecov.io/gh/recisio/php-nats-jetstream-client)[![CI](https://github.com/recisio/php-nats-jetstream-client/actions/workflows/ci.yml/badge.svg)](https://github.com/recisio/php-nats-jetstream-client/actions/workflows/ci.yml)

Async-first NATS and JetStream client for PHP 8.1+ with first-class 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 recisio/php-nats-jetstream-client
```

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

Source repository:

Index
-----

[](#index)

- [Installation](#installation)
- [Features](#features)
- [NATS Server Version Requirements](#nats-server-version-requirements)
- [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)

Most of this library works against any JetStream-enabled server. Some features depend on newer NATS server versions; the table below lists the minimum version per feature. **You do not need to check the version yourself** — if you use a feature against a server that is too old, 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. The check is reactive (derived from the server's own error response), so there is no per-request version probe.

FeatureAPIMin NATSServer config / headerMulti-subject consumer filters`createConsumer(..., ['filter_subjects' => [...]])`2.10`filter_subjects`Per-message / KV TTL`publish(..., ttl:)`, `KeyValueBucket::put/delete/purge(..., ttl:)`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. "The "allow_atomic" feature requires NATS server 2.12+, but the connected server reports 2.10.5."
    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"`.

🚀 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**:
- 🪙 **BTC**: bc1qntms755swm3nplsjpllvx92u8wdzrvs474a0hr
- 💎 **ETH**: 0x08E27250c91540911eD27F161572aFA53Ca24C0a
- ⚡ **TRX**: TVXWaU4ScNV9RBYX5RqFmySuB4zF991QaE
- 🚀 **LTC**: LN5ApP1Yhk4iU9Bo1tLU8eHX39zDzzyZxB

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/).*

```