AnyEvent.pm

package Mojo::IOWatcher::AnyEvent;
use Mojo::Base 'Mojo::IOWatcher';

use AE;
use Scalar::Util 'weaken';

$ENV{MOJO_IOWATCHER} ||= 'Mojo::IOWatcher::AnyEvent';

my $SINGLETON;

sub DESTROY { undef $SINGLETON }

# We have to fall back to Mojo::IOWatcher, since AnyEvent is unique
sub new { $SINGLETON++ ? Mojo::IOWatcher->new : shift->SUPER::new }

sub change {
  my ($self, $handle, $read, $write) = @_;

  my $fd = fileno $handle;
  my $h  = $self->{handles}->{$fd};
  delete $h->{read};
  delete $h->{write};
  weaken $self if $read || $write;
  $h->{read} = AE::io($fd, 0, sub { $self->_io($fd, 0) }) if $read;
  $h->{write} = AE::io($fd, 1, sub { $self->_io($fd, 1) }) if $write;

  return $self;
}

sub drop_handle {
  my ($self, $handle) = @_;
  delete $self->{handles}->{fileno $handle};
  return $self;
}

sub recurring { shift->_timer(shift, 1, @_) }

sub start {
  my $w = shift->{w} = AE::cv;
  $w->recv;
}

sub stop { shift->{w}->send }

sub timer { shift->_timer(shift, 0, @_) }

sub _io {
  my ($self, $fd, $write) = @_;
  my $h = $self->{handles}->{$fd};
  $self->_sandbox('Read', $h->{on_readable}, $h->{handle}) unless $write;
  $self->_sandbox('Write', $h->{on_writable}, $h->{handle}) if $write;
}

sub _timer {
  my $self      = shift;
  my $after     = shift || '0.0001';
  my $recurring = shift;
  my $cb        = shift;

  my $id = $self->SUPER::_timer($cb);
  weaken $self;
  $self->{timers}->{$id}->{watcher} = AE::timer(
    $after,
    $recurring ? $after : 0,
    sub {
      my $w = shift;
      $self->_sandbox("Timer $id", $self->{timers}->{$id}->{cb}, $id);
      delete $self->{timers}->{$id} unless $recurring;
    }
  );

  return $id;
}

1;