Getting Started

Requirements check first. PHP 8.2+, ext-sockets-free (it uses Amp Socket, so no ext needed beyond the usual), and a NATS server with -js enabled if you want JetStream. The library is Amp 3.x based, which drives almost every design decision below.

Install and confirm the API surface. The README is thinner than src/, and the package has a small user base (8 stars, 0 dependents), so treat the source as the spec:

composer require idct/php-nats-jetstream-client
ls vendor/idct/php-nats-jetstream-client/src/{Core,JetStream,Connection,Services}

Where to look, in order:

src/Core/NatsClient.php — connect, publish, subscribe, request, drain/close. This is your entry point for everything.
src/Connection/NatsOptions.php — hosts, TLS, auth, reconnect, timeouts. All configuration lives here.
src/JetStream/ — the context object ($client->jetStream()), stream/consumer config builders, KeyValue, ObjectStore.
features/ (Behat Gherkin) — the most honest documentation in the repo. Files are grouped by core, auth, jetstream-core, jetstream-data, services, resilience, and each one is a runnable statement of intended behavior.

Minimal first connection. Names below follow the source layout; verify exact builder/signature names against your installed version before pasting into production code.

use Amp\Loop;
use IDCT\NATS\Connection\NatsOptions;
use IDCT\NATS\Core\NatsClient;

$options = (new NatsOptions())
    ->withHost('127.0.0.1')
    ->withPort(4222);

$client = new NatsClient($options);
$client->connect();

$client->publish('demo.hello', 'world');

$sub = $client->subscribe('demo.>');
while ($message = $sub->next()) {
    echo $message->subject . ': ' . $message->payload . PHP_EOL;
}

Best first use case. Do not start with plain pub/sub — you can get that from any NATS client. Start by replacing a sleep()-polling database queue with a JetStream durable pull consumer plus explicit acks. That exercises the three things this package actually exists for (async event loop, JetStream context, ack semantics), and it fails loudly if you get the config wrong.

$js = $client->jetStream();

$consumer = $js->consumer('ORDERS', 'billing-worker'); // durable
while (true) {
    foreach ($consumer->fetch(batch: 10, timeoutMs: 1000) as $message) {
        process($message->payload);
        $message->ack();          // explicit ack, not auto
    }
}

Get a local broker up before writing a line of application code. The repo's docker-compose.yml is a ready-made fixture stack: client ports 14222–14229 (with 14229 = WebSocket) and monitoring ports 18222–18229, covering token, user/pass, TLS, NKey, and JWT auth. Copy it into your app's CI rather than hand-rolling NATS config — it is the single most reusable artifact in this repository.

Implementation Patterns

Long-running worker shape

Every real deployment of this package is a long-lived CLI process. Structure it as: build options → connect → assert ConnectionState is connected → register signal handlers → run the consume loop in Amp\async() → drain on SIGTERM (drain, not close: it flushes in-flight messages and unsubscribes cleanly).

$future = Amp\async(function () use ($consumer) {
    while (!$this->shouldStop) {
        $this->consumeBatch($consumer);
    }
});

Amp\Loop::onSignal(SIGTERM, function () use (&$shouldStop) { $shouldStop = true; });
$future->await();
$client->drain();

drain() vs close() matters: close() drops unacked messages back to the server for redelivery, which is correct on crash but causes spurious reprocessing on every deploy.

Publish/subscribe with queue groups

For horizontally scaled workers, a queue group gives you NATS-side load balancing for free; no extra coordination service needed.

$client->subscribe('jobs.email', queue: 'email-workers');

Use core NATS + queue groups for fire-and-forget work, JetStream for anything you cannot afford to lose. The dividing line is "am I willing to lose this message if the process dies right now?" — if no, you need JetStream and acks.

Request/reply as internal RPC

$response = $client->request('users.lookup', json_encode(['id' => 42]), timeoutMs: 500);

Wire timeouts deliberately and short (200–1000 ms). No responders raises an error rather than hanging, but only after your timeout elapses — a fan-out of slow RPCs will silently serialize your request thread.

JetStream: streams, durable consumers, idempotency

Create streams with a config builder rather than raw subjects, and always set:

subjects covering exactly what you publish — over-broad subjects let one bad publisher pollute an unrelated stream
file storage for anything durable (memory storage disappears on restart, which surprises people in staging)
maxAge / maxBytes limits so a stuck consumer does not fill the disk

For publishing, the ack is a future. Await it before considering the write durable — otherwise you have built a fire-and-forget system with extra steps:

$ack = $js->publish('orders.created', $payload, ['Nats-Msg-Id' => $orderId]);
$ack->await(); // confirmed persisted by the stream leader

Nats-Msg-Id gives you server-side deduplication within the stream's duplicate_window. Use a natural key (order ID, event ID), not a random UUID — a random UUID dedupes nothing.

KeyValue

