PHP code example of hivesper / php-events

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

hivesper / php-events example snippets

use Tcds\Io\Raw\EventPublisher;
use Tcds\Io\Raw\EventSubscriberMap;
use Tcds\Io\Raw\Infrastructure\InMemoryEventStore;
use Tcds\Io\Raw\Infrastructure\JacksonSerializer;
use Tcds\Io\Raw\Infrastructure\SequentialEventProcessor;

// 1. Define a typed domain event
final readonly class OrderPlaced
{
    public function __construct(
        public int $orderId,
        public float $total,
    ) {}
}

// 2. Wire up the store, publisher, and processor
$store      = new InMemoryEventStore();
$publisher  = new EventPublisher($store, new JacksonSerializer());

$subscribers = new EventSubscriberMap();
$processor   = new SequentialEventProcessor($subscribers); // JacksonHydrator is the default

// 3. Register subscribers — type-hint the domain event class to receive it fully hydrated
$subscribers->subscribe('OrderPlaced', function (OrderPlaced $event): void {
    echo "Order placed: " . $event->orderId . PHP_EOL;
});

$subscribers->subscribe('OrderPlaced', function (OrderPlaced $event): void {
    echo "Sending confirmation email..." . PHP_EOL;
});

// 4. Publish a domain event
$publisher->publish(new OrderPlaced(orderId: 42, total: 99.99));

// 5. Process — both subscribers fire in registration order
$processor->process($store);

// Output:
// Order placed: 42
// Sending confirmation email...

// Create a brand-new event (generates a UUID v7 id, sets status → pending)
$event = RawEvent::create(
    name: 'payment.received',
    payload: ['amount' => 150, 'currency' => 'USD'],
    publishAt: CarbonImmutable::now(),
);

echo $event->id;        // uuid7 string
echo $event->name;      // "payment.received"
echo $event->status;    // RawEventStatus::pending
print_r($event->payload); // ['amount' => 150, 'currency' => 'USD']

// Reconstruct an event from persisted data (used internally by SqlEventStore)
$event = RawEvent::retrieve(
    id: $row['id'],
    name: $row['name'],
    status: RawEventStatus::from($row['status']),
    payload: json_decode($row['payload'], true),
    createdAt: new CarbonImmutable($row['created_at']),
    publishAt: new CarbonImmutable($row['publish_at']),
);

$publisher->publish(
    new SubscriptionReminder(userId: 7),
    publishAt: CarbonImmutable::now()->addDays(3),
);

interface EventSerializer
{
    public function serialize(object $event): SerializedEvent;
}

use Tcds\Io\Raw\Infrastructure\JacksonSerializer;

$publisher = new EventPublisher($store, new JacksonSerializer());

// OrderPlaced { orderId: 42, total: 99.99 }
// → SerializedEvent { name: 'OrderPlaced', payload: ['orderId' => 42, 'total' => 99.99] }

use Tcds\Io\Raw\EventSerializer;
use Tcds\Io\Raw\SerializedEvent;

final class AppEventSerializer implements EventSerializer
{
    public function serialize(object $event): SerializedEvent
    {
        return match (true) {
            $event instanceof OrderPlaced => new SerializedEvent(
                name: 'order.placed',
                payload: ['order_id' => $event->orderId, 'total' => $event->total],
            ),
            // ...
            default => throw new \InvalidArgumentException('Unknown event: ' . $event::class),
        };
    }
}

interface EventHydrator
{
    /** @param array<string, mixed> $payload */
    public function hydrate(string $name, array $payload, callable|string $subscriber): object;
}

// JacksonHydrator is the default — no explicit argument needed
$processor = new SequentialEventProcessor($subscribers);

// Typed subscriber receives a fully mapped OrderPlaced instance
$subscribers->subscribe('order.placed', function (OrderPlaced $event): void {
    echo $event->orderId; // int, not a stdClass property
});

// Untyped subscriber receives a generic stdClass
$subscribers->subscribe('order.placed', function (object $event): void {
    echo $event->orderId; // stdClass property
});

use Tcds\Io\Raw\EventHydrator;

