lib/Object/Remote/MiniLoop.pm

   1 package Object::Remote::MiniLoop;
   2
   3 use IO::Select;
   4 use Moo;
   5
   6 # this is ro because we only actually set it using local in sub run
   7
   8 has is_running => (is => 'ro', clearer => 'stop');
   9
  10 has _read_watches => (is => 'ro', default => sub { {} });
  11 has _read_select => (is => 'ro', default => sub { IO::Select->new });
  12
  13 sub pass_watches_to {
  14   my ($self, $new_loop) = @_;
  15   foreach my $fh ($self->_read_select->handles) {
  16     $new_loop->watch_io(
  17       handle => $fh,
  18       on_read_ready => $self->_read_watches->{$fh}
  19     );
  20   }
  21 }
  22
  23 sub watch_io {
  24   my ($self, %watch) = @_;
  25   my $fh = $watch{handle};
  26   if (my $cb = $watch{on_read_ready}) {
  27     $self->_read_select->add($fh);
  28     $self->_read_watches->{$fh} = $cb;
  29   }
  30 }
  31
  32 sub unwatch_io {
  33   my ($self, %watch) = @_;
  34   my $fh = $watch{handle};
  35   if ($watch{on_read_ready}) {
  36     $self->_read_select->remove($fh);
  37     delete $self->_read_watches->{$fh};
  38   }
  39 }
  40
  41 sub loop_once {
  42   my ($self) = @_;
  43   my $read = $self->_read_watches;
  44   my ($readable) = IO::Select->select($self->_read_select, undef, undef, 0.5);
  45   # I would love to trap errors in the select call but IO::Select doesn't
  46   # differentiate between an error and a timeout.
  47   #   -- no, love, mst.
  48   foreach my $fh (@$readable) {
  49     $read->{$fh}();
  50   }
  51 }
  52
  53 sub run {
  54   my ($self) = @_;
  55   local $self->{is_running} = 1;
  56   while ($self->is_running) {
  57     $self->loop_once;
  58   }
  59 }
  60
  61 1;