README

A framework-agnostic event bus and reliable message queue for PHP.

Omnibus provides one explicit lifecycle for synchronous commands, PSR-14 events, queued listeners, delayed work, durable consumers, workflows, scheduling adapters, and broadcasts. It works as a standalone Composer library and does not require a framework or command package.

Install

composer require infocyph/omnibus

Requirements:

64-bit PHP ^8.4 with ext-ctype (required by UID 6)
infocyph/uid:^6.0
psr/clock:^1.0
psr/event-dispatcher:^1.0

DBLayer, CacheLayer, Redis/Valkey clients, broker SDKs, Runwire, PCNTL, and POSIX remain optional capabilities. Ordinary request/FPM, direct dispatch, Consumer, and single-process Worker paths do not require process extensions. Explicit Worker signal handling requires ext-pcntl; the standalone native and Runwire WorkerPool backends require the process capabilities they use.

Database integrations support DBLayer 6.x. CacheLayer coordination integrations support CacheLayer 4.x. Runwire integration supports Runwire 2.x from 2.1.1. DBLayer owns database execution, driver behavior, effective bind limits, transaction retries, and after-commit callback lifecycle. Omnibus owns reservations, workflows, failure-state transitions, and delivery guarantees. Queue receive and workflow claim limits remain caller-facing maxima; the adapters may return fewer rows so one atomic reservation or claim stays within the active connection's bind budget.

Highlights

Explicit, cached route, handler, listener, codec, transport, and factory maps
Direct synchronous handlers and ordered PSR-14 events
Ordered handler middleware shared by synchronous and consumer execution
In-memory, DBLayer, Redis/Valkey, AMQP, and SQS transport boundaries
Bounded consumers, long-running workers, and optional fixed process concurrency
Portable cooperative worker heartbeat and external graceful-stop integration
Conditional reservation settlement and visibility-based crash recovery
Bounded retries, poison-payload capture, and durable failure management
Safe versioned JSON envelopes with allow-listed aliases and strict limits
CacheLayer uniqueness, overlap, rate-limit, and circuit-breaker decorators
Redis-free operation through DBLayer, including zero-service SQLite; Memcached may back lease-based uniqueness and overlap but is not a queue transport
Persistent chains and batches with idempotent terminal transitions
Provider-neutral scheduling, broadcasting, after-response, and telemetry
No filesystem scanning, hidden provider initialization, or runtime discovery

Quick start

use Infocyph\Omnibus\Envelope\HandledStamp;
use Infocyph\Omnibus\Handler\HandlerInvoker;
use Infocyph\Omnibus\Handler\HandlerMap;
use Infocyph\Omnibus\MessageBus;
use Infocyph\Omnibus\Routing\RouteMap;
use Infocyph\Omnibus\Transport\SyncTransport;
use Infocyph\Omnibus\Transport\TransportRegistry;

$handlers = new HandlerMap([
    CreateInvoice::class => static fn (CreateInvoice $message): string =>
        $invoiceService->create($message),
]);
$invoker = new HandlerInvoker($handlers);

$bus = new MessageBus(
    new RouteMap(),
    new TransportRegistry([
        'sync' => new SyncTransport($invoker),
    ]),
);

$result = $bus->dispatch(new CreateInvoice($accountId));
$invoiceId = $result->last(HandledStamp::class)?->result;

At bootstrap, pass a route map like this to MessageBus together with a registered Redis transport to send selected messages asynchronously:

use Infocyph\Omnibus\Routing\Route;
use Infocyph\Omnibus\Routing\RouteMap;

$routes = new RouteMap([
    CreateInvoice::class => new Route(
        transport: 'redis',
        queue: 'billing',
        delaySeconds: 2.0,
    ),
]);

Handler middleware

HandlerInvoker applies one ordered, framework-neutral middleware stack to both synchronous and consumer handling. Middleware can observe HandlerContext delivery metadata, short-circuit handling, transform the synchronous result, or throw into the consumer's existing retry/failure path. It does not intercept routing, serialization, transport I/O, PSR events, or worker process lifecycle. See the handler middleware guide and 3.0 upgrade notes.