final class AppEventHydrator implements EventHydrator
{
    public function hydrate(string $name, array $payload, callable|string $subscriber): object
    {
        return match ($name) {
            'order.placed' => new OrderPlaced(
                orderId: $payload['order_id'],
                total: $payload['total'],
            ),
            // ...
            default => throw new \InvalidArgumentException('Unknown event: ' . $name),
        };
    }
}

$subscribers->subscribe('order.placed', function (OrderPlaced $event): void {
    echo "Order placed: " . $event->orderId . PHP_EOL;
});

$processor = new SequentialEventProcessor(
    $subscribers,
    new DefaultListenerDispatcher(hydrator: new AppEventHydrator()),
);

$store = new InMemoryEventStore();

use Vesper\Tool\Event\Infrastructure\SqlEventStore;

$pdo   = new PDO('mysql:host=localhost;dbname=myapp', 'user', 'pass');
$store = new SqlEventStore($pdo);

use Vesper\Tool\Event\Infrastructure\Schema\MysqlEventStoreSchema;
use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema;

// run once at boot
MysqlEventStoreSchema::create($pdo);   // or SqliteEventStoreSchema::create($pdo)

use Carbon\CarbonInterval;

// In a separate cron task — e.g. every minute:
$recovered = $store->recoverStuckEvents(CarbonInterval::minutes(30));

$recovered = $redeliveryStore->recoverStuckRedeliveries(CarbonInterval::minutes(30));

$subscribers = new EventSubscriberMap();

// Closure
$subscribers->subscribe('order.cancelled', function (object $event): void {
    // ...
});

// First-class callable syntax
$subscribers->subscribe('order.shipped', $myService->onOrderShipped(...));

// Class name string — must implement __invoke(); instantiated by DefaultHandlerResolver
$subscribers->subscribe('order.placed', OrderPlacedHandler::class);

// Pre-populate via constructor (useful for DI containers)
$subscribers = new EventSubscriberMap([
    'order.placed' => [$listenerA, $listenerB],
    'payment.failed' => [$alertHandler],
]);

use Tcds\Io\Raw\EventSubscriberBuilder;

$subscribers = EventSubscriberBuilder::create()
    ->eventType('order.placed',     [OrderPlacedHandler::class, AuditLogger::class])
    ->eventType('payment.received', [PaymentHandler::class])
    ->listener(NotificationService::class, types: ['order.placed', 'order.shipped'])
    ->build();

$processor = new SequentialEventProcessor($subscribers);
$processor->process($store);

// In a console command / cron / queue worker:
$processor->process($store);

use Tcds\Io\Raw\EventProcessor;
use Tcds\Io\Raw\EventStore;

class MyProcessor implements EventProcessor
{
    public function process(EventStore $store): void
    {
        while ($event = $store->next()) {
            // your dispatch logic
            $store->markProcessed($event->id);
        }
    }
}

class MyEventStore implements EventStore
{
    public function add(RawEvent $event): void                          { /* ... */ }
    public function next(): ?RawEvent                                   { /* ... */ }
    public function markProcessed(RawEvent $event): void                { /* No-op when there's no persisted status to flip. */ }
    public function recoverStuckEvents(CarbonInterval $olderThan): int  { return 0; /* No rows to recover when there's no `processing` state. */ }
}

// Dev / CI — listener throws propagate out of process()
$processor = new SequentialEventProcessor($subscribers); // DefaultListenerDispatcher by default
$processor->process($store);

// Production — listener throws become redelivery rows; infra throws get logged
use Vesper\Tool\Event\Infrastructure\Dispatch\DefaultListenerDispatcher;
use Vesper\Tool\Event\Infrastructure\Dispatch\RedeliveringListenerDispatcher;
use Vesper\Tool\Event\Infrastructure\SequentialEventProcessor;
use Vesper\Tool\Event\Infrastructure\SilentEventProcessor;

$dispatcher = new RedeliveringListenerDispatcher(
    new DefaultListenerDispatcher(),
    $redeliveryStore,
);

$processor = new SilentEventProcessor(
    new SequentialEventProcessor($subscribers, $dispatcher),
    $logger,
);
$processor->process($store);

// Monolog
use Monolog\Logger;
use Monolog\Handler\StreamHandler;

