lib/App/MPD/Feeder.pm

   1 use v5.28;
   2 use utf8;
   3 use Object::Pad;
   4 class App::MPD::Feeder;
   5
   6 use App::MPD::Feeder::DB;
   7 use App::MPD::Feeder::Options;
   8 use App::MPD::Feeder::WorkQueue;
   9 use DBD::Pg;
  10 use DBI;
  11 use Getopt::Long;
  12 use IO::Async::Signal;
  13 use IO::Async::Timer::Periodic;
  14 use Log::Any qw($log);
  15 use Net::Async::MPD;
  16 use Object::Pad;
  17 use Syntax::Keyword::Try;
  18
  19 has $cfg_file :reader;
  20 has $opt :reader;
  21 has $db :reader;
  22 has $db_needs_update :writer = 1;
  23 has $mpd :reader;
  24 has $mpd_connected = 0;
  25 has $playlist_needs_filling = 1;
  26 has $quit_requested = 0;
  27 has $reload_requested = 0;
  28 has $idler;
  29 has $last_mpd_comm;
  30 has $reconnect_delay = 5;
  31
  32 use constant DEFAULT_CONFIG_FILE => '/etc/mpd-feeder/mpd-feeder.conf';
  33
  34 ADJUST {
  35     Getopt::Long::Configure('pass_through');
  36     Getopt::Long::GetOptions( 'cfg|config=s' => \$cfg_file );
  37     Getopt::Long::Configure('no_pass_through');
  38
  39     $cfg_file //= DEFAULT_CONFIG_FILE if -e DEFAULT_CONFIG_FILE;
  40
  41     $self->configure;
  42
  43     $db_needs_update = 0 if $opt->skip_db_update;
  44 }
  45
  46 method configure {
  47     my $new_opt = App::MPD::Feeder::Options->new;
  48
  49     $new_opt->parse_config_file($cfg_file) if $cfg_file;
  50
  51     $new_opt->parse_command_line;
  52
  53     Log::Any::Adapter->set( Stderr => log_level => $new_opt->log_level );
  54
  55     $opt = $new_opt;
  56
  57     $reconnect_delay = $opt->initial_reconnect_delay;
  58
  59     $db = App::MPD::Feeder::DB->new( opt => $opt );
  60 }
  61
  62 method init_mpd {
  63     return if $mpd;
  64
  65     my %conn = ( auto_connect => 0 );
  66     $conn{host} = $opt->mpd_host if $opt->mpd_host;
  67     $conn{port} = $opt->mpd_port if $opt->mpd_port;
  68
  69     $mpd = Net::Async::MPD->new(%conn);
  70
  71     $mpd->on(
  72         close => sub {
  73             $mpd->loop->stop('disconnected');
  74             $mpd_connected = 0;
  75             $log->warn("Connection to MPD lost");
  76         }
  77     );
  78     $mpd->on(
  79         playlist => sub {
  80             $playlist_needs_filling = 1;
  81         }
  82     );
  83     $mpd->on(
  84         database => sub {
  85             $db_needs_update = 1;
  86         }
  87     );
  88
  89     my $int_signal_handler = sub {
  90         state $signal_count = 0;
  91         $signal_count++;
  92
  93         if ( $signal_count > 1 ) {
  94             $log->warn("Another signal received (#$signal_count)");
  95             $log->warn("Exiting abruptly");
  96             exit 2;
  97         }
  98
  99         $log->debug("Signal received. Stopping loop");
 100         $quit_requested = 1;
 101         $mpd->loop->stop('quit');
 102         $self->break_idle;
 103     };
 104
 105     for (qw(TERM INT)) {
 106         $mpd->loop->add(
 107             IO::Async::Signal->new(
 108                 name       => $_,
 109                 on_receipt => $int_signal_handler,
 110             )
 111         );
 112     }
 113
 114     $mpd->loop->add(
 115         IO::Async::Signal->new(
 116             name       => 'HUP',
 117             on_receipt => sub {
 118                 $log->debug("SIGHUP received. Scheduling reload");
 119                 $reload_requested = 1;
 120                 $mpd->loop->stop('reload');
 121                 $self->break_idle;
 122             },
 123         )
 124     );
 125
 126     $mpd->loop->add(
 127         IO::Async::Signal->new(
 128             name       => 'USR1',
 129             on_receipt => sub {
 130                 $log->debug(
 131                     "SIGUSR1 received. Dumping configuration to STDERR");
 132                 my $old = select \*STDERR;
 133                 try {
 134                     $opt->dump;
 135                 }
 136                 finally {
 137                     select $old;
 138                 }
 139             },
 140         )
 141     );
 142 }
 143
 144 method connect_db {
 145     $db->connect($opt);
 146 }
 147
 148 method update_db( $force = undef ) {
 149     if ( !$db_needs_update and !$force ) {
 150         $log->debug("Skipping DB update");
 151         return;
 152     }
 153
 154     $log->info('Updating song database');
 155
 156     my $rows = $mpd->send('listallinfo')->get;
 157
 158     $log->trace('got all songs from MPD');
 159
 160     $db->start_update;
 161     try {
 162         my $song_count;
 163
 164         foreach my $entry (@$rows) {
 165             next unless exists $entry->{file};
 166
 167             $self->db->store_song( $entry->{file},
 168                 $entry->{AlbumArtist} // $entry->{Artist},
 169                 $entry->{Album} );
 170
 171             $song_count++;
 172         }
 173
 174         my ($total_songs, $total_artists, $total_albums,
 175             $new_songs,   $new_artists,   $new_albums
 176         ) = $self->db->finish_update;
 177
 178         $log->info(
 179             "Updated data about $song_count songs (including $new_songs new), "
 180                 . "$total_artists artists (including $new_artists new) "
 181
 182                 . "and $total_albums albums (including $new_albums new)"
 183         );
 184
 185         $db_needs_update = 0;
 186     }
 187     catch {
 188         my $err = $@;
 189         $self->db->cancel_update;
 190         die $err;
 191     }
 192 }
 193
 194 method queue_songs( $num = undef ) {
 195     if ( !defined $num ) {
 196         return unless $playlist_needs_filling;
 197
 198         $log->trace("Requesting playlist");
 199         my $present = $mpd->send('playlist')->get // [];
 200         $present = scalar(@$present);
 201
 202         $log->notice( "Playlist contains $present songs. Wanted: "
 203                 . $opt->target_queue_length );
 204         if ( $present < $opt->target_queue_length ) {
 205             $self->queue_songs( $opt->target_queue_length - $present );
 206         }
 207         else {
 208             $playlist_needs_filling = 0;
 209         }
 210
 211         return;
 212     }
 213
 214     my @list = $self->db->find_suitable_songs($num);
 215
 216     die "Found no suitable songs" unless @list;
 217
 218     if ( @list < $num ) {
 219         $log->warn(
 220             sprintf(
 221                 'Found only %d suitable songs instead of %d',
 222                 scalar(@list), $num
 223             )
 224         );
 225     }
 226
 227     $log->info("About to add $num songs to the playlist");
 228
 229     my @paths;
 230     for my $song (@list) {
 231         my $path = $song->{song};
 232         $path =~ s/"/\\"/g;
 233         push @paths, $path;
 234     }
 235
 236     $log->debug( "Adding " . join( ', ', map {"«$_»"} @paths ) );
 237
 238     # MPD needs raw bytes
 239     utf8::encode($_) for @paths;
 240     my @commands;
 241     for (@paths) {
 242         push @commands, [ add => "\"$_\"" ];
 243     }
 244     my $f = $mpd->send( \@commands );
 245     $f->on_fail( sub { die @_ } );
 246     $f->on_done(
 247         sub {
 248             $self->db->note_song_qeued($_) for @list;
 249             $playlist_needs_filling = 0;
 250         }
 251     );
 252     $f->get;
 253 }
 254
 255 method reexec {
 256     $log->notice("disconnecting and re-starting");
 257     $db->disconnect;
 258     undef $mpd;
 259
 260     my @exec = ( $0, '--config', $self->cfg_file, '--skip-db-update' );
 261     if ( $log->is_trace ) {
 262         $log->trace( 'exec ' . join( ' ', map { /\s/ ? "'$_'" : $_ } @exec ) );
 263     }
 264     exec(@exec);
 265 }
 266
 267 method break_idle {
 268     if ( $idler && !$idler->is_ready ) {
 269         $log->trace("hand-sending 'noidle'");
 270         undef $idler;
 271         $mpd->{mpd_handle}->write("noidle\n");
 272     }
 273     else {
 274         $log->trace("no idler found");
 275     }
 276 }
 277
 278 method sleep_before_reconnection {
 279     $self->debug( "Waiting for "
 280             . duration_exact($reconnect_delay)
 281             . " before re-connecting" );
 282
 283     $mpd->loop->add(
 284         IO::Async::Timer::Countdown->new(
 285             delay     => $reconnect_delay,
 286             on_expire => sub { $mpd->loop->stop },
 287         )->start
 288     );
 289
 290     $reconnect_delay = $reconnect_delay * 1.5;
 291     $reconnect_delay = 120 if $reconnect_delay > 120;
 292     $mpd->loop->run;
 293 }
 294
 295 method pulse {
 296     unless ($mpd_connected) {
 297         $mpd->connect->then(
 298             sub {
 299                 $mpd_connected          = 1;
 300                 $playlist_needs_filling = 1;
 301                 $reconnect_delay        = $opt->initial_reconnect_delay;
 302             }
 303         )->get;
 304
 305         $mpd->loop->later( sub { $self->pulse } );
 306         return;
 307     }
 308
 309     if ($db_needs_update) {
 310         $self->update_db;
 311         $mpd->loop->later( sub { $self->pulse } );
 312         return;
 313     }
 314
 315     if ($playlist_needs_filling) {
 316         $self->queue_songs;
 317         $mpd->loop->later( sub { $self->pulse } );
 318         return;
 319     }
 320
 321     $log->debug("Waiting idle. PID=$$");
 322     $last_mpd_comm = time;
 323     $idler         = $mpd->send("idle database playlist");
 324     my $result = $idler->get;
 325     undef $idler;
 326
 327     if ( ref $result and $result->{changed} ) {
 328         my $changed = $result->{changed};
 329         $changed = [$changed] unless ref $changed;
 330
 331         $mpd->emit($_) for @$changed;
 332     }
 333
 334     $log->trace('got out of idle');
 335     $mpd->loop->stop;
 336 }
 337
 338 method run_loop {
 339     $self->connect_db;
 340
 341     $self->init_mpd;
 342
 343     $mpd->loop->add(
 344         IO::Async::Timer::Periodic->new(
 345             interval => 60,
 346             on_tick  => sub {
 347                 if (!$mpd_connected) {
 348                     $log->trace("Not connected to MPD. Skipping alive check.");
 349                     return;
 350                 }
 351
 352                 if ( time - $last_mpd_comm > 300 ) {
 353
 354                     $log->trace(
 355                         "no active MPD communication for more that 5 minutes");
 356                     $log->debug("forcing alive check");
 357                     $self->break_idle;
 358                 }
 359                 else {
 360                     $log->trace(
 361                         "contacted MPD less than 5 minutes ago. skipping alive check"
 362                     );
 363                 }
 364             },
 365         )->start
 366     );
 367
 368     for ( ;; ) {
 369         $mpd->loop->later( sub { $self->pulse } );
 370
 371         $log->trace('About to run the loop');
 372
 373         $mpd->loop->run;
 374
 375         if ( $quit_requested ) {
 376             $log->trace("about to quit");
 377             undef $mpd;
 378             $db->disconnect;
 379             last;
 380         }
 381         elsif ( $reload_requested ) {
 382             $self->reexec;
 383             die "Not reached";
 384         }
 385         elsif ( !$mpd_connected ) {
 386             $self->sleep_before_reconnection;
 387             next;
 388         }
 389     }
 390 }
 391
 392 1;