While Mercure is the recommended option for most cases, implementing a custom Server-sent Event (SSE) backend relying on Symfony components is still a good alternative (considering you're building an app with a few users only)

Example using today’s code:

public function __invoke(): StreamedResponse
{
    $response = new StreamedResponse(function () {
        foreach ($this->watchJobsInProgress() as $job) {
            echo "type: jobs\n";
            echo "data: ".$job->toJson()."\n\n";

            StreamedResponse::closeOutputBuffers(0, true);
            flush();

            if (connection_aborted()) {
                break;
            }

            sleep(1);
        }
    });
    $response->headers->set('Content-Type', 'text/event-stream');
    $response->headers->set('Cache-Control', 'no-cache');
    $response->headers->set('Connection', 'keep-alive');
    $response->headers->set('X-Accel-Buffering', 'no');

    return $response;
}

So, this PR simplifies that implementation by adding two new artifacts:

✨ New ServerEvent DTO class

Spec: https://html.spec.whatwg.org/multipage/server-sent-events.html

The ServerEvent class represents individual events streamed from the server to clients. It includes fields like data, type, id, retry, and comment, so it can cover a variety of use cases.

One key feature is the getIterator() method, which automatically concat all these fields into the proper SSE format. This saves us from having to manually format the events, making the process a lot smoother. Plus, it supports iterable data, meaning it can handle multi-line messages or more complex data structures all in one event.

✨ New EventStreamResponse class

• Simplified Headers Setup

  return new EventStreamResponse(/* ... */);

  This class automatically sets the required headers (Content-Type, Cache-Control, and Connection), ensuring that every response is properly configured for event streaming without additional code.

• Generator-based Event Streaming with Auto Flush

  return new EventStreamResponse(function (): \Generator {    
      yield new ServerEvent(time(), type: 'ping');
  }); 

  The callback function can now yield events directly, automatically handling serialization and output buffering as each new event occurs.

• Event Streaming with Sending Control

  return new EventStreamResponse(function (EventStreamResponse $response) {    
      $response->sendEvent(new ServerEvent('...'));
      // do something in between ...
      $response->sendEvent(new ServerEvent('...'));
  }); 

  The callback function receives a new argument—the response instance—allowing you to manually sendEvent().

  This method is especially useful when yield is not possible due to scope limitations. In the example below, the callback handles the event inside an event stream response, listening for Redis messages and directly triggering the sendEvent() method:

  return new EventStreamResponse(function (EventStreamResponse $response) {  
      $redis = new \Redis();  
      $redis->connect('127.0.0.1');  
      $redis->subscribe(['message'], function (/* ... */, string $message) use ($response) {  
          $response->sendEvent(new ServerEvent($message));  
      });  
  }); 

• Retry Event Fallback

  return new EventStreamResponse(/* ... */, retry: 1000);

  ServerEvent allows setting individual retry intervals, which take precedence over the default retry configured in EventStreamResponse. If not specified at the event level, the global retry setting applies.

This is the final, optimized version after the proposed improvements:

public function __invoke(): EventStreamResponse
{
    return new EventStreamResponse(function () {
        foreach ($this->watchJobsInProgress() as $job) {
            yield new ServerEvent($job->toJson(), type: 'jobs');

            sleep(1);
        }
    });
}

Cheers!