Consumer::run() performs one bounded receive call. Worker provides the long-running loop for one process. Hosts may supply a framework-neutral WorkerLifecycle for heartbeat and graceful external-stop policy without requiring signal delivery for each lifecycle decision. WorkerOptions defaults to signal-free operation; opt into SIGTERM/SIGINT handling explicitly with handleSignals: true when ext-pcntl is available. The optional native WorkerPool requires ext-pcntl and ext-posix. Runwire 2.x from 2.1.1 remains an optional explicitly selected alternative backend; construct PDO, Redis/Valkey, AMQP, SQS, and other process-bound resources inside its worker factory after fork. External Supervisor, systemd, Docker, or Kubernetes remains the preferred production supervisor when available.

The consumer verification guide provides runnable source, packed-consumer, FPM, and explicit Runwire forwarding probes.

Host Runwire context

A framework or intermediary library can pass its existing runtime, active request, and optional coroutine scope into Omnibus. Use null for $hostScope when the host has no coroutine scope:

$envelope = $bus->withRunwire(
    runtime: $hostRuntime,
    callback: static fn () => $bus->dispatch(new CreateInvoice($accountId)),
    request: $hostRequest,
    scope: $hostScope,
);

Consumer and Worker expose the same withRunwire() entry point. Forward the same instances through intermediary libraries; each binding is restored when the callback exits. The host owns request completion, workers, and event loops. A live coroutine scope and runtime capability enable cooperative waits; otherwise waits use the synchronous blocking fallback. Ordinary dispatch works without Runwire. See the integration guide for shared DBLayer bindings and the runnable forwarding example.

For a shared bus, the framework may register a provider on its existing binding:

$bus->runwireBinding()->setRunwireContextProvider([
    \Infocyph\CacheLayer\Integration\Runwire\RunwireIntegration::class,
    'current',
]);

The provider returns the current native RunwireExecutionContext or null. Ordinary dispatch() calls then borrow that context, check cancellation and restore the binding after each operation, including exceptions and overlapping Fibers. Explicit withRunwire() bindings take precedence. Installing this hook does not bind a runtime globally or start workers; the host owns its contexts.

Delivery semantics

Durable transports provide at-least-once delivery. Terminal failures are persisted before rejection, stale receipts cannot settle reclaimed work, and telemetry failures cannot change queue or handler outcomes. Handlers that produce durable side effects must remain idempotent.

Workflows use expiring dispatch claims and a durable handled reconciliation state. Queue and workflow settlement can be atomic when DBLayer adapters share one connection; cross-system compositions remain at-least-once.

Future integrations

Potential future releases may add optional NATS JetStream and Kafka transports without changing Omnibus's role as the application-level message bus:

NATS JetStream through the broker boundary, mapping durable pull consumers, ACK/NAK, delayed redelivery, and stable message IDs.
Kafka through a dedicated partition-aware transport with manual offset settlement, consumer-group rebalance handling, partition keys, and durable retry topics.
Additional brokers only where their native delivery guarantees can satisfy Omnibus's explicit settlement contract.
Expanded opt-in infrastructure tests for real MySQL, MariaDB, PostgreSQL, and SQL Server replica lag, reconnect, and failover behavior.

These integrations will remain optional and will not add broker clients to the core runtime dependencies. See future integrations for the intended boundaries and acceptance criteria.

Quality checks

composer ic:tests
composer ic:ci
composer benchmark
composer soak:consumer
composer soak:durable
composer soak:workflow

Security

Do not disclose suspected vulnerabilities in a public issue, discussion or pull request. Follow SECURITY.md and use GitHub private vulnerability reporting.

Omnibus is protected by PHPForge, which provides automated tests, static and taint analysis, dependency auditing, architecture checks and release-readiness gates. Automated controls do not replace responsible disclosure or manual review.

_{Made with ❤️ for the PHP community}
_{MIT Licensed}
Documentation • Security • Code of Conduct • Contributing
🗂️ Bug • Feature • Documentation • Question • CI failure
🔀 General • Bug fix • Feature • Refactor • Performance • Security & reliability • Documentation • Maintenance

infocyph / omnibus

Package info

Security

Statistics

README

Install

Highlights

Quick start

Handler middleware

Host Runwire context

Delivery semantics

Future integrations

Quality checks

Security