Small Swoole Rx Events

Reactive event bus for PHP powered by RxPHP and Swoole.

It lets you publish/subscribe domain and infrastructure events, compose pipelines with Rx operators, and run time-based operators on Swoole’s event loop.

• EventBus — simple Rx‐backed bus with on(), onMany(), payloads(), once(), request()
• SwooleScheduler — AsyncSchedulerInterface using Swoole\Timer (works with RxPHP time operators)
• Event model — BasicEvent (name, payload, meta, rid) and EventInterface (correlation id)

Requirements

• PHP 8.3+
• ext-swoole 4.8+ / 5.x
• reactivex/rxphp (2.x)

Installation

composer require small/swoole-rx-events 

Quick start

use Small\SwooleRxEvents\EventBus; use Small\SwooleRxEvents\SwooleScheduler; use Small\SwooleRxEvents\Event\BasicEvent; // Use the Swoole async scheduler $bus = new EventBus(new SwooleScheduler()); // Subscribe by name $bus->on('order.created')->subscribe(function ($e) { echo "order rid={$e->getRid()} payload=", json_encode($e->getPayload()), PHP_EOL; }); // Emit an event $bus->emitName('order.created', ['id' => 123]); // If you’re in a plain CLI script, keep the loop alive briefly: \Swoole\Timer::after(20, fn () => \Swoole\Event::exit()); \Swoole\Event::wait(); 

Concepts

Event

All event must implement EventInterface

namespace Small\SwooleRxEvents\Contract; interface EventInterface { public function getName(): string; public function getRid(): string; public function setRid(string $rid): self; } 

BasicEvent carries:

• name (string)
• payload (array)
• meta (array, e.g. tracing, user)
• rid (string, auto‐generated correlation id)

Bus

• stream() — all events
• on($name) / onMany([...]) — filtered streams
• payloads($name) — payload‐only stream
• once($name, ?map, ?timeoutMs) — resolve first matching event (optionally mapped)
• request($requestName, $responseName, $payload = [], $meta = [], ?$timeoutMs) Emits a request with a new rid, waits for the first response with the same rid.

Timeouts require an async scheduler. This library provides SwooleScheduler which implements AsyncSchedulerInterface.

API Examples

1) Listen & emit

$bus->on('user.created')->subscribe(fn($e) => audit($e->getMeta(), $e->getPayload())); $bus->emitName('user.created', ['id' => 42], ['by' => 'admin']); 

2) Request/Response with correlation id

// Responder: copies rid from incoming 'REQ' and emits 'RESP' $bus->on('REQ')->subscribe(function ($e) use ($bus) { $bus->emit( (new BasicEvent('RESP', ['ok' => true], $e->getMeta())) ->setRid($e->getRid()) // correlate ); }); // Caller: request() subscribes FIRST, then emits; no race conditions $bus->request('REQ', 'RESP', ['foo' => 'bar'], ['trace' => 'abc'], 100) ->subscribe( fn($resp) => var_dump($resp->getPayload()), // ['ok' => true] fn($err) => error_log($err->getMessage()) ); 

3) once() with mapping & timeout

$bus->once('health.ok', fn($e) => $e->getMeta()['node'] ?? 'unknown', 50) ->subscribe( fn($node) => echo "node=$node\n", fn($err) => echo "timeout\n" ); $bus->emitName('health.ok', [], ['node' => 'api-1']); 

4) Backpressure / batching (Rx composition)

$bus->on('order.created') ->bufferWithTimeOrCount(500, 100, $bus->scheduler()) // every 0.5s or 100 items ->filter(fn($batch) => !empty($batch)) ->subscribe(fn(array $batch) => persist_batch($batch)); 

Swoole integration tips

• HTTP server: in on('request'), emit an event with meta containing a respond callable or the Response object. Downstream subscribers can produce a ResponseEvent.
• Coroutines per subscriber: use Swoole coroutines in your subscribers if you do IO; Rx operators will orchestrate sequencing.
• Event loop in CLI: outside a Swoole Server, start/stop the reactor with Swoole\Event::wait() / Event::exit() for timers to fire.