view release on metacpan or  search on metacpan

lib/AnyEvent/ITM.pm  view on Meta::CPAN

AnyEvent::Handle::register_read_type(itm => sub {
  my ( $self, $cb ) = @_;
  sub {
    if (defined $_[0]{rbuf}) {
      my $first = substr($_[0]{rbuf},0,1);
      my $len = length($_[0]{rbuf});
      my $f = ord($first);
      my $header = itm_header($first);
      if ($header) {
        my $size = $header->{size} ? $header->{size} : 0;
        my $payload = substr($_[0]{rbuf},1,$size);
        if (defined $payload && length($payload) == $size) {
          my $itm = itm_parse($header,$size ? ($payload) : ());
          $_[0]{rbuf} = substr($_[0]{rbuf},$size + 1);
          $cb->( $_[0], $itm );
          return 1;          
        }
        return 0;
      } else {
        croak sprintf("unknown packet type");
      }
    }
    return 0;
  };
});

sub _ts {
  my @t = localtime;
  return sprintf "%04d-%02d-%02d %02d:%02d:%02d",
    $t[5]+1900, $t[4]+1, $t[3], $t[2], $t[1], $t[0];
}

sub _print_ts {
  my ( $label, $line ) = @_;
  chomp $line;
  if ($label eq '2>') {
    print STDERR _ts()." $label $line\n";
  } else {
    print STDOUT _ts()." $label $line\n";
  }
};

sub handle {
  my ( $class, $file, $payload_sub, $cv ) = @_;

  my $has_cv = defined $cv ? 1 : 0;

  $cv = AE::cv unless $has_cv;

  # Choose flags so open won't block
  my $flags;
  if (-p $file) {
    $flags = O_RDWR | O_NONBLOCK;
  } else {
    $flags = O_RDONLY | O_NONBLOCK;
    $flags |= O_NOCTTY if -c $file;
  }

  sysopen(my $fh, $file, $flags) or die "sysopen $file: $!";
  binmode($fh, ':raw');

  my $handle = AnyEvent::Handle->new(
    fh => $fh,
    on_error => sub {
      my ( $handle, $fatal, $message ) = @_;
      $handle->destroy;
      $cv->send("$fatal: $message");
    },
    on_eof => sub {
      my ( $handle ) = @_;
      $handle->destroy;
      $cv->send("EOF");
    },
    on_read => sub {
      my $handle = shift;
      $handle->push_read( itm => $payload_sub );
    },
  );

  $cv->recv unless $has_cv;

  return $handle;
}

sub _run_cmd {
  my ($class, @cmd) = @_;
  die "run_cmd: no command" unless @cmd;

  my ($out_r, $out_w) = portable_pipe;
  my ($err_r, $err_w) = portable_pipe;

  my $proc = run_cmd \@cmd, '>' => $out_w, '2>' => $err_w, close_all => 1;

  close $out_w;
  close $err_w;

  my %cmd = (
    cv   => AE::cv,
    proc => $proc,
  );

  $cmd{hout} = AnyEvent::Handle->new(
    fh      => $out_r,
    on_read => sub {
      my ($h) = @_;
      $h->push_read(line => sub {
        my ($h, $line) = @_;
        _print_ts('>', $line);
      });
    },
    on_eof   => sub { shift->destroy },
    on_error => sub { shift->destroy },
  );

  $cmd{herr} = AnyEvent::Handle->new(
    fh      => $err_r,
    on_read => sub {
      my ($h) = @_;
      $h->push_read(line => sub {
        my ($h, $line) = @_;