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