README

Typed event bus on NATS.

Publish with EventBus::emit(). Consume with #[AsEventBusListener]. Deliver fanout (every subscriber) or compete (one member of a queue group).

This library does not replace an in-process event dispatcher (e.g. Symfony EventDispatcher). Use that for same-process domain hooks; use this bus when events must cross process or role boundaries over NATS.

The host injects SWSN\Nats\NatsClient (and/or MessagePublisherInterface). This package never constructs a NATS client or owns connect/URL lifecycle.

Concepts

Term	Meaning
Event	Serializable PHP object (typed payload)
Channel	Logical topic / NATS subject (`app.config.changed`)
Fanout	Every subscriber receives a copy
Compete	One subscriber in a queue group wins
Transport	NATS via injected client/publisher

Channel metadata lives on the event class:

use SWSN\NatsEventBus\Attribute\BusChannel;
use SWSN\NatsEventBus\Bus\BusDelivery;

#[BusChannel(name: 'app.config.changed', delivery: BusDelivery::Fanout)]
final class ConfigChanged
{
    public function __construct(public string $key) {}

    /** @return array{key: string} */
    public function __serialize(): array
    {
        return ['key' => $this->key];
    }

    /** @param array{key: string} $data */
    public function __unserialize(array $data): void
    {
        $this->key = $data['key'];
    }
}

Compete example with an explicit queue group:

#[BusChannel(
    name: 'app.jobs.resize',
    delivery: BusDelivery::Compete,
    queueGroup: 'image-workers',
)]
final class ResizeImageRequested { /* … __serialize / __unserialize … */ }

When queueGroup is omitted for compete delivery, the library derives event-bus-{channel} (dots → dashes).

Quick start (Symfony DI, no FrameworkBundle)

use SWSN\Nats\NatsClient;
use SWSN\NatsEventBus\Bus\EventBus;
use SWSN\NatsEventBus\DependencyInjection\ConfigureEventBus;
use SWSN\NatsEventBus\Transport\MessagePublisherInterface;
use SWSN\NatsEventBus\Transport\Nats\NatsClientPublisher;
use SWSN\NatsEventBus\Transport\Nats\NatsConnector;
use Symfony\Component\DependencyInjection\ContainerBuilder;

$container = new ContainerBuilder();
$container->register(NatsClient::class, NatsClient::class)->setAutowired(true);
$container->register(NatsClientPublisher::class, NatsClientPublisher::class)
    ->setAutowired(true);
$container->setAlias(MessagePublisherInterface::class, NatsClientPublisher::class);
$container->register(NatsConnector::class, NatsConnector::class)
    ->setAutowired(true)
    ->setArgument('$natsUrl', '%env(NATS_URL)%');

ConfigureEventBus::register($container);
$container->compile();

$bus = $container->get(EventBus::class);
$bus->emit(new ConfigChanged('mail.from'));

Consumer:

use SWSN\NatsEventBus\Attribute\AsEventBusListener;

final class ConfigReloadListener
{
    #[AsEventBusListener]
    public function onConfigChanged(ConfigChanged $event): void
    {
        // reload caches, etc.
    }
}

Without DI

use SWSN\NatsEventBus\Bus\BusDelivery;
use SWSN\NatsEventBus\Bus\EventBus;
use SWSN\NatsEventBus\Transport\Nats\NatsEventBusTransport;

$transport = new NatsEventBusTransport($publisher); // MessagePublisherInterface from host
$bus = new EventBus($transport);

$transport->subscribe('app.config.changed', BusDelivery::Fanout, function (
    string $channel,
    string $payload,
): void {
    $envelope = unserialize($payload, ['allowed_classes' => true]);
    // handle $envelope->event
});

$bus->emit(new ConfigChanged('mail.from'));

Delivery semantics

	Fanout	Compete
Intent	notify everyone	exactly one worker
NATS	normal subscribe	queue group

Subscribe only where you have consumers. Publish never joins a compete group, so publisher-only processes cannot steal jobs.

Package layout

src/
  Attribute/          #[BusChannel], #[AsEventBusListener]
  Bus/                EventBus, channels, envelopes, replay
  Transport/
    EventBusTransport.php
    MessagePublisherInterface.php
    Nats/             NATS transport, connector, publisher
  DependencyInjection/

Requirements

PHP ^8.4
swsn/react-nats (client injected by the host)
Optional: symfony/dependency-injection for ConfigureEventBus

Development

composer install
composer test

When this package is a path repository inside a host monorepo, run the host testsuite (or vendor/bin/phpunit --testsuite nats-event-bus from the host root).

License

MIT

swsn / nats-event-bus

Package info

Security

Statistics