PHP code example of event-stream / laravel

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

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

    

event-stream / laravel example snippets

// app/Streams/JobProgressStream.php
namespace App\Streams;

use EventStream\EventStream;
use Illuminate\Http\Request;

class JobProgressStream extends EventStream
{
    // {jobId} is a placeholder; a key like "job.progress:abc-123"
    // makes $this->args->jobId === "abc-123".
    const streamPattern = 'job.progress:{jobId}';

    public function authorize(Request $request): bool
    {
        // Return true to allow the subscription, false for a 403.
        return $request->user() !== null;
    }

    // A typed publish method. `send()` returns the new Redis message id.
    public function sendProgress(int $percent): string
    {
        return $this->send('progress', ['percent' => $percent]);
    }
}

'streams' => [
    App\Streams\JobProgressStream::class,
],

// routes/web.php
use EventStream\Http\StreamSubscriptionController;

Route::get('/streams', [StreamSubscriptionController::class, 'stream'])
    ->middleware(['web', 'auth'])
    ->name('streams.subscribe');

use App\Streams\JobProgressStream;

JobProgressStream::make('abc-123')->sendProgress(50);

// Backend
namespace App\Streams;

use EventStream\StateStream;
use Illuminate\Http\Request;

class DocumentStateStream extends StateStream
{
    const streamPattern = 'document:{documentId}';

    public function authorize(Request $request): bool
    {
        return $request->user() !== null;
    }
}

// Mutate the graph anywhere — each call publishes a granular mutation:
$doc = DocumentStateStream::make('abc-123');
$doc->setState('title', 'Quarterly report');
$doc->setState('author.name', 'Ada');
$doc->arrayAppend('comments', ['body' => 'Looks good']);

const streamPattern = 'job.progress:{jobId}';          // one placeholder
const streamPattern = 'report:{reportId}:{format}';    // two placeholders
const streamPattern = 'system.notifications';          // no placeholders

$this->args->jobId; // "abc-123"

JobProgressStream::make('abc-123');           // job.progress:abc-123
ReportStream::make('rpt-99', 'pdf');          // report:rpt-99:pdf
SystemNotificationsStream::make();            // system.notifications

public function authorize(Request $request): bool
{
    return Job::query()
        ->where('id', $this->args->jobId)
        ->where('user_id', $request->user()?->id)
        ->exists();
}

public function sendCompleted(array $summary): string
{
    return $this->send('completed', ['summary' => $summary]);
}

public function initialCursor(): string
{
    return '0-0'; // replay from the beginning of the Redis stream
}

public function preludeMessage(): ?array
{
    return ['snapshot' => $this->currentSnapshot()];
}

use EventStream\Messages\DeliveredStreamMessage;

protected function registerInterceptors(): void
{
    // Runs on every message: enrich the payload.
    $this->registerInterceptor(function (DeliveredStreamMessage $message) {
        return $message->withData([
            ...$message->content->data,
            'serverTime' => now()->toIso8601String(),
        ]);
    });

    // Runs only on a specific type: hydrate, or suppress by returning null.
    $this->registerInterceptor('progress', function (DeliveredStreamMessage $message) {
        return $this->shouldHide($message) ? null : $message;
    });
}

$s = DocumentStateStream::make('abc-123');

$s->resetState(['title' => 'Draft', 'tags' => ['a', 'b']]); // replace whole graph
$s->setState('title', 'Final');                             // set one path
$s->setState(['title' => 'Final', 'author.name' => 'Ada']); // set many at once
$s->arrayAppend('comments', ['body' => 'Nice']);            // append to an array
$s->getState('tags');                                       // read back: ['a', 'b']
$s->getState();                                             // whole graph
$s->deleteState('title');                                   // remove a path
$s->deleteState();                                          // wipe the graph

protected function jsonPathTypes(): array
{
    return [
        'comments'   => 'array',  // always an array, even when empty
        'author'     => 'object',
        'sections.*' => 'array',
    ];
}

// Shown by getState() (and to clients) when nothing is stored yet.
protected function defaultState(): array
{
    return ['status' => 'pending'];
}

// Sliding-window lifetime of the state hash; refreshed on every mutation.
// Return null to keep state forever. Default: 3600 seconds.
protected function stateTtlSeconds(): ?int
{
    return 3600;
}

$connection = JobProgressStream::make('abc-123')->connect();

while (true) {
    $message = $connection->next(blockMs: 1000);
    if ($message === null) {
        continue; // timed out this interval — heartbeat point
    }
    // handle $message (a DeliveredStreamMessage)
}

$message = $connection->waitUntil(
    fn ($message) => $message->content->type === 'completed',
    deadlineMs: 30_000,
);
// $message is null if the deadline passed without a match.

class JobProgressStream extends EventStream
{
    const streamPattern = 'job.progress:{jobId}';

    // Keep this stream's messages for a day.
    public function messageExpirationSeconds(): ?int
    {
        return 86_400;
    }

    // ...
}

public function messageExpirationSeconds(): ?int
{
    return Tenant::find($this->args->tenantId)?->logExpiration;
}

Schedule::command('event-stream:prune')->everyFiveMinutes();

return [
    // EventStream subclasses the endpoint may resolve. Patterns must be unique.
    'streams' => [
        App\Streams\JobProgressStream::class,
    ],

    // ms the server blocks per read before sending a heartbeat comment.
    'heartbeat_interval_ms' => 10_000,

    // Max seconds a connection stays open before asking the browser to
    // reconnect (it reconnects automatically). Default: 55 minutes.
    'max_connection_seconds' => 55 * 60,

    // Seconds a message lives before `event-stream:prune` may delete it. Used by
    // streams whose messageExpirationSeconds() returns null. Null keeps messages
    // indefinitely. Default: 1 hour.
    'default_message_expiration_seconds' => 3600,
];

bash
php artisan vendor:publish --tag=event-stream-config

bash
php artisan event-stream:prune