KV is a JetStream stream with a KV_ prefix, so the mental model is "eventually consistent cache/map", not "database". Reach for it for feature flags, small config blobs, and distributed locks via compare-and-set. Do not reach for it for anything you need to query or range-scan cheaply.

ObjectStore

For payloads above the comfortable message size (roughly >1 MB), ObjectStore chunks and reassembles for you. The tradeoff is that a read fetches chunks sequentially — for hot large objects, cache at the application layer rather than re-fetching per request.

Service framework

src/Services/ gives you endpoint registration, schema validation, and service grouping on top of request/reply. It is worth using if you are building a genuine PHP microservice mesh on NATS; if you are only a producer or consumer inside an existing Laravel/Symfony app, skip it and use plain request()/subscribe(). The framework is the least-documented area and the one most likely to shift.

Auth and TLS

Pick exactly one auth mode per connection and configure it in NatsOptions: token, user/pass, credentials file (.creds, which the library parses and nonce-signs), or NKey seed. For credentials/NKey, the library handles the CONNECT nonce signing — you supply the seed or file path, never the signed nonce. TLS options accept a CA path and client cert/key; verify against your fixture stack (14225-ish in the compose file) before pointing at production, because TLS misconfiguration surfaces as a connect timeout, not a clear error.

Testing strategy that mirrors upstream

Copy the upstream split: unit-test your own message handlers in isolation (no broker), and gate broker-dependent tests behind an env flag exactly as they do with RUN_INTEGRATION=1. Their scripts/wait-for-nats-services.sh pattern — poll the monitoring port until ready — is the right way to avoid flaky CI; reuse it.

Gotchas and Tips

Blocking the event loop is the #1 way to break this library. Anything synchronous inside a coroutine — PDO, file_get_contents() on a network mount, a sync Redis client, sleep() — stalls every subscription on that connection. Use Amp\delay() instead of sleep()/usleep(), and keep DB/HTTP calls short or hand them to an async client. Symptoms look like "heartbeats missed / connection dropped" rather than "my handler is slow".

Ack timeout vs batch size. If you fetch(100) but process serially at 50 ms each, messages 90–100 sit unacked past ack_wait and get redelivered to another worker — you get duplicate processing that looks like a library bug. Size the fetch batch to ack_wait / per-message-processing-time, and prefer ackSync() over ack() when you need to guarantee the server got it.

max_ack_pending is a consumer-side flow control knob, not a suggestion. Set it and the server stops delivering beyond it. With multiple workers on one durable consumer, a low value on one worker starves only that worker; a high value lets one slow worker hoard messages.

JetStream permission mistakes fail late and quietly. A user allowed to publish orders.> but not $JS.API.> will publish fine and fail every stream/consumer API call. Always grant $JS.API.> (scoped tighter in production) and test with the real credentials, not nats CLI admin.

"No responders" is an exception, not an empty reply. Wrap request() and treat no-responders as retryable-with-backoff; treat a returned error payload as a business error. Do not conflate the two.

Empty fetch is not an error. A pull fetch() with no traffic returns an empty collection after the timeout. Loop with a small delay or a longer timeout — a tight while(true) { fetch(1, 1); } will peg a CPU and hammer the server.

Reconnect restores subscriptions but not your application state. After a reconnect, core subscriptions come back and JetStream durable consumers resume from the server-side cursor — but any in-memory batching buffer, dedupe set, or "current position" you tracked locally is gone. Persist progress server-side (acks) or in your own store.

One client per process, never shared across forks. If you use pcntl_fork() or a preforking server, create the client after the fork. A copied connection gives two processes writing interleaved frames on one socket — corruption that is very hard to diagnose.

KV and ObjectStore both live on JetStream. If the server lacks -js, $client->jetStream() calls fail; so do KV and ObjectStore. Feature-detect at boot rather than discovering it on the first write.

Headers are validated by the library — do not hand-roll them. The protocol layer rejects CR/LF injection into headers, which is the right defensive posture. Build headers through the message/header API; if you find yourself concatenating wire strings, you are bypassing the one place that guards framing.

Extension points worth knowing. src/Transport/ is an interface with an Amp socket implementation — this is the seam for a custom transport (proxies, fault injection, capture in tests). src/Exception/ is a typed hierarchy: catch the specific subclasses rather than \Throwable, so a protocol violation and a business-level error are distinguishable in logs.

Pin the version. BSD-3-Clause, one maintainer org, 0 known dependents, and a release stream that moves. composer require with a caret on a specific minor, commit the lockfile, and read src/ before relying on an undocumented method — the AGENTS.md notes that public APIs and exception behavior are intended to stay stable, but the small install base means you are partly the QA.

Don't fight the async model at the boundaries. If your framework (Laravel's queue worker, Symfony Messenger, RoadRunner) already owns the process loop, run the NATS client inside a single Amp\async() block and hand messages to your framework's dispatch as they arrive — do not ->await() a long-running future inside a controller or middleware, or you have converted an async client into a blocking one with extra dependencies.

Php Nats Jetstream Client Laravel Package