swsn / react-nats

dev-main 2026-10-02 22:00 UTC

Requires

• php: ^8.4
• evenement/evenement: ^3.0
• react/event-loop: ^1
• react/promise: ^3
• react/socket: ^1.16

Requires (Dev)

• phpunit/phpunit: ^11.0

MIT

Published on Packagist.org on 2026-10-02 20:00 UTC

README

Async NATS Core client for ReactPHP.
Non-blocking TCP I/O via react/socket, promises via react/promise, events via evenement.

Namespace	SWSN\Nats\
PHP	^8.4
Protocol	NATS Core (not JetStream / not NATS Request-Reply helpers yet)
License	MIT

Features

• Connect with server INFO handshake, then client CONNECT + keepalive PING
• Automatic reconnect with exponential backoff (DNS/TCP failure and dropped sockets)
• Replays SUB (same sid + queue group) after reconnect; close() stops retries
• PUB / SUB / UNSUB
• Queue groups on subscribe() for load-balanced delivery
• Automatic PONG replies to server PING
• Incremental framing (chunked TCP, sized MSG payloads)
• EventEmitter hooks: info, connect, reconnect, pong, error, close

Not included (yet): JetStream, TLS/nats:// URL parsing beyond tcp://, auth tokens/user-pass in CONNECT options (encoder supports custom CONNECT options if you wire them), request/reply inbox helpers.

Installation

Path repository (as used by SkyMx):

{
  "repositories": [
    {
      "type": "path",
      "url": "_libs/react-nats",
      "options": { "symlink": true }
    }
  ],
  "require": {
    "swsn/react-nats": "*@dev"
  }
}

Standalone (from this directory):

composer install

Quick start

<?php

use React\EventLoop\Loop;
use SWSN\Nats\NatsClient;

$nats = new NatsClient();

$nats->connect('tcp://127.0.0.1:4222')->then(
    function (NatsClient $client): void {
        $client->subscribe('demo.hello', function (string $subject, string $payload): void {
            echo "{$subject}: {$payload}\n";
        });

        $client->publish('demo.hello', 'world');
    },
    function (\Throwable $e): void {
        fwrite(STDERR, 'connect failed: ' . $e->getMessage() . "\n");
    },
);

Loop::run();

Queue groups (load balancing)

With a queue group, NATS delivers each published message to only one subscriber in that group (random member). Without a group, every subscriber receives a copy (fan-out).

// Workers share the group name — one worker handles each wakeup.
$client->subscribe(
    'skymx.queue.wakeup',
    function (string $subject, string $payload): void {
        // claim next outbound job …
    },
    'skymx-workers',
);

// Fan-out (omit queue group): every process must see the event.
$client->subscribe('skymx.config.changed', function () use ($settings): void {
    $settings->warm();
});

Wire form: SUB <subject> <queue> <sid>\r\n.

Public API (NatsClient)

Constructor

new NatsClient(
    ConnectorInterface $connector = new Connector(),
    ?ServerFrameParser $parser = null,
    ?SubscriptionRegistry $subscriptions = null,
    ?LoopInterface $loop = null,
    float $retryMinSeconds = 2.0,
    float $retryMaxSeconds = 30.0,
);

Inject a custom ConnectorInterface for timeouts, TLS, or proxying. Parser/registry are optional seams for tests.
LoopInterface is used for reconnect timers (defaults to Loop::get()).

connect(string $uri = 'tcp://127.0.0.1:4222'): PromiseInterface

1. Opens the TCP connection
2. Waits for the server INFO … line
3. Sends CONNECT + PING, then replays any stored SUBs
4. Resolves with $this and emits connect

The promise resolves once, on the first successful handshake. Later reconnects emit connect and reconnect but do not require another connect() call.

TCP/DNS failures emit error and retry with exponential backoff ($retryMinSeconds … $retryMaxSeconds). The promise rejects only if close() is called before the first handshake.

Calling publish / subscribe before the promise resolves throws RuntimeException. publish also throws while a reconnect is in progress (fail cleanly; no outbound queue).

