repejota/nats
A Laravel/PHP client for NATS messaging that makes it easy to publish and subscribe to subjects, handle request/reply, and connect services with simple configuration. Suitable for lightweight microservices and plain PHP projects.
Install with Composer — the package name is repejota/nats (despite the repo being named phpnats):
composer require repejota/nats
Before writing any code, read two files in the repo: src/Nats/Connection.php (the only class you'll touch 95% of the time) and src/Nats/ConnectionOptions.php (the option keys and defaults). The tests/ directory doubles as usage documentation and is more reliable than the README, which has drifted across versions.
The smallest working flow against a local nats-server:
use Nats\Connection;
$conn = new Connection(); // defaults to 127.0.0.1:4222
$conn->connect();
$conn->publish('orders.created', json_encode(['id' => 42]));
$conn->subscribe('orders.created', function ($message) {
// $message is a Nats\Message object, not a string
logger()->info($message->getBody());
});
$conn->wait(1); // blocking read loop, 1 = seconds
$conn->close(); // or $conn->disconnect()
Your first real use case in a Laravel app is almost always publishing a domain event from a service or model observer, then running a standalone php artisan command as a subscriber under Supervisor. Don't try to subscribe from inside an HTTP request — wait() blocks the worker until the timeout expires.
Bind a singleton in a service provider. The connection wraps a single TCP socket and is not safe to open per-call. Register it once and resolve it via the container:
$this->app->singleton(Connection::class, function ($app) {
$options = new ConnectionOptions();
$options->setHost(config('nats.host'));
$options->setPort(config('nats.port'));
$options->setUser(config('nats.user'));
$options->setPass(config('nats.pass'));
$options->setTimeout(config('nats.timeout', 1));
$conn = new Connection($options);
$conn->connect();
return $conn;
});
Wrap this in a small NatsPublisher service class so application code depends on your abstraction, not on Nats\Connection directly. That makes it trivial to swap in a newer client later and to unit-test publishers with a fake.
Publish from a queued job, not from the request path. If NATS is unreachable, connect()/publish() will throw and you'll take down the user's request. Dispatch a job and publish there:
class PublishOrderCreated implements ShouldQueue
{
public function handle(NatsPublisher $nats): void
{
$nats->publish('orders.created', $payload, ['queue' => 'orders']);
}
}
Subscriber commands with an explicit loop. The idiomatic worker is a command that subscribes once and then re-enters wait() in a while loop, with your own shutdown handling:
class NatsListen extends Command
{
protected $signature = 'nats:listen {subject}';
public function handle(Connection $conn): int
{
$conn->subscribe($this->argument('subject'), function ($message) {
try {
$this->dispatch(new HandleMessage($message->getBody()));
} catch (\Throwable $e) {
report($e); // never let a callback throw out of wait()
}
});
while (true) {
$conn->wait(1); // returns after ~1s of silence
if ($this->shouldStop()) {
break;
}
}
return self::SUCCESS;
}
}
Run it under Supervisor with numprocs for horizontal scaling, and use NATS queue groups (if supported in your version — check the subscribe() signature) rather than PHP-level locking to distribute work.
Request/reply for synchronous lookups. NATS supports a reply subject; Connection::request($subject, $payload, callable $cb) wraps subscribe-with-reply-to. Use it sparingly — it reintroduces synchronous coupling, but it's handy for small config or RPC-style lookups where a queue round-trip is overkill.
Standardize an envelope. NATS payloads are opaque byte strings; the client does no serialization. Pick one envelope and stick to it in every publisher and consumer:
['type' => 'order.created', 'version' => 1, 'occurred_at' => ..., 'payload' => [...]]
Then your subscriber can decode once and route on type.
Subject naming. Treat subjects like Laravel event names: orders.created, orders.paid, users.updated. Publish to concrete subjects only; wildcards (* for one token, > for the rest) are subscribe-side only.
Lifecycle in tests. Since the connection is a container singleton, bind a fake NatsPublisher in TestCase::setUp() so your feature tests never try to open a socket.
wait() is blocking and CPU-idle. It reads the socket with a timeout. Always pass a finite timeout and loop, otherwise a misconfigured supervisor can hang an entire worker pool with no visible activity.try/catch. An exception propagates back through Connection::wait() and the process dies with no retry.flush() (or at least close()) before shutdown. publish() is fire-and-forget; on a short-lived script the process can exit before the server acknowledges the message. In long-running workers, flush() after a batch.Connection::connect() body and, if it doesn't, wrap your publish/subscribe in a reconnectWithBackoff() helper and listen for ConnectionException. Treat this as a "must verify" item on any new project.Nats\Message is not a string. $message is an object with getBody(), getSubject(), getReplyTo(), and getSid(). Calling (string) $message or passing it into json_decode silently fails.nats-server with a TLS leaf and connect to a localhost plaintext listener, or terminate TLS in a sidecar. Don't assume setHost() accepts a tls:// scheme.subscribe() returns a SID; keep it. If you ever need unsubscribe(), that string is the only handle. Duplicate subscriptions to the same subject on one connection each get their own SID and each handler fires — a common source of duplicate processing.max_payload (default 1 MB) before publishing blobs; oversize messages fail server-side, and depending on version you may not get a clean exception.$options->setDebug(true) dumps protocol traffic to stdout. Turn it on only in a local terminal, never under Supervisor (it will fill your log at message rate).new Connection(['host' => ...]) works in some versions; using ConnectionOptions with explicit setHost()/setPort() is the portable path.Connection::connect(), subscribe(), and wait(). If you need metrics, wrap Connection in a decorator and time publish() / count wait() iterations — that's cheaper than forking the library.How can I help you explore Laravel packages today?