Differential D4803 Diff 13758 src/app/Http/Controllers/SSEController.php

Changeset View

Standalone View

src/app/Http/Controllers/SSEController.php

This file was added.

				<?php

				namespace App\Http\Controllers;

				use App\Backends\Events;
				use Illuminate\Routing\Controller;
				use Symfony\Component\HttpFoundation\StreamedResponse;

				class SSEController extends Controller
				{
				/**
				* Time interval to wait for an event. It's a maximum time spent between
				* ping events sent to the client to keep the connection alive.
				*
				* @const int
				*/
				protected const KEEPALIVE = 10;

				/** @var string Debug line prefix */
				protected $debugPrefix = '';


				/**
				* Send notifications (Server-Side Events) to the client.
				*
				* @param string $clientId Client identifier
				*
				* @return \Illuminate\Http\Response\|\Symfony\Component\HttpFoundation\StreamedResponse
				*/
				public function notify($clientId)
				{
				// TODO: Validate that the client id exists and belongs to the authenticated user

				// TODO: Per SSE spec. "a client can be told to stop reconnecting using the HTTP 204 No Content response code."
				// should we send 204 if client is unknown/unauthorized?

				// Octane has it's own max_execution_time limit. Which means when
				// octane.max_execution_time is reached the connection will be
				// dropped with "HTTP/1.1 408 Request Timeout" sent to the client.
				// Clients are expected to re-connect when connection is dropped anyway.
				// We could also send "please reconnect" event before we time out.
				set_time_limit(0);

				// Note: At the moment multiple connections from the same client are possible,
				// but that means an event will be sent only to a single connection.
				$cid = md5($clientId . microtime(true));
				$start = time();

				$this->debugPrefix = "PUSH [client={$clientId}, conn={$cid}]:";

				\Log::debug("{$this->debugPrefix} Connection started");

				$response = new StreamedResponse();

				$response->headers->set('Content-Type', 'text/event-stream');
				$response->headers->set('X-Accel-Buffering', 'no');
				$response->headers->set('Cache-Control', 'no-cache,private');

				$response->setCallback(function() use ($clientId, $start) {
				// Note: This first output makes the headers to be sent too
				$this->sendEvent('start', ['type' => 'start']);

				while (true) {
				// FIXME: connection_aborted() detects connection drop only after server
				mollekopfUnsubmitted Not Done Inline Actions I don't think there is a way to reliably detect dropped connections without sending some data (there's tcp keepalive, which sends data, but I'm not sure we can rely on that here). If we want to guarantee delivery of notifications there needs to be a protocol for acknowledgement from the client and retransmission otherwise. mollekopf: I don't think there is a way to reliably detect dropped connections without sending some data…
				// sent something to the client, i.e. not immediately. Is there a better solution?
				// We hope that at least the client can detect connection drops immediately
				// and reconnect, otherwise potential to drop some notifications is quite big
				if (connection_aborted()) {
				\Log::debug("{$this->debugPrefix} Connection aborted");
				return;
				}

				// Under Octane we exit before Octane does this for us. Octane does output
				// "HTTP/1 408 Request Timeout" text which causes KTOR client to throw an exception.
				$max = !empty($_SERVER['LARAVEL_OCTANE']) ? \config('octane.max_execution_time') : null;
				if ($max && ($start + $max - self::KEEPALIVE < time())) {
				return;
				}

				// Pull next message from the list (wait for it)
				if ($element = Events::pullMessage($clientId, self::KEEPALIVE)) {
				$this->sendEvent($element->type, $element->data);
				} else {
				$this->sendEvent('ping', ['type' => 'ping']);
				}
				}
				});

				return $response;
				}

				/**
				* Outputs a Server Side Event
				*/
				protected function sendEvent($event, $content): void
				{
				if (is_array($content)) {
				$content = json_encode($content);
				}

				echo 'event: ' . $event . PHP_EOL
				. 'data: ' . $content . PHP_EOL . PHP_EOL;

				// If not under Octane we need to force output flush
				if (empty($_SERVER['LARAVEL_OCTANE'])) {
				ob_flush();
				flush();
				}

				\Log::debug("{$this->debugPrefix} Sent $event");
				}
				}