File Coverage

blib/lib/AnyEvent/Onkyo.pm

Criterion	Covered	Total	%
statement	93	115	80.8
branch	7	16	43.7
condition			n/a
subroutine	23	29	79.3
pod	4	4	100.0
total	127	164	77.4

use strict;
use warnings;
package AnyEvent::Onkyo;
{
  $AnyEvent::Onkyo::VERSION = '1.130220';
}
use base 'Device::Onkyo';
use AnyEvent::Handle;
use AnyEvent::SerialPort;
use Carp qw/croak carp/;
use Sub::Name;
use Scalar::Util qw/weaken/;

use constant {
  DEBUG => $ENV{ANYEVENT_ONKYO_DEBUG},
};


# ABSTRACT: AnyEvent module for controlling Onkyo/Integra AV equipment


sub new {
  my ($pkg, %p) = @_;
  croak $pkg.'->new: callback parameter is required' unless ($p{callback});
  my $self = $pkg->SUPER::new(device => 'discover', %p);
  $self;
}


sub command {
  my $self = shift;
  my $cv = AnyEvent->condvar;
  my $weak_cv = $cv;
  weaken $weak_cv;
  $self->SUPER::command(@_, subname 'command_cb' => sub {
                          $weak_cv->send() if ($weak_cv);
                        });
  return $cv;
}

sub _open {
  my $self = shift;
  $self->SUPER::_open($self->_open_condvar);
  return 1;
}

sub _open_tcp_port {
  my ($self, $cv) = @_;
  my $dev = $self->{device};
  print STDERR "Opening $dev as tcp socket\n" if DEBUG;
  my ($host, $port) = split /:/, $dev, 2;
  $port = $self->{port} unless (defined $port);
  $self->{handle} =
    AnyEvent::Handle->new(connect => [$host, $port],
                          on_connect => subname('tcp_connect_cb' => sub {
                            my ($hdl, $h, $p) = @_;
                            warn ref $self, " connected to $h:$p\n" if DEBUG;
                            $cv->send();
                          }),
                          on_connect_error =>
                          subname('tcp_connect_error_cb' => sub {
                            my ($hdl, $msg) = @_;
                            my $err =
                              (ref $self).": Can't connect to $dev: $msg";
                            warn "Connect error: $err\n" if DEBUG;
                            $self->cleanup($err);
                            $cv->croak;
                          }));
  return $cv;
}

sub _open_serial_port {
  my ($self, $cv) = @_;
  $self->{handle} =
    AnyEvent::SerialPort->new(serial_port =>
                              [ $self->device,
                                [ baudrate => $self->baud ] ]);
  $cv->send();
  return $cv;
}

sub _handle_setup {
  my $self = shift;
  my $handle = $self->{handle};
  my $weak_self = $self;
  weaken $weak_self;

  $handle->on_error(subname('on_error' => sub {
                              my ($hdl, $fatal, $msg) = @_;
                              print STDERR $hdl.": error $msg\n" if DEBUG;
                              $hdl->destroy;
                              if ($fatal) {
                                $weak_self->cleanup($msg);
                              }
                            }));

  $handle->on_eof(subname('on_eof' => sub {
                            my ($hdl) = @_;
                            print STDERR $hdl.": eof\n" if DEBUG;
                            $weak_self->cleanup('connection closed');
                          }));

  $handle->on_read(subname 'on_read_cb' => sub {
    my ($hdl) = @_;
    $hdl->push_read(ref $self => $self,
                    subname 'push_read_cb' => sub {
                      $weak_self->{callback}->(@_);
                      $weak_self->_write_now();
                      return 1;
                    });
  });

  $self->{handle}->on_timeout($self->{on_timeout}) if ($self->{on_timeout});
  $self->{handle}->timeout($self->{timeout}) if ($self->{timeout});
  1;
}

sub DESTROY {
  $_[0]->cleanup;
}


sub cleanup {
  my ($self, $error) = @_;
  print STDERR $self."->cleanup\n" if DEBUG;
  $self->{handle}->destroy if ($self->{handle});
  delete $self->{handle};
}

sub _open_condvar {
  my $self = shift;
  print STDERR $self."->open_condvar\n" if DEBUG;
  my $cv = AnyEvent->condvar;
  my $weak_self = $self;
  weaken $weak_self;

  $cv->cb(subname 'open_cb' => sub {
            print STDERR "start cb ", $weak_self->{handle}, " @_\n" if DEBUG;
            $weak_self->_handle_setup();
            $weak_self->_write_now();
          });
  $weak_self->{_waiting} = ['fake for async open'];
  return $cv;
}

sub _real_write {
  my ($self, $str, $desc, $cb) = @_;
  print STDERR "Sending: ", $desc, "\n" if DEBUG;
  $self->{handle}->push_write($str);
}

sub _time_now {
  AnyEvent->now;
}


sub anyevent_read_type {
  my ($handle, $cb, $self) = @_;

  my $weak_self = $self;
  weaken $weak_self;

  subname 'anyevent_read_type_reader' => sub {
    my ($handle) = @_;
    my $rbuf = \$handle->{rbuf};
    while (1) { # read all message from the buffer
      print STDERR "Before: ", (unpack 'H*', $$rbuf||''), "\n" if DEBUG;
      my $res = $weak_self->read_one($rbuf);
      return unless ($res);
      print STDERR "After: ", (unpack 'H*', $$rbuf), "\n" if DEBUG;
      $res = $cb->($res);
    }
  }
}

1;

__END__