Aktualizace stavu vysilani pres websockets

lib/PiTube.pm

@@ -30,6 +30,7 @@ sub startup {
     ;
     $pg->migrations->from_file($self->home . '/sql/migrations.sql');
     $pg->migrations->migrate();
+    $self->helper( pg => sub { return $pg; } );
 
     # Spojeni s databazi
     my $schema = PiTube::Schema->connect($cfg->{database});
@@ -84,6 +85,8 @@ sub startup {
 #    $r->put('/streams/:id')->over( is => 'administrator'  )->to('Stream#update');
 #    $r->delete('/streams/:id')->over( is => 'administrator'  )->to('Stream#delete');
 
+    $r->websocket('/ws/streams')->to('Stream#ws');
+
     $r->get('/oauth2')->to('OAuth2#callback');
     $r->get('/logout')->to('OAuth2#do_logout');
     $r->post('/callback/rtmp')->to('Nginx#callback_rtmp');

lib/PiTube/Controller/Nginx.pm

@@ -7,7 +7,6 @@ use constant HLS => qr/hls\/([a-z0-9\-]+)(_\w+)?(\/\w+)?\.(m3u8|ts)$/i;
 sub callback_rtmp {
     my $c = shift;
     my $pubsub = Mojo::Pg::PubSub->new(pg => $c->pg);
-    $pubsub = $pubsub->json('messages');
 
     if ( $c->param('call') =~ /publish/ ) {
         my $name = $c->param('name');
@@ -38,10 +37,10 @@ sub callback_rtmp {
             is_live          => ( $c->param('call') =~ /done/ ) ? 'f':'t',
         });
 
-        $pubsub->notify( messages => {
+        $pubsub->json('streams')->notify( streams => {
             call      => $c->param('call'),
+            stream_id => $stream->id,
         });
-
     }
 
     $c->render( status => 204, text => '' );

lib/PiTube/Controller/Stream.pm

 package PiTube::Controller::Stream;
+
+use feature 'signatures';
 use Mojo::Base 'Mojolicious::Controller';
 use Mojo::UserAgent;
+use Mojo::Pg::PubSub;
 
 use constant CONTENT_TYPE => {
     m3u8 => 'application/vnd.apple.mpegurl',
@@ -175,6 +178,23 @@ sub recording {
     $c->render( json => { recording => $stream->recording  } );
 }
 
+sub ws {
+    my $c = shift;
+
+    $c->inactivity_timeout(300);
+
+    my $pubsub = Mojo::Pg::PubSub->new(pg => $c->pg);
+
+    $pubsub->listen(streams => sub($pubsub, $payload) {
+        $c->send($payload);
+    });
+
+    $c->on(finish => sub ($c, $code, $reason = undef) {
+        $pubsub->unlisten('streams');
+        $c->app->log->debug("WebSocket closed with status $code");
+    });
+}
+
 1;
 
 __END__

nginx.conf

@@ -90,6 +90,14 @@ http {
             expires 24h;
         }
 
+        location /ws {
+            proxy_pass http://pitube:3000/ws;
+            proxy_http_version 1.1;
+            proxy_set_header Upgrade $http_upgrade;
+            proxy_set_header Connection "Upgrade";
+            proxy_set_header Host $host;
+        }
+
         location / {
             proxy_pass http://pitube:3000/;
             add_header Access-Control-Allow-Origin *;

pi_tube.conf

@@ -38,4 +38,5 @@
   },
   matomo_id         => $ENV{MATOMO_ID} // 0,
   base_url          => $ENV{BASE_URL},
+  ws_url            => $ENV{WS_URL},
 };