Nats Laravel Package

Getting Started

Install and wire it up. There is no Laravel service provider or facade in this package, so day one is: composer require basis-company/nats, add a config file, and bind the client as a singleton.

// config/nats.php
return [
    'host'  => env('NATS_HOST', '127.0.0.1'),
    'port'  => (int) env('NATS_PORT', 4222),
    'user'  => env('NATS_USER'),
    'pass'  => env('NATS_PASS'),
    'delay' => 0.001,
];

// app/Providers/NatsServiceProvider.php
public function register(): void
{
    $this->app->singleton(Client::class, function () {
        $configuration = new Configuration(
            host: config('nats.host'),
            port: config('nats.port'),
            user: config('nats.user'),
            pass: config('nats.pass'),
        );
        $configuration->setDelay(config('nats.delay'));

        return new Client($configuration);
    });
}

Note the Configuration constructor takes named arguments for the static settings (host, port, credentials, TLS file paths, JWT/NKey), while behavioural settings — retry delay and mode — are mutated through setters afterwards. That split trips people up on the first pass.

Where to look first. Three entry points cover ~90% of usage:

First use case: an outbound event. The lowest-risk adoption path in a Laravel app is to keep Eloquent/MySQL as the source of truth and use NATS to fan out domain events to other services (or other parts of the same app). Publish from a listener, consume in an Artisan command. That gets you value without betting your job queue on a package with zero dependents.

$client->publish('orders.created', new Payload(json_encode($order), [
    'Nats-Msg-Id' => "order-{$order->id}",
]));

Sanity check connectivity with $client->ping() before writing any consumer code — it returns true and fails fast on bad credentials or TLS settings.

Implementation Patterns

The blocking model drives the architecture. This client is synchronous. subscribe(...)->next() blocks; process() runs a callback loop until interrupted. That is perfect for long-running daemons and wrong for HTTP requests. The practical rule for Laravel: publish inside the request cycle, consume inside php artisan commands supervised by Supervisor/Horizon-adjacent tooling.

class NatsConsumeCommand extends Command
{
    protected $signature = 'nats:consume {stream} {consumer}';

    public function handle(Client $client): int
    {
        $consumer = $client->getApi()->getStream($this->argument('stream'))
            ->getConsumer($this->argument('consumer'));

        $consumer->handle(function (Payload $payload) {
            // dispatch to a Laravel job if the work is heavy
            ProcessThing::dispatch($payload->payload);
        });

        return self::SUCCESS;
    }
}

