nats.php

Modern PHP 8.5 client library for NATS messaging system with zero runtime dependencies.

Covers the NATS feature set: core messaging, JetStream, Key-Value, Object Store, and the Micro service framework.

Requirements

PHP >= 8.5
ext-sodium (bundled with PHP)

Installation

composer require sofascore/nats.php

Quick Start

use Sofascore\Nats\Connection;
use Sofascore\Nats\Message;

// Connect
$conn = Connection::connect(); // localhost:4222

// Async subscribe + process loop
$conn->subscribe('greetings', function (Message $msg) {
    echo "{$msg->subject}: {$msg->data}\n";
});
$conn->publish('greetings', 'Hello NATS!');
$conn->process(0.5);

// JetStream
$js = $conn->jetStream();
$kv = $js->createKeyValue(new \Sofascore\Nats\KeyValue\KeyValueConfig(bucket: 'cache'));
$kv->put('key', 'value');
$entry = $kv->get('key');

$conn->close();

For full runnable scripts grouped by feature, see the example files below. That includes a dedicated request/reply example. The later snippets in this README are shorter API overviews.

Features

Feature	Description
Core NATS	Publish, subscribe (sync/async), request/reply, queue groups, wildcards, RTT, barrier
Headers	HTTP-style message headers (NATS/1.0 wire format)
Authentication	User/pass, token, NKey (Ed25519/sodium), JWT, credentials file, TLS
Reconnection	Automatic reconnect with jitter, buffer, custom delay, server discovery, dynamic handlers
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	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

Architecture

readonly classes for all immutable data (configs, entries, info, metadata)
Backed string enums for all policy types — JSON serialize/deserialize friendly
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
sodium_crypto_sign_detached for NKey auth — bundled in PHP

Runnable Examples

Core

Example	Description
01-connect.php	Connection variants (default, URL, cluster, options)
02-auth.php	User/pass, token, NKey, JWT, TLS
03-pub-sub.php	Publish, subscribe (async/sync), wildcards, queue groups
04-request-reply.php	Synchronous RPC, request with headers
05-headers.php	Set/get/iterate headers, wire format
06-events-reconnect.php	Event handlers, reconnect config, custom backoff, forceReconnect
07-connection-info.php	RTT, addresses, client info, serverInfo, server pool, barrier
08-subscriptions-advanced.php	Pending limits, max pending, queued msgs, closed handler
09-error-handling.php	Exception hierarchy, no-responders, JetStream errors, retry
10-safeguards.php	Max payload, subject validation, slow consumer detection
11-graceful-shutdown.php	Drain vs close, lame duck, signal handling, status transitions

JetStream

Example	Description
01-streams.php	Create, info, getMessage, purge, list, delete
02-consumers.php	Pull fetch/next/messages/consume, ConsumeContext, ordered, ACK variants
03-publish.php	Deduplication, optimistic concurrency, sync publish
04-streams-advanced.php	CreateOrUpdate, streamNameBySubject, secureDelete, mirrors, sources, subject transforms
05-consumers-advanced.php	Pause/resume, push consumer CRUD/consume, backoff, fetchBytes, headersOnly
06-publish-advanced.php	msgTtl, expectLastMsgId, stallWait, publishMessage
07-direct-get.php	Direct message access, account info, fast lookups

Key-Value

Example	Description
01-basic.php	Put/get/create/update/delete/purge, history, listKeys, status
02-watch.php	Real-time watching, patterns, updatesOnly, includeHistory
03-advanced.php	getRevision, keyTtl, listKeysFiltered, watchFiltered, bucket management

Object Store

Example	Description
01-basic.php	Put bytes/string/file, get, delete, list, metadata, status
02-advanced.php	getInfo, addLink, addBucketLink, seal, watch, showDeleted, compression

Micro Services

Example	Description
01-service.php	Endpoints, groups, discovery, handler classes, statistics

API Overview

The snippets below are compact reference examples. For complete setup/cleanup flow, prefer the runnable files above.

Core — Connect

use Sofascore\Nats\Connection;
use Sofascore\Nats\ConnectionOptions;

// Default (localhost:4222)
$conn = Connection::connect();

// Multiple servers (automatic failover)
$conn = Connection::connect(['nats://srv1:4222', 'nats://srv2:4222']);

// With options (named constructor arguments)
$conn = Connection::connect('nats://localhost:4222', new ConnectionOptions(
    name: 'my-app',
    timeout: 5.0,
    maxReconnects: 10,
    reconnectWait: 1.0,
));

The built-in core transport only supports nats://, tls://, and nats+tls:// URLs. For ws:// / wss:// deployments or other custom dial behavior, pass a custom ConnectionOptions(transportFactory: ...).

