File Coverage

blib/lib/Net/Async/MPD.pm

Criterion	Covered	Total	%
statement	43	169	25.4
branch	1	62	1.6
condition	0	12	0.0
subroutine	14	36	38.8
pod	7	8	87.5
total	65	287	22.6

package Net::Async::MPD;

use strict;
use warnings;

our $VERSION = '0.005';

use IO::Async::Loop;
use IO::Async::Stream;
use IO::Socket::IP;
use Future::Utils qw( repeat );
use Scalar::Util qw( weaken );
use Carp;

use namespace::clean;

use Moo;
use MooX::HandlesVia;
with 'Role::EventEmitter';

use Types::Standard qw(
  InstanceOf Int ArrayRef HashRef Str Maybe Bool CodeRef
);

use Log::Any;
my $log = Log::Any->get_logger( category => __PACKAGE__ );

has auto_connect => (
  is => 'ro',
  isa => Bool,
  default => 0,
);

has state => (
  is => 'rw',
  isa => Str,
  init_arg => undef,
  default => 'created',
  trigger => sub {
    $_[0]->emit( state => $_[0]->{state} );
  },
);

has loop => (
  is => 'ro',
  lazy => 1,
  default => sub { IO::Async::Loop->new },
);

has read_queue => (
  is => 'ro',
  isa => ArrayRef [CodeRef],
  lazy => 1,
  init_arg => undef,
  default => sub { [] },
  handles_via => 'Array',
  handles => {
    push_read    => 'push',
    pop_read     => 'pop',
    shift_read   => 'shift',
    unshift_read => 'unshift',
  },
);

has password => (
  is => 'ro',
  isa => Maybe[Str],
  lazy => 1,
);