publish(string $subject, string $payload = ''): void

Sends a Core PUB with the given payload bytes (length is derived from strlen($payload)).

subscribe(string $subject, callable $handler, ?string $queueGroup = null): int

Registers a local handler and sends SUB.
Handler signature: function (string $subject, string $payload): void.

Optional $queueGroup: when set, NATS load-balances messages across all connections that joined the same group for that subject.

Returns the numeric subscription id (sid) used by the server.

unsubscribe(int $sid): void

Removes the local handler and sends UNSUB.

close(): void

Stops reconnecting, closes the socket, clears subscriptions, and emits close.
An unexpected peer drop also emits close, then the client reconnects on its own.

Events

NatsClient extends Evenement\EventEmitter.

Event	Payload	When
info	array $fields	Server INFO JSON decoded
connect	—	After CONNECT handshake completed (including reconnects)
reconnect	—	Handshake completed after a previous successful session
pong	—	Server answered our PING
error	\Throwable	Server -ERR …, TCP/DNS connect failure, or socket error
close	—	Socket closed (peer drop or close())

$nats->on('error', function (\Throwable $e): void {
    // log; the client retries unless you called close()
});

Server PING is answered automatically with PONG (no event).

Architecture

Protocol pieces follow the same split used elsewhere in SkyMx (framing / parser DTOs / codec / session):

src/
  NatsClient.php                      # façade: I/O + lifecycle
  Protocol/
    Framing/
      ByteBuffer.php                  # append / shiftLine / shiftSizedPayload
    Parser/
      ServerFrameParser.php           # incremental decode → ServerFrame DTOs
      MsgHeaderParser.php             # MSG subject sid [reply] size
      Dto/
        ServerFrame.php               # marker interface
        MsgFrame.php
        InfoFrame.php
        PingFrame.php
        PongFrame.php
        ErrFrame.php
        MsgHeader.php
    Codec/
      ClientCommandEncoder.php        # CONNECT / PUB / SUB / UNSUB / PING / PONG
    Session/
      SubscriptionRegistry.php        # sid → handler

Data path (inbound)

TCP chunk
  → ByteBuffer
  → ServerFrameParser::pull()
      → MsgFrame     → SubscriptionRegistry::dispatch(sid, subject, payload)
      → InfoFrame    → emit('info')
      → PingFrame    → write PONG
      → PongFrame    → emit('pong')
      → ErrFrame     → emit('error')

Data path (outbound)

NatsClient::publish/subscribe/…
  → ClientCommandEncoder::*
  → ConnectionInterface::write()

Protocol coverage

Direction	Opcode	Support
←	INFO	yes
→	CONNECT	yes (default options; encoder accepts overrides)
→	PUB	yes (optional reply-to in encoder; client API uses plain PUB)
→	SUB	yes (subject-only or subject + queue group)
→	UNSUB	yes
↔	PING / PONG	yes
←	MSG	yes (with optional reply-to header)
←	+OK	ignored (verbose: false by default)
←	-ERR	yes → error event

Testing

From the SkyMx monorepo:

vendor/bin/phpunit --testsuite nats

Or inside this package after composer install:

vendor/bin/phpunit

Unit tests cover framing, MSG header parsing (with/without reply-to), incremental payload assembly, command encoding, connect retry, SUB replay, and close() cancelling retries — no live NATS server required.

Design notes

• Async-first: never block the React event loop; all I/O is stream-based.
• Partial reads: MSG headers and payloads may arrive across multiple data events; the buffer waits until header + size + CRLF is complete.
• Queue groups vs fan-out: use a group for “one worker should act”; omit it when every instance must observe the event.
• Keep the façade thin: wire protocol details live under Protocol\* so the client stays readable and testable.
• Reconnect in the client: apps call connect() once, subscribe once; the library owns retry and re-SUB. close() is the stop signal.
• Stable consumer API: apps depend on SWSN\Nats\NatsClient only; internal parser/codec classes may evolve.