transportFactory is a low-level hook. The factory must return a fresh TransportInterface instance because the client recreates transports on reconnect. When a custom transport is used, TransportInterface::connect() receives the original server URL (for example ws://gateway.example.test/nats) rather than a resolved tcp://host:port address, so the transport owns scheme/path-specific dialing. If the transport exposes a live stream resource from getStreamResource(), helpers such as connectedAddr(), localAddr(), and tlsConnectionState() can surface runtime socket metadata; otherwise those helpers return null.

Core — Pub/Sub

// Async subscribe (callback)
$conn->subscribe('orders.>', function (Message $msg) {
    echo "{$msg->subject}: {$msg->data}\n";
});
$conn->publish('orders.created', '{"id": 1}');
$conn->process(0.5); // Drive callback delivery

// Sync subscribe
$sub = $conn->subscribeSync('events.*');
$msg = $sub->nextMessage(timeout: 1.0);

// Queue groups (load balancing)
$conn->queueSubscribe('tasks', 'workers', function (Message $msg) {
    echo "Processing: {$msg->data}\n";
});

// Auto-unsubscribe after N messages
$sub->autoUnsubscribe(100);

Core — Request/Reply

$conn->subscribe('math.add', function (Message $msg) {
    $data = json_decode($msg->data, true);
    $msg->respond(json_encode(['result' => $data['a'] + $data['b']]));
});

$reply = $conn->request('math.add', '{"a":5,"b":3}', timeout: 2.0);
echo json_decode($reply->data, true)['result']; // 8

If you are running a long-lived service with callback subscriptions, you do not need to spin your own while (true) { $conn->process(); } loop unless you want custom tick timing. Use:

$conn->subscribe('math.add', function (Message $msg) {
    $msg->respond('ok');
});

$conn->wait();        // Blocks and keeps dispatching callbacks
// or:
$conn->process(0.5);  // One foreground tick you drive yourself

Core — Headers

use Sofascore\Nats\Headers;
use Sofascore\Nats\Message;

$headers = new Headers(['Content-Type' => 'application/json', 'X-Trace' => 'abc']);
$msg = new Message(subject: 'events.order', data: '{}', headers: $headers);
$conn->publishMessage($msg);

Core — Authentication

// User/Password
$options = (new ConnectionOptions())->withUserInfo('user', 'pass');

// Token
$options = (new ConnectionOptions())->withToken('s3cr3t');

// NKey (Ed25519)
$options = (new ConnectionOptions())->withNkey($nkeySeed);

// Credentials file
$options = (new ConnectionOptions())->withCredentials('/path/to/user.creds');

// TLS
$options = new ConnectionOptions(
    tlsEnabled: true,
    tlsCertFile: '/path/to/cert.pem',
    tlsKeyFile: '/path/to/key.pem',
    tlsCaFiles: ['/path/to/ca.pem'],
);

Core — Events & Reconnect

$options = new ConnectionOptions(
    maxReconnects: -1,              // Infinite
    reconnectWait: 2.0,
    reconnectJitter: 0.5,
    reconnectJitterTls: 1.0,
    reconnectBufferSize: 8 * 1024 * 1024,
    onConnect: fn(Connection $c) => echo "Connected to {$c->connectedUrl()}\n",
    onDisconnect: fn(Connection $c) => echo "Disconnected!\n",
    onReconnect: fn(Connection $c) => echo "Reconnected!\n",
    onError: fn(Connection $c, \Throwable $e, ?\Sofascore\Nats\Subscription $sub = null) => echo "Error: {$e->getMessage()}\n",
);

onError handlers receive the related Subscription as an optional third argument when the async error is subscription-scoped, such as slow-consumer detection.

JetStream — Streams

use Sofascore\Nats\JetStream\Stream\StreamConfig;
use Sofascore\Nats\Enum\{RetentionPolicy, StorageType, DiscardPolicy};

$js = $conn->jetStream();

$stream = $js->createStream(new StreamConfig(
    name: 'ORDERS',
    subjects: ['orders.>'],
    retention: RetentionPolicy::Limits,
    maxAge: 86400,
    maxBytes: 100 * 1024 * 1024,
    storage: StorageType::File,
    discard: DiscardPolicy::Old,
));

$ack = $js->publish('orders.new', '{"id": 1}');
echo "seq={$ack->sequence}\n";

$rawMsg = $stream->getMessage(1);
$stream->purge();

JetStream — Consumers

use Sofascore\Nats\JetStream\Consumer\{
    ConsumerConfig,
    ConsumeLoopOptions,
    FetchOptions,
    MessagePollOptions,
};
use Sofascore\Nats\Enum\{DeliveryPolicy, AckPolicy};

$consumer = $stream->createOrUpdateConsumer(new ConsumerConfig(
    name: 'processor',
    durable: 'processor',
    deliverPolicy: DeliveryPolicy::All,
    ackPolicy: AckPolicy::Explicit,
    filterSubject: 'orders.>',
));

// Fetch batch
$batch = $consumer->fetch(10, new FetchOptions(timeout: 2.0));
foreach ($batch as $msg) {
    echo $msg->data() . "\n";
    $msg->ack();
}

// Single message
$msg = $consumer->next(timeout: 2.0);
$msg->ack();

// Pull iterator
$msgs = $consumer->poll(new MessagePollOptions(maxMessages: 100, expires: 5.0));
foreach ($msgs as $msg) {
    $msg->ack();
}

// Poll one message at a time without handling TimeoutException yourself.
// If you raise maxMessages above 1, the context buffers a fetched batch.
$poller = $consumer->poll(new MessagePollOptions(maxMessages: 1, expires: 1.0));
while (($msg = $poller->tryNext()) !== null) {
    $msg->ack();
}

// Foreground callback loop
$ctx = $consumer->consumeLoop(
    function ($msg) use (&$ctx): void {
        $msg->ack();
        $ctx->drain();
    },
    new ConsumeLoopOptions(maxMessages: 1, expires: 1.0),
);
$ctx->run();

// ACK variants: ack(), ackSync(), nak(), nakWithDelay(5.0), term(), inProgress()

If you are migrating from older queue-style snippets, the modern equivalents are:

Older pattern	Current API
`create()->getQueue()->fetchAll(100)`	`fetch(100, ...)` and iterate the batch
`getQueue()->next()`	`next()` or `poll(new MessagePollOptions(maxMessages: 1, ...))->tryNext()`
`handle(...)`	`consumeLoop(...)->run()`

JetStream - Push Consumers

use Sofascore\Nats\JetStream\Consumer\{ConsumerConfig, PushConsumeOptions};
use Sofascore\Nats\JetStream\Message\JetStreamMessage;
use Sofascore\Nats\Enum\AckPolicy;

$orders = $js->stream('ORDERS');
$push = $orders->createPushConsumer(new ConsumerConfig(
    durable: 'push-orders',
    deliverSubject: 'deliver.orders',
    ackPolicy: AckPolicy::Explicit,
    flowControl: true,
    idleHeartbeat: 5.0,
));

$received = 0;
$ctx = null;
$ctx = $push->consume(
    function (JetStreamMessage $msg) use (&$received, &$ctx): void {
        $msg->ack();
        $received++;
        if ($received >= 3) {
            $ctx->stop();
        }
    },
    new PushConsumeOptions(
        errorHandler: function (\Throwable $e): void {
            echo "Push error: {$e->getMessage()}\n";
        },
    ),
);

while (!$ctx->isStopped()) {
    $conn->process(0.1);
    $push->checkHeartbeat(); // Useful when idleHeartbeat is configured
}

Push consume() subscribes immediately and runs from your application's connection loop. Flow-control replies are handled automatically. If you enable idleHeartbeat, call checkHeartbeat() alongside process() so total silence is surfaced as a consumer error.

JetStream — Publish Options

use Sofascore\Nats\JetStream\Publish\PublishOptions;

// Deduplication
$js->publish('orders.new', $data, PublishOptions::msgId('order-1'));

// Optimistic concurrency
$js->publish('orders.new', $data, PublishOptions::expectLastSequence(42));
$js->publish('orders.new', $data, PublishOptions::expectStream('ORDERS'));

// Synchronous publish acknowledgements
$ack = $js->publish('orders.new', $data);

Key-Value Store

use Sofascore\Nats\KeyValue\KeyValueConfig;

$kv = $js->createKeyValue(new KeyValueConfig(
    bucket: 'settings',
    history: 5,
    ttl: 3600.0,
));

// CRUD
$revision = $kv->put('theme', 'dark');
$entry = $kv->get('theme');          // KeyValueEntry{key, value, revision, operation, timestamp}
$rev = $kv->create('lang', 'en');    // Fails if exists
$rev = $kv->update('theme', 'light', $entry->revision);  // CAS
$kv->delete('lang');
$kv->purge('theme');

// History & keys
$history = $kv->history('counter');  // KeyValueEntry[]
$keys = $kv->listKeys();            // string[]

// Watch
$watcher = $kv->watch('>', WatchOptions::updatesOnly());
foreach ($watcher as $entry) {
    echo "{$entry->key} = {$entry->value}\n";
}

Object Store

use Sofascore\Nats\ObjectStore\ObjectStoreConfig;

$store = $js->createObjectStore(new ObjectStoreConfig(
    bucket: 'files',
    maxBytes: 100 * 1024 * 1024,
    maxChunkSize: 131072,
));

// Store
$info = $store->putBytes('config.json', $jsonBytes);
$info = $store->putString('readme.txt', 'Hello!');
// $info = $store->putFile('/path/to/file.zip');

// Retrieve
$result = $store->get('config.json'); // Lazy chunk reader
$chunk = $result->read(8192);
$result->close();

$content = $store->getBytes('config.json');
$text = $store->getString('readme.txt');
$store->getFile('archive.tar', '/tmp/archive.tar'); // Streams directly to disk

// List & delete
$objects = $store->list();           // Latest ObjectInfo[] per object name
$history = $store->list(includeHistory: true); // Full metadata history
$store->delete('config.json');

Object Store reads now use internal pull consumers for lazy chunk streaming and latest-only listing. Deployments with granular JetStream auth need CONSUMER.CREATE and CONSUMER.DELETE on the OBJ_<bucket> stream in addition to existing stream read permissions.

Micro Services

use Sofascore\Nats\Micro\{Service, ServiceConfig, EndpointConfig, RequestInterface};

$svc = Service::create($conn, new ServiceConfig(
    name: 'calculator',
    version: '1.0.0',
));

// Add endpoints
$svc->addEndpoint('add', function (RequestInterface $req) {
    $data = json_decode($req->data(), true);
    $req->respondJson(['result' => $data['a'] + $data['b']]);
}, new EndpointConfig(subject: 'calc.add'));

// Groups
$math = $svc->addGroup('math');
$math->addEndpoint('sqrt', function (RequestInterface $req) {
    $req->respondJson(['result' => sqrt(json_decode($req->data(), true)['n'])]);
});

// Auto-discovery: $SRV.PING, $SRV.INFO.calculator, $SRV.STATS.calculator
$info = $svc->info();
$stats = $svc->stats();
$svc->stop();

Testing

# Unit tests
vendor/bin/phpunit --testsuite=unit

# Integration tests (requires NATS server)
docker compose up -d
vendor/bin/phpunit --testsuite=integration

# Static analysis
vendor/bin/phpstan analyse src/

Docker

The project includes a docker-compose.yml for integration testing:

docker compose up -d    # Starts nats-server on port 4222
docker compose down     # Stop

Project Structure

src/
├── Connection.php              # Main class — connect, pub/sub, request, event loop
├── ConnectionOptions.php       # Immutable connection options (readonly, withers)
├── Subscription.php            # Sync/async subscriptions
├── Message.php                 # readonly — Subject, Data, ReplyTo, Headers
├── Headers.php                 # Header map (IteratorAggregate, Countable)
├── Inbox.php                   # _INBOX.<nuid> generator
├── ServerInfo.php              # readonly — INFO payload
├── Statistics.php              # readonly — in/out msgs/bytes/reconnects
├── NatsException.php           # Base exception
│
├── Enum/                       # Backed string enums
├── Protocol/                   # Parser (state machine), Writer (formatters)
├── Transport/                  # TCP/TLS transports, ring buffer
├── Auth/                       # UserPass, Token, NKey, JWT, Credentials
├── JetStream/                  # Streams, consumers, publish, iterators
│   ├── Stream/                 # StreamConfig, StreamInfo, Stream
│   ├── Consumer/               # ConsumerConfig, fetch/consume/messages
│   ├── Publish/                # PubAck, PublishOptions
│   ├── Message/                # JetStreamMessage (ack/nak/term)
│   └── Iterator/               # Paginated list iterators
├── KeyValue/                   # KV store, watch, history
├── ObjectStore/                # Chunked object storage
└── Micro/                      # Service framework, discovery

License

MIT

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
docs/examples		docs/examples
src		src
tests		tests
.gitignore		.gitignore
.php-cs-fixer.dist.php		.php-cs-fixer.dist.php
README.md		README.md
composer.json		composer.json
docker-compose.yml		docker-compose.yml
phpstan.neon		phpstan.neon
phpunit.xml		phpunit.xml

Navigation Menu

Repository navigation

About

Repository files navigation

nats.php

Requirements

Installation

Quick Start

Features

Architecture

Runnable Examples

Core

JetStream

Key-Value

Object Store

Micro Services

API Overview

Core — Connect

Core — Pub/Sub

Core — Request/Reply

Core — Headers

Core — Authentication

Core — Events & Reconnect

JetStream — Streams

JetStream — Consumers

JetStream - Push Consumers

JetStream — Publish Options

Key-Value Store

Object Store

Micro Services

Testing

Docker

Project Structure

License

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

nats.php

Requirements

Installation

Quick Start

Features

Architecture

Runnable Examples

Core

JetStream

Key-Value

Object Store

Micro Services

API Overview

Core — Connect

Core — Pub/Sub

Core — Request/Reply

Core — Headers

Core — Authentication

Core — Events & Reconnect

JetStream — Streams

JetStream — Consumers

JetStream - Push Consumers

JetStream — Publish Options

Key-Value Store

Object Store

Micro Services

Testing

Docker

Project Structure

License

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages