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