ServerSentEvents is a wrapper class that provides a fluent interface for streaming real-time data to clients using Server Sent Events (SSE) with configurable conditions, timeouts, connection handling, and execution callbacks.

ServerSentEvents.php

<?php

namespace App\Support\Libraries\ServerSentEvents;

class ServerSentEvents
{

    protected mixed $condition = null;
    protected mixed $onExecuteCallback = null;
    protected mixed $onConnectionAborted = null;
    protected mixed $onFinishedCallback = null;
    protected mixed $onTimeoutCallback = null;
    protected ?int $timeoutSeconds = null;
    protected int $sleepSeconds = 1;


    public function __construct()
    {
        header("X-Accel-Buffering: no");
        header("Content-Type: text/event-stream");
        header("Cache-Control: no-cache");
    }

    public static function make()
    {
        return new static();
    }

    public function send($eventName,iterable $data)
    {
        echo  "event: $eventName" . "\n";
        echo  "data: " . json_encode([
                ...$data,
                ...($this->timeoutSeconds ? ['timeout' => $this->timeoutSeconds] : [])
            ], JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES) . "\n";
        echo "\n";

        return $this;
    }


    protected function flush()
    {
        if (ob_get_contents()) {
            ob_end_flush();
        }
        flush();

        return $this;
    }

    public function onConnectionAborted(callable $callback)
    {
        $this->onConnectionAborted = $callback;
        return $this;
    }

    public function loopWhen($condition)
    {
        $this->condition = $condition;
        return $this;
    }

    public function onExecute(callable $callback)
    {
        $this->onExecuteCallback = $callback;
        return $this;
    }

    public function onFinished(callable $callback)
    {
        $this->onFinishedCallback = $callback;
        return $this;
    }

    public function onTimeout(callable $callback)
    {
        $this->onTimeoutCallback = $callback;
        return $this;
    }

    public function executeEvery(int $seconds)
    {
        $this->sleepSeconds = $seconds;
        return $this;
    }

    public function timeoutFor(?int $seconds)
    {
        $this->timeoutSeconds = $seconds;
        return $this;
    }

    public function sendOnce($eventName, $data)
    {
        $this->send($eventName, $data);
        value($this->onFinishedCallback, $this);
        exit;
    }

    public function start()
    {
        $startTime = now();
        while(value($this->condition)){
            value($this->onExecuteCallback, $this);
            if(connection_aborted()){
                value($this->onConnectionAborted, $this);
                break;
            }
            $this->flush();
            if($this->timeoutSeconds && (now()->diffInSeconds($startTime) >= $this->timeoutSeconds)){
                value($this->onTimeoutCallback, $this);
                $this->flush();
                exit;
            }
            sleep(max(1, $this->sleepSeconds));
        }
        value($this->onFinishedCallback, $this);
        $this->flush();

        exit;
    }

}

ServerSentEvents Class

A Laravel utility class for implementing Server-Sent Events (SSE) with a fluent, chainable API. It handles real-time data streaming to web clients with built-in connection management, timeout controls, and customizable execution loops. The class automatically sets proper SSE headers and provides callbacks for various connection states.

// Real-time order status updates
Route::get('/orders/{order}/status-stream', function (Order $order) {
    return ServerSentEvents::make()
        ->timeoutFor(300) // 5 minutes timeout
        ->executeEvery(2) // Check every 2 seconds
        ->loopWhen(fn() => $order->fresh()->status !== 'completed')
        ->onExecute(function($sse) use ($order) {
            $order = $order->fresh();
            $sse->send('order-update', [
                'status' => $order->status,
                'progress' => $order->progress_percentage,
                'updated_at' => $order->updated_at->toISOString()
            ]);
        })
        ->onConnectionAborted(fn() => Log::info("Client disconnected from order {$order->id}"))
        ->onTimeout(fn($sse) => $sse->send('timeout', ['message' => 'Stream timeout reached']))
        ->onFinished(fn() => Log::info("Order {$order->id} completed"))
        ->start();
});

// Frontend JavaScript
const eventSource = new EventSource('/orders/123/status-stream');

eventSource.addEventListener('order-update', function(event) {
    const data = JSON.parse(event.data);
    updateOrderUI(data.status, data.progress);
});

eventSource.addEventListener('timeout', function(event) {
    console.log('Stream timed out');
    eventSource.close();
});