Two schools of thought here. Cheap work can be handled inline (the README's mail() example). Anything with retries, backoff, or DB transactions is better handed to Laravel's queue after acking — you get failed_jobs, queue:retry, and all the tooling the NATS client simply doesn't provide.

Work-queue streams map cleanly onto Laravel jobs. The README's mailer example is exactly the shape you want:

$stream = $client->getApi()->getStream('mailer');
$stream->getConfiguration()
    ->setRetentionPolicy(RetentionPolicy::WORK_QUEUE)
    ->setStorageBackend(StorageBackend::FILE)
    ->setSubjects(['mailer.greet', 'mailer.bye']);
$stream->create();

$greeter = $stream->getConsumer('greeter');
$greeter->getConfiguration()->setSubjectFilter('mailer.greet');
$greeter->handle(fn ($address) => mail($address, 'Hi there!'));

One consumer per subject filter, one worker per consumer. RetentionPolicy::WORK_QUEUE gives you competing-consumer semantics across replicas; MEMORY is fine for tests but loses messages on restart — default to file storage in anything that resembles production.

Batching for throughput. setBatching(N) controls how many messages are requested per round-trip, setIterations(N) how many round-trips before stopping, and interrupt() stops cleanly at a batch boundary. For a nightly digest job, fetchAll() with a timeout is often simpler than a persistent handle() loop:

$queue = $goodbyer->setBatching(50)->create()->getQueue();
$messages = $queue->setTimeout(1)->fetchAll(100);
foreach ($messages as $m) { /* ... */ $m->ack(); }

Ack discipline is your retry policy. $message->ack() on success, $message->nack(1) to redeliver after a delay. Because nack/redelivery means at-least-once, every handler should be idempotent — key it on a business identifier, not on message order.

Request/response for inter-service RPC. dispatch() is the synchronous call; request() with a callback plus process() is the async form. This is a good fit for Laravel-to-Laravel service calls where you'd otherwise build an HTTP client and a timeout policy:

$client->subscribe('user.lookup', fn ($id) => User::find($id)?->toArray());
// caller
$user = $client->dispatch('user.lookup', $id);

Reuse the service abstraction for typed endpoints. Implementing EndpointHandler on a class gives you a container-friendly, testable unit, though the README's example passes the class name — wire your own resolver if you want constructor injection:

$version->addEndpoint('posts', IndexPosts::class);
$service->run();

KeyValue where you'd otherwise add Redis. Optimistic concurrency via getEntry()->revision + update() is genuinely useful for distributed locks/flags and small config blobs, and it inherits the stream's durability:

$entry = $bucket->getEntry('username');
$bucket->update('username', 'neky', $entry->revision); // throws on conflict

Use getAll() only for small buckets — it's a full read, not a scan API.

Gotchas and Tips

The README contains broken copy-paste snippets. The JWT example is missing a comma between the CredentialsParser::fromFile(...) argument and host:. The mutual-TLS example has stray quotes (tlsKeyFile':, tlsCaFile':) and tlsCaFile points at client-key.pem — a copy/paste artifact; point it at your CA bundle. Budget time for fixing these rather than assuming your setup is wrong.

process() is mandatory for callbacks. subscribe('x', fn() => ...) does nothing observable until you call $client->process(). Conversely $client->dispatch() blocks, so never put it behind a web endpoint without a hard timeout.

Consumers are created lazily. getConsumer() returns a handle; the server-side consumer is created on the first handle() call. If you want consumers provisioned ahead of the worker (e.g., in a deploy step or migration), call ->create() explicitly. Ephemeral consumers are the inverse: constructed immediately and deleted by the server shortly after the socket closes — don't rely on them surviving reconnects.

Deliver policy determines whether you replay history. DeliverPolicy::NEW (the ephemeral example) skips everything already in the stream. If you're wiring up a consumer for the first time on a stream with backlog, decide deliberately between NEW, ALL, and BY_START_TIME — the default choice is rarely the one you want after an incident.

Unsubscribe or leak. $client->unsubscribe($queue) is required when you break out of a while loop; otherwise the subscription stays alive on the connection and you keep receiving into a buffer nobody drains.

Retry/delay defaults are aggressive. Default mode is constant 1ms — a tight reconnect storm against a downed server. Use DELAY_EXPONENTIAL with a 10ms base for anything production-facing, and treat connection loss as expected, not exceptional.

There is no failed_jobs equivalent. No dead-letter queue, no retry bookkeeping beyond nack and server-side max_deliver. If you need auditability, you build it: persist failures to your own table inside the handler, and configure a max-delivery limit on the consumer so poison messages stop cycling.

Long-running workers leak. Standard PHP daemon hygiene applies — unset() large payloads, avoid accumulating in container-resolved singletons, and expose a restart signal (mirror queue:restart by checking a cache flag between setIterations batches and calling interrupt()).

Debugging is env-var driven. The perf test harness uses NATS_CLIENT_LOG and NATS_CLIENT_VERBOSE; the client logs protocol frames at DEBUG. Reach for that before assuming a routing problem, because subject typos and wildcard mismatches fail silently by design.

Testing needs a real server. There's no in-memory fake — spin up nats:latest with JetStream enabled (-js) in CI and point your test config at it. Budget for port conflicts and stream-name collisions between parallel test runs by namespacing subjects/streams per test class.

Watch the naming of consumer config setters. setSubjectFilter() (singular string) and setSubjectFilters() (array) both exist and are not interchangeable aliases — pick one and be consistent, because the mismatch surfaces as a consumer that receives nothing.

Package risk is real. 212 stars, zero dependents, README-and-releases as the only maturity signals. Pin the version, wrap every client interaction behind your own small interface (an EventBus/NatsClient contract), and keep a Redis/SQS fallback in mind. The upside — ~370k msg/s publish throughput on commodity hardware and a genuinely complete JetStream surface — is worth it for event fan-out and worker pipelines, but not worth coupling your domain code to directly.