Initial implementation of Centrifugo.

docker-compose.dev.yml

@@ -3,7 +3,8 @@ services:
     build:
       context: .
     ports:
-      - "127.0.0.1:3306:3306"
+      - "127.0.0.1:3306:3306" # MariaDB
+      - "127.0.0.1:6025:6025" # Centrifugo
     volumes:
       - $PWD/util/local_ssl/default.crt:/var/azuracast/acme/ssl.crt:ro
       - $PWD/util/local_ssl/default.key:/var/azuracast/acme/ssl.key:ro

frontend/vue/components/Common/NowPlaying.vue

@@ -8,9 +8,19 @@ export const nowPlayingProps = {
             type: String,
             required: true
         },
+        useSse: {
+            type: Boolean,
+            required: false,
+            default: false
+        },
+        sseUri: {
+            type: String,
+            required: false,
+            default: null
+        },
         initialNowPlaying: {
             type: Object,
-            default () {
+            default() {
                 return NowPlaying;
             }
         }
@@ -19,29 +29,52 @@ export const nowPlayingProps = {
 
 export default {
     mixins: [nowPlayingProps],
-    mounted () {
+    data() {
+        return {
+            'sse': null
+        };
+    },
+    mounted() {
         // Convert initial NP data from prop to data.
         this.setNowPlaying(this.initialNowPlaying);
 
         setTimeout(this.checkNowPlaying, 5000);
     },
     methods: {
-        checkNowPlaying () {
-            this.axios.get(this.nowPlayingUri, {
-                headers: {
-                    'Cache-Control': 'no-cache',
-                    'Pragma': 'no-cache',
-                    'Expires': '0',
-                }
-            }).then((response) => {
-                this.setNowPlaying(response.data);
+        checkNowPlaying() {
+            if (this.useSse) {
+                this.sse = new EventSource(this.sseUri);
 
-                setTimeout(this.checkNowPlaying, (!document.hidden) ? 15000 : 30000);
-            }).catch((error) => {
-                setTimeout(this.checkNowPlaying, (!document.hidden) ? 30000 : 120000);
-            });
+                this.sse.onopen = (e) => {
+                    console.log(e);
+                };
+
+                this.sse.onmessage = (e) => {
+                    console.log(e);
+
+                    const data = JSON.parse(e.data);
+                    const np = data.np || null;
+                    if (np) {
+                        this.setNowPlaying(np);
+                    }
+                };
+            } else {
+                this.axios.get(this.nowPlayingUri, {
+                    headers: {
+                        'Cache-Control': 'no-cache',
+                        'Pragma': 'no-cache',
+                        'Expires': '0',
+                    }
+                }).then((response) => {
+                    this.setNowPlaying(response.data);
+
+                    setTimeout(this.checkNowPlaying, (!document.hidden) ? 15000 : 30000);
+                }).catch((error) => {
+                    setTimeout(this.checkNowPlaying, (!document.hidden) ? 30000 : 120000);
+                });
+            }
         },
-        setNowPlaying (np_new) {
+        setNowPlaying(np_new) {
             // Update the browser metadata for browsers that support it (i.e. Mobile Chrome)
             if ('mediaSession' in navigator) {
                 navigator.mediaSession.metadata = new MediaMetadata({

frontend/vue/components/Public/Player.vue

@@ -217,13 +217,7 @@ import IsMounted from "~/components/Common/IsMounted";
 export const radioPlayerProps = {
     ...nowPlayingProps,
     props: {
-        nowPlayingUri: {
-            type: String,
-            required: true
-        },
-        initialNowPlaying: {
-            type: Object
-        },
+        ...nowPlayingProps.props,
         showHls: {
             type: Boolean,
             default: true

src/Controller/Frontend/PublicPages/PlayerAction.php

@@ -8,6 +8,7 @@ use App\Entity;
 use App\Exception\StationNotFoundException;
 use App\Http\Response;
 use App\Http\ServerRequest;
+use App\Service\Centrifugo;
 use Psr\Http\Message\ResponseInterface;
 
 final class PlayerAction
@@ -15,6 +16,7 @@ final class PlayerAction
     public function __construct(
         private readonly Entity\ApiGenerator\NowPlayingApiGenerator $npApiGenerator,
         private readonly Entity\Repository\CustomFieldRepository $customFieldRepo,
+        private readonly Centrifugo $centrifugo
     ) {
     }
 
@@ -56,6 +58,11 @@ final class PlayerAction
                 : $router->named('api:nowplaying:index', ['station_id' => $station->getShortName()]),
         ];
 
+        if ($customization->useStaticNowPlaying() && $this->centrifugo->isSupported()) {
+            $props['useSse'] = true;
+            $props['sseUri'] = $this->centrifugo->getSseUrl($station);
+        }
+
         // Render embedded player.
         if (!empty($embed)) {
             $pageClasses = [];

src/Service/Centrifugo.php

@@ -0,0 +1,62 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Service;
+
+use App\Entity\Station;
+use App\Environment;
+use GuzzleHttp\Client;
+
+final class Centrifugo
+{
+    public function __construct(
+        private readonly Environment $environment,
+        private readonly Client $client,
+    ) {
+    }
+
+    public function isSupported(): bool
+    {
+        return $this->environment->isDocker();
+    }
+
+    public function publishToStation(Station $station, mixed $message): void
+    {
+        $this->client->post(
+            'http://localhost:6025/api',
+            [
+                'json' => [
+                    'method' => 'publish',
+                    'params' => [
+                        'channel' => $this->getChannelName($station),
+                        'data' => [
+                            'np' => $message,
+                        ],
+                    ],
+                ],
+            ]
+        );
+    }
+
+    public function getSseUrl(Station $station): string
+    {
+        return '/api/live/nowplaying/sse?' . http_build_query(
+                [
+                    'cf_connect' => json_encode(
+                        [
+                            'subs' => [
+                                $this->getChannelName($station) => []
+                            ],
+                        ],
+                        JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT
+                    ),
+                ]
+            );
+    }
+
+    public function getChannelName(Station $station): string
+    {
+        return 'station:'.$station->getShortName();
+    }
+}

src/Service/ServiceControl.php

@@ -17,7 +17,8 @@ final class ServiceControl
 {
     public function __construct(
         private readonly SupervisorInterface $supervisor,
-        private readonly Environment $environment
+        private readonly Environment $environment,
+        private readonly Centrifugo $centrifugo
     ) {
     }
 
@@ -82,8 +83,13 @@ final class ServiceControl
             'php-nowplaying' => __('Now Playing manager service'),
             'php-worker' => __('PHP queue processing worker'),
             'sftpgo' => __('SFTP service'),
+            'centrifugo' => __('Live Now Playing updates'),
         ];
 
+        if (!$this->centrifugo->isSupported()) {
+            unset($services['centrifugo']);
+        }
+
         if (!$this->environment->useLocalDatabase()) {
             unset($services['mariadb']);
         }

src/Webhook/Dispatcher.php

@@ -124,6 +124,8 @@ final class Dispatcher
             $np->resolveUrls($this->router->getBaseUrl());
             $np->cache = 'event';
 
+            $this->localHandler->dispatch($station, $np);
+
             $connectorObj = $this->connectors->getConnector($webhook->getType());
             $connectorObj->dispatch($station, $webhook, $np, [
                 WebhookTriggers::SongChanged->value,

src/Webhook/LocalWebhookHandler.php

@@ -6,6 +6,7 @@ namespace App\Webhook;
 
 use App\Entity;
 use App\Environment;
+use App\Service\Centrifugo;
 use Monolog\Logger;
 use Symfony\Component\Filesystem\Filesystem;
 
@@ -18,6 +19,7 @@ final class LocalWebhookHandler
     public function __construct(
         private readonly Logger $logger,
         private readonly Environment $environment,
+        private readonly Centrifugo $centrifugo
     ) {
     }
 
@@ -57,13 +59,15 @@ final class LocalWebhookHandler
         // Write JSON file to disk so nginx can serve it without calling the PHP stack at all.
         $this->logger->debug('Writing static nowplaying text file...');
 
-        $staticArtPath = $staticNpDir . '/' . $station->getShortName() . '.webp';
-
-
         $staticNpPath = $staticNpDir . '/' . $station->getShortName() . '.json';
         $fsUtils->dumpFile(
             $staticNpPath,
             json_encode($np, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) ?: ''
         );
+
+        // Publish to websocket library
+        if ($this->centrifugo->isSupported()) {
+            $this->centrifugo->publishToStation($station, $np);
+        }
     }
 }

util/docker/web/centrifugo/config.json

@@ -0,0 +1,23 @@
+{
+  "allow_anonymous_connect_without_token": true,
+  "api_insecure": true,
+  "admin": true,
+  "admin_insecure": true,
+  "port": 6020,
+  "internal_port": 6025,
+  "websocket_disable": true,
+  "uni_websocket": true,
+  "uni_sse": true,
+  "uni_http_stream": true,
+  "namespaces": [
+    {
+      "name": "station",
+      "history_size": 1,
+      "history_ttl": "30s",
+      "allow_subscribe_for_client": true,
+      "allow_subscribe_for_anonymous": true,
+      "allow_history_for_client": true,
+      "allow_history_for_anonymous": true
+    }
+  ]
+}

util/docker/web/nginx/azuracast.conf.tmpl

@@ -6,6 +6,10 @@ upstream php-fpm-www {
     server unix:/var/run/php-fpm-www.sock;
 }
 
+upstream centrifugo {
+    server 127.0.0.1:6020;
+}
+
 # Internal connection handler for PubSub and internal API calls
 server {
     listen 127.0.0.1:6010;
@@ -79,6 +83,12 @@ server {
         try_files $uri =404;
     }
 
+    # Websocket/SSE Now Playing Updates
+    location ~ ^/api/live/nowplaying/(\w+)$ {
+        include proxy_params;
+        proxy_pass http://centrifugo/connection/uni_$1?$args;
+    }
+
     # Default clean URL routing
     location / {
         try_files $uri @clean_url;

util/docker/web/service.full/centrifugo.conf

@@ -0,0 +1,19 @@
+[program:centrifugo]
+command=centrifugo -c /var/azuracast/centrifugo/config.json
+dir=/var/azuracast/centrifugo
+user=azuracast
+priority=700
+numprocs=1
+autostart=true
+autorestart=true
+
+stopasgroup=true
+killasgroup=true
+
+stdout_logfile=/var/azuracast/www_tmp/service_centrifugo.log
+stdout_logfile_maxbytes=5MB
+stdout_logfile_backups=5
+redirect_stderr=true
+
+stdout_events_enabled = true
+stderr_events_enabled = true

util/docker/web/setup/centrifugo.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+set -e
+set -x
+
+# Per-architecture installs
+ARCHITECTURE=amd64
+if [[ "$(uname -m)" = "aarch64" ]]; then
+    ARCHITECTURE=arm64
+fi
+
+cd /tmp
+wget -O centrifugo.tar.gz "https://github.com/centrifugal/centrifugo/releases/download/v4.0.4/centrifugo_4.0.4_linux_${ARCHITECTURE}.tar.gz"
+
+tar -xzvf centrifugo.tar.gz
+
+cp centrifugo /usr/local/bin/centrifugo
+chmod a+x /usr/local/bin/centrifugo
+
+mkdir -p /var/azuracast/centrifugo
+cp /bd_build/web/centrifugo/config.json /var/azuracast/centrifugo/config.json
+