has port => (
  is => 'ro',
  isa => Int,
  lazy => 1,
  default => sub { $ENV{MPD_PORT} // 6600 },
);

has host => (
  is => 'ro',
  isa => Str,
  lazy => 1,
  default => sub { $ENV{MPD_HOST} // 'localhost' },
);

sub version { $_[0]->{version} };

sub _parse_block {
  my $self = shift;

  return sub {
    my ( $handle, $buffref, $eof ) = @_;

    while ( $$buffref =~ s/^(.*)\n// ) {
      my $line = $1;

      if ($line =~ /\w/) {
        $log->tracef('< %s', $line);
        if ($line =~ /^OK/) {
          if ($line =~ /OK MPD (.*)/) {
            $log->trace('Connection established');
            $self->{version} = $1;

            $self->send( password => $self->password )
              if $self->password;

            $self->state( 'ready' );
          }
          else {
            pop @{$self->{mpd_buffer}} unless @{$self->{mpd_buffer}[-1]};
            $self->shift_read->( 1, $self->{mpd_buffer} );
            $self->{mpd_buffer} = [[]];
          }
        }
        elsif ($line =~ /^list_OK/) {
          push @{$self->{mpd_buffer}}, [];
        }
        elsif ($line =~ /^ACK/) {
          $self->shift_read->( 0, $line );
          $self->{mpd_buffer} = [[]];
          last;
        }
        else {
          push @{$self->{mpd_buffer}[-1]}, $line;
        }
      }
    }

    return 0;
  };
}

# Set up response parsers for each command
my $parsers = { none => sub { @_ } };
{
  my $item = sub {
    return {
      map {
        my ($key, $value) = split /: /, $_, 2;
        $key => $value;
      } @{$_[0]}
    };
  };

  my $flat_list = sub { [ map { (split /: /, $_, 2)[1] } @{$_[0]} ] };

  my $base_list = sub {
    my @main_keys = @{shift()};
    my @list_keys = @{shift()};
    my @lines     = @{shift()};

    my @return;
    my $item = {};

    foreach my $line (@lines) {
      my ($key, $value) = split /: /, $line, 2;

      if ( grep { /$key/ } @main_keys ) {
        push @return, $item if defined $item->{$key};
        $item = { $key => $value };
      }
      elsif ( grep { /$key/ } @list_keys ) {
        unless (defined $item->{$key}) {
          $item->{$key} = []
        }
        push @{$item->{$key}}, $value;
      }
      else {
        $item->{$key} = $value;
      }
    }
    push @return, $item if keys %{$item};

    return \@return;
  };

  my $grouped_list = sub {
    my @lines = @{shift()};

    # Our main category comes at the top of the list of lines
    my ($main) = split /:\s+/, $lines[0], 2;

    # Make a list of any other categories we might have
    my @categories;
    foreach (@lines) {
      my ($key) = split /:\s+/, $_, 2;
      if ($key eq $main) {
        last if @categories;
      }
      else {
        push @categories, $key;
      };
    }

    my $return = {};

    while (@lines) {
      # Generate a has with all the data returned for a single item
      # This will be over several lines if we are grouping
      my $item = do {
        my $set;
        my %missing_keys = map { $_ => 1 } $main, @categories;

        while ( my $line = shift @lines ) {
          my ($key, $value) = split /:\s+/, $line, 2;

          $set->{$key} = $value;
          delete $missing_keys{$key};

          last unless %missing_keys;
        }

        $set;
      };

      # Find or create the array of results we need to push the data into
      my $pointer = $return;
      for my $category (@categories) {
        my $value = $item->{$category} // '';
        $pointer = $pointer->{$category}{$value} //= {};
      }

      push @{ $pointer->{$main} //= [] }, delete $item->{$main};
    }

    return $return;
  };

  # Untested commands: what do they return?
  # consume
  # crossfade

  my $file_list = sub { $base_list->( [qw( directory file )], [], @_ ) };

  $parsers->{$_} = $flat_list foreach qw(
    commands notcommands channels tagtypes urlhandlers listplaylist
  );

  $parsers->{$_} = $item foreach qw(
    currentsong stats idle status addid update
    readcomments replay_gain_status rescan
  );

  $parsers->{$_} = $file_list foreach qw(
    find playlistinfo listallinfo search find playlistid playlistfind
    listfiles plchanges listplaylistinfo playlistsearch listfind
  );

  $parsers->{list} = $grouped_list;

  foreach (
      [ outputs        => [qw( outputid )],  [] ],
      [ plchangesposid => [qw( cpos )],      [] ],
      [ listplaylists  => [qw( playlist )],  [] ],
      [ listmounts     => [qw( mount )],     [] ],
      [ listneighbors  => [qw( neighbor )],  [] ],
      [ listall        => [qw( directory )], [qw( file )] ],
      [ readmessages   => [qw( channel )],   [qw( message )] ],
      [ lsinfo         => [qw( directory file playlist )], [] ],
      [ decoders       => [qw( plugin )], [qw( suffix mime_type )] ],
    ) {

    my ($cmd, $header, $list) = @{$_};
    $parsers->{$cmd} = sub { $base_list->( $header, $list, @_ ) };
  }

  $parsers->{playlist} = sub {
    my $lines = [ map { s/^\w*?://; $_ } @{shift()} ];
    $flat_list->( $lines, @_ )
  };

  $parsers->{count} = sub {
    my $lines = shift;
    my ($main) = split /:\s+/, $lines->[0], 2;
    $base_list->( [ $main ], [qw( )], $lines, @_ )
  };

  $parsers->{sticker} = sub {
    my $lines = shift;
    return {} unless scalar @{$lines};

    my $single = ($lines->[0] !~ /^file/);

    my $base = $base_list->( [qw( file )], [qw( sticker )], $lines, @_ );
    my $return = [ map {
      $_->{sticker} = { map { split(/=/, $_, 2) } @{$_->{sticker}} }; $_;
    } @{$base} ];

    return $single ? $return->[0] : $return;
  };
}

sub idle {
  my ($self, @subsystems) = @_;

  $self->{idle_future} = $self->loop->new_future;

  weaken $self;
  repeat {
    $self->send( idle => @subsystems, sub {
      $self->emit( shift->{changed} );
    });
  } until => sub { $self->{idle_future}->is_ready };

  return $self->{idle_future};
}

sub noidle {
  my ($self) = @_;

  my $idle = $self->{idle_future};
  return unless defined $idle;
  return if $idle->is_ready;

  $self->send( 'noidle' );
  $idle->done;

  return;
}

sub send {
  my $self = shift;

  my $opt  = ( ref $_[0] eq 'HASH' ) ? shift : {};
  my $cb = pop if ref $_[-1] eq 'CODE';
  my (@commands) = @_;

  croak 'Need commands to send'
    unless @commands;

  # Normalise input
  if (ref $commands[0] eq 'ARRAY') {
    @commands = map {
      ( ref $_ eq 'ARRAY' ) ? join( q{ }, @{$_} ) : $_;
    } @{$commands[0]};
  }
  else {
    @commands = join q{ }, @commands;
  }

  my $command = '';
  # Remove underscores from (most) command names
  @commands = map {
    my $args;
    ($command, $args) = split /\s/, $_, 2;
    $command =~ s/_//g unless $command =~ /^(replay_gain_|command_list)/;
    $args //= q{};
    $command . ($args ne q{} ? " $args" : q{});
  } @commands;

  # Ensure a command list if sending multiple commands
  if (scalar @commands > 1) {
    my $list = $opt->{list} // 1;
    my $list_start =
      'command_list' . ( $list ? '_ok' : q{} ) . '_begin';

    unshift @commands, $list_start
      unless $commands[0] =~ /^command_list/;
    push @commands, 'command_list_end'
      unless $commands[-1] =~ /^command_list/;
  }

  my $parser;

  if (defined $opt->{parser}) {
    my $input = delete $opt->{parser};
    $parser = (ref $input eq 'CODE') ? $input : $parsers->{$input};
    croak 'Not a code reference or recognised parser name'
      unless defined $parser;
  }
  else {
    $parser = sub {
      my ($input, $commands) = @_;

      my @result =  map {
        my $command;
        do { $command = shift @{$commands} }
          until !defined $command or $command !~ /^command_list/;

        my $sub = $parsers->{$command // ''} // $parsers->{none};

        $sub->( $input->[$_] );
      } 0 .. $#{$input};

      return @result
    };
  }

  my $future = $self->loop->new_future;
  $future->on_done( $cb ) if $cb;

  return $future->fail('No connection to MPD server' )
    unless $self->{mpd_handle};

  $self->push_read( sub {
    my ($success, $result) = @_;

    if ($success) {
      $future->done( $parser->(
        $result, [ map { my ($name) = split /\s/, $_, 2 } @commands ]
      ));
    }
    else {
      $self->emit( error => $result );
      $future->fail( $result );
    }
  });

  $log->tracef( '> %s', $_ ) foreach @commands;
  $self->{mpd_handle}->write( join("\n", @commands) . "\n" );

  return $future;
}

sub get { shift->send( @_ )->get }

sub until {
  my ($self, $name, $check, $cb) = @_;

  weaken $self;
  my $wrapper;
  $wrapper = sub {
    if ($check->(@_)) {
      $self->unsubscribe($name => $wrapper);
      $cb->(@_);
    }
  };
  $self->on($name => $wrapper);
  weaken $wrapper;

  return $wrapper;
}

sub BUILD {
  my ($self, $args) = @_;
  $self->connect->get if $self->auto_connect;
  $self->catch( sub {} );
  $self->{mpd_buffer} = [[]];
}

sub connect {
  my ($self) = @_;
  my $loop = $self->loop;
  my $connected = $loop->new_future;

  return $connected->done if $self->state eq 'ready';

  my $socket = IO::Socket::IP->new($self->host . ':' . $self->port)
    or return $connected->fail("MPD connect failed: $!");

  $log->debugf('Connecting to %s:%s', $self->host, $self->port);

  my $on_error = sub { $self->emit( error => shift ) };

  my $handle = IO::Async::Stream->new(
    handle => $socket,
    on_read_error  => sub { $on_error->('Read error: '  . shift) },
    on_write_error => sub { $on_error->('Write error: ' . shift) },
    on_read_eof    => sub { shift->close },
    on_closed => sub {
      $self->{mpd_handle} = undef;
      $self->emit( 'close' );
    },
    on_read => $self->_parse_block,
  );

  unless ($self->{mpd_handle}) {
    $self->{mpd_handle} = $handle;
    $loop->add( $handle );
  }

  $self->until( state =>
    sub { $_[1] eq 'ready' },
    sub { $connected->done; }
  );

  return $connected;
}

1;

__END__