|
@@ -2,16 +2,21 @@
|
|
|
/**
|
|
/**
|
|
|
* Statum frontend demo backend.
|
|
* Statum frontend demo backend.
|
|
|
*
|
|
*
|
|
|
- * Run from this directory with PHP's built-in server:
|
|
|
|
|
|
|
+ * Run from this directory with PHP's built-in server. Because the realtime
|
|
|
|
|
+ * channel is a long-lived Server-Sent Events connection, you MUST start the
|
|
|
|
|
+ * server with several workers so an open SSE stream does not block other
|
|
|
|
|
+ * requests:
|
|
|
*
|
|
*
|
|
|
- * php -S localhost:8000 server.php
|
|
|
|
|
|
|
+ * PHP_CLI_SERVER_WORKERS=5 php -S localhost:8000 server.php
|
|
|
*
|
|
*
|
|
|
- * Then open http://localhost:8000/ in your browser.
|
|
|
|
|
|
|
+ * Then open http://localhost:8000/ in your browser (and a second tab to see
|
|
|
|
|
+ * realtime pushes).
|
|
|
*
|
|
*
|
|
|
* This is a DEMO only: there is no real signing or encryption. Snapshots are
|
|
* This is a DEMO only: there is no real signing or encryption. Snapshots are
|
|
|
* wrapped in trivial frames, and per-item action context is carried in the
|
|
* wrapped in trivial frames, and per-item action context is carried in the
|
|
|
* action's `private` blob as plain base64. State lives in the PHP session so
|
|
* action's `private` blob as plain base64. State lives in the PHP session so
|
|
|
- * that actions are stateful across requests.
|
|
|
|
|
|
|
+ * that actions are stateful across requests. Realtime subscriptions are shared
|
|
|
|
|
+ * between workers through small files in the system temp dir.
|
|
|
*/
|
|
*/
|
|
|
|
|
|
|
|
session_start();
|
|
session_start();
|
|
@@ -70,7 +75,7 @@ function key_for(string $type): string
|
|
|
* pre-flight via /_statum/slots before each request. Load the page, wait ~10s,
|
|
* pre-flight via /_statum/slots before each request. Load the page, wait ~10s,
|
|
|
* then trigger an action to see the pre-flight request.
|
|
* then trigger an action to see the pre-flight request.
|
|
|
*/
|
|
*/
|
|
|
-function make_snapshot(string $type, string $scope, array $public, array $opts = []): array
|
|
|
|
|
|
|
+function make_snapshot(string $type, string $scope, array $public, array $opts = [], ?string $key = null): array
|
|
|
{
|
|
{
|
|
|
$now = time();
|
|
$now = time();
|
|
|
$asAt = gmdate('Y-m-d\TH:i:s\Z', $now);
|
|
$asAt = gmdate('Y-m-d\TH:i:s\Z', $now);
|
|
@@ -83,7 +88,7 @@ function make_snapshot(string $type, string $scope, array $public, array $opts =
|
|
|
}
|
|
}
|
|
|
return [
|
|
return [
|
|
|
'as_at' => $asAt,
|
|
'as_at' => $asAt,
|
|
|
- 'slot' => ['key' => key_for($type), 'scope' => $scope, 'type' => $type],
|
|
|
|
|
|
|
+ 'slot' => ['key' => $key ?? key_for($type), 'scope' => $scope, 'type' => $type],
|
|
|
'transmit_after' => $transmit,
|
|
'transmit_after' => $transmit,
|
|
|
'invalid_after' => gmdate('Y-m-d\TH:i:s\Z', $now + 600),
|
|
'invalid_after' => gmdate('Y-m-d\TH:i:s\Z', $now + 600),
|
|
|
'private' => $opts['private'] ?? '',
|
|
'private' => $opts['private'] ?? '',
|
|
@@ -219,6 +224,98 @@ function all_directives(): array
|
|
|
return $dirs;
|
|
return $dirs;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// ---------------------------------------------------------------------------
|
|
|
|
|
+// Realtime channel helpers (demo pub/sub via the filesystem)
|
|
|
|
|
+// ---------------------------------------------------------------------------
|
|
|
|
|
+
|
|
|
|
|
+/** A `set` directive wrapping an already-framed snapshot. */
|
|
|
|
|
+function set_dir_frame(array $frame): array
|
|
|
|
|
+{
|
|
|
|
|
+ return ['type' => 'set', 'snapshot_frame' => $frame];
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Directory holding per-channel subscription and event-log files. */
|
|
|
|
|
+function channel_dir(): string
|
|
|
|
|
+{
|
|
|
|
|
+ $dir = rtrim(sys_get_temp_dir(), '/') . '/statum_demo';
|
|
|
|
|
+ if (!is_dir($dir)) @mkdir($dir, 0777, true);
|
|
|
|
|
+ return $dir;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+function channel_subs_path(string $id): string { return channel_dir() . '/ch_' . $id . '.subs'; }
|
|
|
|
|
+function channel_events_path(string $id): string { return channel_dir() . '/ch_' . $id . '.events'; }
|
|
|
|
|
+function gen_channel_id(): string { return bin2hex(random_bytes(8)); }
|
|
|
|
|
+
|
|
|
|
|
+/** Read a channel's subscribed slot keys (empty array if unknown). */
|
|
|
|
|
+function read_channel_subs(string $id): array
|
|
|
|
|
+{
|
|
|
|
|
+ $f = channel_subs_path($id);
|
|
|
|
|
+ if (!is_file($f)) return [];
|
|
|
|
|
+ $arr = json_decode((string)@file_get_contents($f), true);
|
|
|
|
|
+ return is_array($arr) ? array_values($arr) : [];
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Overwrite a channel's subscribed slot keys. */
|
|
|
|
|
+function set_channel_subs(string $id, array $subs): void
|
|
|
|
|
+{
|
|
|
|
|
+ @file_put_contents(channel_subs_path($id), json_encode(array_values(array_unique($subs))), LOCK_EX);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Append one event to a channel's event log (a JSON line). */
|
|
|
|
|
+function append_channel_event(string $id, string $type, $data): void
|
|
|
|
|
+{
|
|
|
|
|
+ $line = json_encode(['t' => $type, 'd' => $data]) . "\n";
|
|
|
|
|
+ @file_put_contents(channel_events_path($id), $line, FILE_APPEND | LOCK_EX);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Channel ids currently subscribed to a given slot key. */
|
|
|
|
|
+function channels_subscribed_to(string $key): array
|
|
|
|
|
+{
|
|
|
|
|
+ $ids = [];
|
|
|
|
|
+ foreach ((array)glob(channel_dir() . '/ch_*.subs') as $f) {
|
|
|
|
|
+ if (!is_file($f)) continue;
|
|
|
|
|
+ $id = substr(basename($f), 3, -5); // ch_{ID}.subs
|
|
|
|
|
+ if (in_array($key, read_channel_subs($id), true)) $ids[] = $id;
|
|
|
|
|
+ }
|
|
|
|
|
+ return $ids;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Push a `set` frame to every channel subscribed to a slot key. */
|
|
|
|
|
+function fanout_set(string $key, array $frame): void
|
|
|
|
|
+{
|
|
|
|
|
+ foreach (channels_subscribed_to($key) as $id) append_channel_event($id, 'set', $frame);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Shared "live" broadcast value, stored on disk so all workers agree. */
|
|
|
|
|
+function live_path(): string { return channel_dir() . '/live.json'; }
|
|
|
|
|
+
|
|
|
|
|
+function live_read(): array
|
|
|
|
|
+{
|
|
|
|
|
+ if (!is_file(live_path())) return ['text' => 'No live messages yet.', 'at' => ''];
|
|
|
|
|
+ $d = json_decode((string)@file_get_contents(live_path()), true);
|
|
|
|
|
+ return is_array($d) ? $d : ['text' => 'No live messages yet.', 'at' => ''];
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Snapshot for the shared `live` slot (transient scope, fixed key `k_live`). */
|
|
|
|
|
+function snap_live(): array
|
|
|
|
|
+{
|
|
|
|
|
+ $d = live_read();
|
|
|
|
|
+ return make_snapshot('live', 'transient', [
|
|
|
|
|
+ 'text' => $d['text'],
|
|
|
|
|
+ 'at' => $d['at'],
|
|
|
|
|
+ 'sendAction' => ['uri' => '/api/live/send', 'method' => 'POST'],
|
|
|
|
|
+ 'muteAction' => ['uri' => '/api/live/mute', 'method' => 'POST'],
|
|
|
|
|
+ 'resumeAction' => ['uri' => '/api/live/resume', 'method' => 'POST'],
|
|
|
|
|
+ ], [], 'k_live');
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** The initial frame to push to a channel when it subscribes to a key. */
|
|
|
|
|
+function initial_frame_for_key(string $key): ?array
|
|
|
|
|
+{
|
|
|
|
|
+ if ($key === 'k_live') return frame(snap_live());
|
|
|
|
|
+ return null;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// ---------------------------------------------------------------------------
|
|
// ---------------------------------------------------------------------------
|
|
|
// Router
|
|
// Router
|
|
|
// ---------------------------------------------------------------------------
|
|
// ---------------------------------------------------------------------------
|
|
@@ -233,6 +330,12 @@ if ($path === '/statum.js') {
|
|
|
exit;
|
|
exit;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+if ($path === '/_statum/worker.js') {
|
|
|
|
|
+ header('Content-Type: application/javascript');
|
|
|
|
|
+ readfile(__DIR__ . '/../js/statum-worker.js');
|
|
|
|
|
+ exit;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// The page itself.
|
|
// The page itself.
|
|
|
if ($path === '/' || $path === '/index.html') {
|
|
if ($path === '/' || $path === '/index.html') {
|
|
|
header('Content-Type: text/html; charset=utf-8');
|
|
header('Content-Type: text/html; charset=utf-8');
|
|
@@ -254,7 +357,10 @@ if ($path === '/api/checkout-done') {
|
|
|
// --- Statum endpoints -------------------------------------------------------
|
|
// --- Statum endpoints -------------------------------------------------------
|
|
|
|
|
|
|
|
if ($path === '/_statum/entrypoint') {
|
|
if ($path === '/_statum/entrypoint') {
|
|
|
- statum_respond(all_directives());
|
|
|
|
|
|
|
+ $dirs = all_directives();
|
|
|
|
|
+ $dirs[] = set_dir(snap_live()); // current broadcast + actions
|
|
|
|
|
+ $dirs[] = ['type' => 'subscribe', 'key' => 'k_live']; // begin realtime delivery
|
|
|
|
|
+ statum_respond($dirs);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
if ($path === '/_statum/slots') {
|
|
if ($path === '/_statum/slots') {
|
|
@@ -274,6 +380,86 @@ if ($path === '/_statum/slots') {
|
|
|
statum_respond($dirs);
|
|
statum_respond($dirs);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// --- Realtime channel -------------------------------------------------------
|
|
|
|
|
+
|
|
|
|
|
+// Open the SSE stream. The first event carries the channel id; later events
|
|
|
|
|
+// are `set`/`clear` for the slots this channel has subscribed to (via PATCH).
|
|
|
|
|
+if ($path === '/_statum/channel' && $method === 'GET') {
|
|
|
|
|
+ // Long-lived connection: release the session lock immediately so other
|
|
|
|
|
+ // requests (served by other workers) for the same session are not blocked.
|
|
|
|
|
+ session_write_close();
|
|
|
|
|
+ @set_time_limit(0);
|
|
|
|
|
+ @ini_set('zlib.output_compression', '0');
|
|
|
|
|
+ while (ob_get_level() > 0) { @ob_end_flush(); }
|
|
|
|
|
+ header('Content-Type: text/event-stream');
|
|
|
|
|
+ header('Cache-Control: no-cache');
|
|
|
|
|
+ header('X-Accel-Buffering: no');
|
|
|
|
|
+
|
|
|
|
|
+ $id = gen_channel_id();
|
|
|
|
|
+ set_channel_subs($id, []);
|
|
|
|
|
+ @file_put_contents(channel_events_path($id), '');
|
|
|
|
|
+
|
|
|
|
|
+ echo "event: channel\ndata: {$id}\n\n";
|
|
|
|
|
+ flush();
|
|
|
|
|
+
|
|
|
|
|
+ $offset = 0;
|
|
|
|
|
+ $eventsFile = channel_events_path($id);
|
|
|
|
|
+ while (!connection_aborted()) {
|
|
|
|
|
+ clearstatcache(true, $eventsFile);
|
|
|
|
|
+ $size = is_file($eventsFile) ? filesize($eventsFile) : 0;
|
|
|
|
|
+ if ($size > $offset) {
|
|
|
|
|
+ $fp = @fopen($eventsFile, 'rb');
|
|
|
|
|
+ if ($fp) {
|
|
|
|
|
+ fseek($fp, $offset);
|
|
|
|
|
+ $chunk = (string)stream_get_contents($fp);
|
|
|
|
|
+ $offset = (int)ftell($fp);
|
|
|
|
|
+ fclose($fp);
|
|
|
|
|
+ foreach (preg_split('/\r?\n/', trim($chunk)) as $line) {
|
|
|
|
|
+ if ($line === '') continue;
|
|
|
|
|
+ $ev = json_decode($line, true);
|
|
|
|
|
+ if (!is_array($ev)) continue;
|
|
|
|
|
+ if (($ev['t'] ?? '') === 'set') {
|
|
|
|
|
+ echo 'event: set' . "\ndata: " . json_encode($ev['d']) . "\n\n";
|
|
|
|
|
+ } elseif (($ev['t'] ?? '') === 'clear') {
|
|
|
|
|
+ echo 'event: clear' . "\ndata: " . $ev['d'] . "\n\n";
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ flush();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ echo ": ping\n\n"; // heartbeat keeps the stream alive
|
|
|
|
|
+ flush();
|
|
|
|
|
+ usleep(1000000);
|
|
|
|
|
+ }
|
|
|
|
|
+ // Client went away: drop this channel's files.
|
|
|
|
|
+ @unlink(channel_subs_path($id));
|
|
|
|
|
+ @unlink($eventsFile);
|
|
|
|
|
+ exit;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// Manage a channel's subscriptions: {subscribe:[keys], unsubscribe:[keys]}.
|
|
|
|
|
+if (preg_match('#^/_statum/channel/([^/]+)$#', $path, $m) && $method === 'PATCH') {
|
|
|
|
|
+ $id = $m[1];
|
|
|
|
|
+ $body = json_decode((string)file_get_contents('php://input'), true);
|
|
|
|
|
+ if (!is_array($body)) $body = [];
|
|
|
|
|
+ $add = $body['subscribe'] ?? [];
|
|
|
|
|
+ $rem = $body['unsubscribe'] ?? [];
|
|
|
|
|
+ $subs = read_channel_subs($id);
|
|
|
|
|
+ foreach ($add as $k) {
|
|
|
|
|
+ if (!in_array($k, $subs, true)) $subs[] = $k;
|
|
|
|
|
+ }
|
|
|
|
|
+ $subs = array_values(array_diff($subs, $rem));
|
|
|
|
|
+ set_channel_subs($id, $subs);
|
|
|
|
|
+ // Push the current value of any newly-subscribed slot so a fresh client
|
|
|
|
|
+ // sees the latest state right away.
|
|
|
|
|
+ foreach ($add as $k) {
|
|
|
|
|
+ $fr = initial_frame_for_key((string)$k);
|
|
|
|
|
+ if ($fr !== null) append_channel_event($id, 'set', $fr);
|
|
|
|
|
+ }
|
|
|
|
|
+ http_response_code(204);
|
|
|
|
|
+ exit;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// --- Actions ----------------------------------------------------------------
|
|
// --- Actions ----------------------------------------------------------------
|
|
|
|
|
|
|
|
if ($path === '/api/counter/inc' && $method === 'POST') {
|
|
if ($path === '/api/counter/inc' && $method === 'POST') {
|
|
@@ -343,6 +529,31 @@ if ($path === '/api/reset-all' && $method === 'POST') {
|
|
|
statum_respond($dirs);
|
|
statum_respond($dirs);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// --- Live channel actions ---------------------------------------------------
|
|
|
|
|
+
|
|
|
|
|
+if ($path === '/api/live/send' && $method === 'POST') {
|
|
|
|
|
+ $msg = trim((string)($_POST['msg'] ?? ''));
|
|
|
|
|
+ if ($msg === '') $msg = '(empty message)';
|
|
|
|
|
+ @file_put_contents(live_path(), json_encode(['text' => $msg, 'at' => date('H:i:s')]), LOCK_EX);
|
|
|
|
|
+ $fr = frame(snap_live());
|
|
|
|
|
+ fanout_set('k_live', $fr); // push to every subscribed tab via the channel
|
|
|
|
|
+ statum_respond([
|
|
|
|
|
+ set_dir_frame($fr), // immediate feedback (idempotent vs the push)
|
|
|
|
|
+ ['type' => 'notify', 'kind' => 'success', 'message' => 'Broadcast to all tabs'],
|
|
|
|
|
+ ]);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// Unsubscribing the only slot closes the realtime connection for this client.
|
|
|
|
|
+if ($path === '/api/live/mute' && $method === 'POST') {
|
|
|
|
|
+ statum_respond([['type' => 'unsubscribe', 'key' => 'k_live']]);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// Re-subscribing reopens the connection; the PATCH initial-push re-sends the
|
|
|
|
|
+// current broadcast.
|
|
|
|
|
+if ($path === '/api/live/resume' && $method === 'POST') {
|
|
|
|
|
+ statum_respond([['type' => 'subscribe', 'key' => 'k_live']]);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// Fallback 404.
|
|
// Fallback 404.
|
|
|
http_response_code(404);
|
|
http_response_code(404);
|
|
|
header('Content-Type: text/plain');
|
|
header('Content-Type: text/plain');
|