$logger = new Logger('events');
$logger->pushHandler(new StreamHandler('php://stderr'));

$processor = new SilentEventProcessor(
    new SequentialEventProcessor($subscribers, $dispatcher),
    $logger,
);

// Laravel (already PSR-3 compatible)
$processor = new SilentEventProcessor(
    new SequentialEventProcessor($subscribers, $dispatcher),
    app('log'),
);

use Vesper\Tool\Event\Infrastructure\Dispatch\DefaultListenerDispatcher;
use Vesper\Tool\Event\Infrastructure\Dispatch\LoggingListenerDispatcher;
use Vesper\Tool\Event\Infrastructure\Dispatch\RedeliveringListenerDispatcher;

$logged = new LoggingListenerDispatcher(new DefaultListenerDispatcher(), $logger);

// Event path: log + persist failure as a redelivery row.
$eventDispatcher = new RedeliveringListenerDispatcher($logged, $redeliveryStore);

// Redelivery path: pass $logged directly — the redelivery processor handles reschedule/fail.
$redeliveryProcessor = new SequentialRedeliveryProcessor(
    $subscribers,
    $logged,
    new ExponentialBackoffRetryPolicy(),
);

use Vesper\Tool\Event\Retry\RetryPolicy;

interface RetryPolicy
{
    /** @return CarbonImmutable|null  null when no further retries should be made */
    public function nextRetryAt(int $previousAttempt): ?CarbonImmutable;
}

use Vesper\Tool\Event\Infrastructure\Retry\ExponentialBackoffRetryPolicy;

// Default delays — 100ms, 500ms, 1min, 5min.
$retryPolicy = new ExponentialBackoffRetryPolicy();

// Or roll your own delays:
$retryPolicy = new ExponentialBackoffRetryPolicy(delaysMs: [50, 250, 1_000, 30_000]);

use Vesper\Tool\Event\Infrastructure\Redelivery\SqlRedeliveryStore;

$store = new SqlRedeliveryStore($pdo);

use Vesper\Tool\Event\Infrastructure\Dispatch\DefaultListenerDispatcher;
use Vesper\Tool\Event\Infrastructure\Dispatch\RedeliveringListenerDispatcher;
use Vesper\Tool\Event\Infrastructure\Redelivery\SequentialRedeliveryProcessor;
use Vesper\Tool\Event\Infrastructure\Redelivery\SilentRedeliveryProcessor;
use Vesper\Tool\Event\Infrastructure\Redelivery\SqlRedeliveryStore;
use Vesper\Tool\Event\Infrastructure\Retry\ExponentialBackoffRetryPolicy;
use Vesper\Tool\Event\Infrastructure\SequentialEventProcessor;
use Vesper\Tool\Event\Infrastructure\SilentEventProcessor;

$default = new DefaultListenerDispatcher(
    ignoredExceptions: [
        UserNotFoundException::class,
        InvalidPayloadException::class,
    ],
);

$redeliveryStore = new SqlRedeliveryStore($pdo);

// Event flow: failures route into $redeliveryStore via the dispatcher decorator;
// anything else that escapes the batch is logged by SilentEventProcessor.
$eventProcessor = new SilentEventProcessor(
    new SequentialEventProcessor(
        $subscribers,
        new RedeliveringListenerDispatcher($default, $redeliveryStore),
    ),
    $logger,
);

// Redelivery flow: plain dispatcher (no Redelivering wrap — that would double-schedule).
// The retry policy decides whether each failed attempt reschedules or marks failed.
$redeliveryProcessor = new SilentRedeliveryProcessor(
    new SequentialRedeliveryProcessor(
        $subscribers,
        $default,
        new ExponentialBackoffRetryPolicy(),
    ),
    $logger,
);

// In one cron task / queue worker:
$eventProcessor->process($eventStore);

// In another cron task / queue worker:
$redeliveryProcessor->process($redeliveryStore);

RawEventStatus::pending     // event is waiting to be processed
RawEventStatus::processing  // event has been claimed by a worker; dispatch in flight
RawEventStatus::processed   // event was dispatched to every listener (per-listener outcomes live in event_outbox_redelivery)

bash
composer