File Coverage

blib/lib/POE/Wheel/FollowTail.pm

Criterion	Covered	Total	%
statement	241	284	84.8
branch	87	152	57.2
condition	21	55	38.1
subroutine	21	23	91.3
pod	4	5	80.0
total	374	519	72.0

package POE::Wheel::FollowTail;

use strict;

use vars qw($VERSION @ISA);
$VERSION = '1.367'; # NOTE - Should be #.### (three decimal places)

use Carp qw( croak carp );
use Symbol qw( gensym );
use POSIX qw(SEEK_SET SEEK_CUR SEEK_END S_ISCHR S_ISBLK);
use POE qw(Wheel Driver::SysRW Filter::Line);
push @ISA, qw(POE::Wheel);
use IO::Handle ();
use Errno qw(ENOENT);

sub CRIMSON_SCOPE_HACK ($) { 0 }

sub SELF_HANDLE      () {  0 }
sub SELF_FILENAME    () {  1 }
sub SELF_DRIVER      () {  2 }
sub SELF_FILTER      () {  3 }
sub SELF_INTERVAL    () {  4 }
sub SELF_EVENT_INPUT () {  5 }
sub SELF_EVENT_ERROR () {  6 }
sub SELF_EVENT_RESET () {  7 }
sub SELF_UNIQUE_ID   () {  8 }
sub SELF_STATE_READ  () {  9 }
sub SELF_LAST_STAT   () { 10 }
sub SELF_FOLLOW_MODE () { 11 }
sub SELF_EVENT_IDLE  () { 12 }

sub MODE_TIMER  () { 0x01 } # Follow on a timer loop.
sub MODE_SELECT () { 0x02 } # Follow via select().

# Turn on tracing.  A lot of debugging occurred just after 0.11.
sub TRACE_POLL         () { 0 }
sub TRACE_RESET        () { 0 }
sub TRACE_STAT         () { 0 }
sub TRACE_STAT_VERBOSE () { 0 }

# Tk doesn't provide a SEEK method, as of 800.022
BEGIN {
  if (exists $INC{'Tk.pm'}) {
    eval <<'    EOE';
      sub Tk::Event::IO::SEEK {
        my $o = shift;
        $o->wait(Tk::Event::IO::READABLE);
        my $h = $o->handle;
        sysseek($h, shift, shift);
      }
    EOE
  }
}

#------------------------------------------------------------------------------

sub new {
  my $type = shift;
  my %params = @_;

  croak "wheels no longer require a kernel reference as their first parameter"
    if @_ and (ref($_[0]) eq 'POE::Kernel');

  croak "$type requires a working Kernel" unless (defined $poe_kernel);

  croak "FollowTail requires a Handle or Filename parameter, but not both"
    unless $params{Handle} xor defined $params{Filename};

  my $driver = delete $params{Driver};
  $driver = POE::Driver::SysRW->new() unless defined $driver;

  my $filter = delete $params{Filter};
  $filter = POE::Filter::Line->new() unless defined $filter;

  croak "InputEvent required" unless defined $params{InputEvent};

  my $handle   = $params{Handle};
  my $filename = $params{Filename};

  my $poll_interval = (
    defined($params{PollInterval})
    ?  $params{PollInterval}
    : 1
  );

  my $seek;
  if (exists $params{SeekBack}) {
    $seek = $params{SeekBack} * -1;
    if (exists $params{Seek}) {
      croak "can't have Seek and SeekBack at the same time";
    }
  }
  elsif (exists $params{Seek}) {
    $seek = $params{Seek};
  }
  else {
    $seek = -4096;
  }

  my $self = bless [
    $handle,                          # SELF_HANDLE
    $filename,                        # SELF_FILENAME
    $driver,                          # SELF_DRIVER
    $filter,                          # SELF_FILTER
    $poll_interval,                   # SELF_INTERVAL
    delete $params{InputEvent},       # SELF_EVENT_INPUT
    delete $params{ErrorEvent},       # SELF_EVENT_ERROR
    delete $params{ResetEvent},       # SELF_EVENT_RESET
    &POE::Wheel::allocate_wheel_id(), # SELF_UNIQUE_ID
    undef,                            # SELF_STATE_READ
    [ (-1) x 8 ],                     # SELF_LAST_STAT
    undef,                            # SELF_FOLLOW_MODE
    delete $params{IdleEvent},        # SELF_EVENT_IDLE
  ], $type;

  if (defined $filename) {
    $handle = $self->[SELF_HANDLE] = _open_file($filename);
    $self->[SELF_LAST_STAT] = [ (stat $filename)[0..7] ] if $handle;
  }
  elsif (defined $handle) {
    $self->[SELF_LAST_STAT] = [ (stat $handle)[0..7] ];
  }

  # Honor SeekBack and discard partial input if we have a plain file
  # that is successfully open at this point.
  #
  # SeekBack attempts to position the file pointer somewhere before
  # the end of the file.  If it's specified, we assume the user knows
  # where a record begins.  Otherwise we just seek back and discard
  # everything to EOF so we can frame the input record.

  if (defined $handle) {

    # Handle is a plain file.  Honor SeekBack and PollInterval.

    if (-f $handle) {
      my $end = sysseek($self->[SELF_HANDLE], 0, SEEK_END);

      # Seeking back from EOF.
      if ($seek < 0) {
        if (defined($end) and ($end < -$seek)) {
          sysseek($self->[SELF_HANDLE], 0, SEEK_SET);
        }
        else {
          sysseek($self->[SELF_HANDLE], $seek, SEEK_END);
        }
      }

      # Seeking forward from the beginning of the file.
      elsif ($seek > 0) {
        if ($seek > $end) {
          sysseek($self->[SELF_HANDLE], 0, SEEK_END);
        }
        else {
          sysseek($self->[SELF_HANDLE], $seek, SEEK_SET);
        }
      }

      # If they set Seek to 0, we start at the beginning of the file.
      # If it was SeekBack, we start at the end.
      elsif (exists $params{Seek}) {
        sysseek($self->[SELF_HANDLE], 0, SEEK_SET);
      }
      elsif (exists $params{SeekBack}) {
        sysseek($self->[SELF_HANDLE], 0, SEEK_END);
      }
      else {
        die;  # Should never happen.
      }

      # Discard partial input chunks unless a SeekBack was specified.
      unless (defined $params{SeekBack} or defined $params{Seek}) {
        while (defined(my $raw_input = $driver->get($self->[SELF_HANDLE]))) {
          # Skip out if there's no more input.
          last unless @$raw_input;
          $filter->get($raw_input);
        }
      }

      # Start the timer loop.
      $self->[SELF_FOLLOW_MODE] = MODE_TIMER;
      $self->_define_timer_states();

      return $self;
    }

    # Strange things that ought not be tailed?  Directories...

    if (-d $self->[SELF_HANDLE]) {
      croak "FollowTail does not tail directories";
    }

    # Handle is not a plain file.
    # Can only honor SeekBack if it's zero.

    carp "POE::Wheel::FollowTail can't SeekBack special files"
      if $params{SeekBack};

    # The handle isn't legal to multiplex on this platform.
    if (POE::Kernel::RUNNING_IN_HELL and not -S $handle) {
      $self->[SELF_FOLLOW_MODE] = MODE_TIMER;
      $self->_define_timer_states();
      return $self;
    }

    # Multiplexing should be more efficient where it's supported.

    carp "FollowTail does not need PollInterval for special files"
      if defined $params{PollInterval};

    $self->[SELF_FOLLOW_MODE] = MODE_SELECT;
    $self->_define_select_states();
    return $self;
  }

  # We don't have an open filehandle yet.  We can't tell whether
  # multiplexing is legal, and we can't seek back yet.  Don't honor
  # either.

  $self->[SELF_FOLLOW_MODE] = MODE_TIMER;
  $self->_define_timer_states();
  return $self;
}

### Define the select based polling loop.  This relies on stupid
### closure tricks to keep references to $self out of anonymous
### coderefs.  Otherwise a circular reference would occur, and the
### wheel would never self-destruct.

sub _define_select_states {
  my $self = shift;

  my $filter      = $self->[SELF_FILTER];
  my $driver      = $self->[SELF_DRIVER];
  my $handle      = \$self->[SELF_HANDLE];
  my $unique_id   = $self->[SELF_UNIQUE_ID];
  my $event_input = \$self->[SELF_EVENT_INPUT];
  my $event_error = \$self->[SELF_EVENT_ERROR];
  my $event_reset = \$self->[SELF_EVENT_RESET];
  my $event_idle  = \$self->[SELF_EVENT_IDLE];

  TRACE_POLL and warn " defining select state";

  $poe_kernel->state(
    $self->[SELF_STATE_READ] = ref($self) . "($unique_id) -> select read",
    sub {

      # Protects against coredump on older perls.
      0 && CRIMSON_SCOPE_HACK('<');

      # The actual code starts here.
      my ($k, $ses) = @_[KERNEL, SESSION];

      # Reset position.
      eval { sysseek($$handle, 0, SEEK_CUR) };
      $! = 0;

      TRACE_POLL and warn " " . time . " read ok";

      # Read the next chunk, and return its data.  Go around again.
      if (defined(my $raw_input = $driver->get($$handle))) {
        TRACE_POLL and warn " " . time . " raw input";
        $filter->get_one_start($raw_input);
        my $cooked_array;
        while (@{$cooked_array = $filter->get_one()}) {
          foreach my $cooked_input (@$cooked_array) {
            TRACE_POLL and warn " " . time . " cooked input";
            $k->call($ses, $$event_input, $cooked_input, $unique_id);
          }
        }

        # Clear the filehandle's EOF status, if any.
        IO::Handle::clearerr($$handle);

        return;
      }

      # Error reading.  Report the error if it's not EOF, or if it's
      # EOF on a socket or TTY.  Shut down the select, too.
      else {
        if ($! or (-S $$handle) or (-t $$handle)) {
          TRACE_POLL and warn " " . time . " error: $!";
          $$event_error and
            $k->call($ses, $$event_error, 'read', ($!+0), $!, $unique_id);
        }
        elsif (defined $$event_idle) {
          $k->call($ses, $$event_idle, $unique_id);
        }

        $k->select_read($$handle => undef);
        eval { IO::Handle::clearerr($$handle) }; # could be a globref
      }
    }
  );

  $poe_kernel->select_read($$handle, $self->[SELF_STATE_READ]);
}

### Define the timer based polling loop.  This also relies on stupid
### closure tricks.

sub _define_timer_states {
  my $self = shift;

  # Tail by filename.
  if (defined $self->[SELF_FILENAME]) {
    TRACE_POLL and warn " defining timer state for filename tail";
    $self->_generate_filename_timer();
  }
  else {
    TRACE_POLL and warn " defining timer state for handle tail";
    $self->_generate_filehandle_timer();
  }

  # Fire up the loop.  The delay() aspect of the loop will prevent
  # duplicate events from being significant for long.
  $poe_kernel->delay($self->[SELF_STATE_READ], 0);
}

sub _generate_filehandle_timer {
  my $self = shift;

  my $filter        = $self->[SELF_FILTER];
  my $driver        = $self->[SELF_DRIVER];
  my $unique_id     = $self->[SELF_UNIQUE_ID];
  my $poll_interval = $self->[SELF_INTERVAL];
  my $last_stat     = $self->[SELF_LAST_STAT];

  my $filename      = \$self->[SELF_FILENAME];
  my $handle        = \$self->[SELF_HANDLE];
  my $event_input   = \$self->[SELF_EVENT_INPUT];
  my $event_error   = \$self->[SELF_EVENT_ERROR];
  my $event_reset   = \$self->[SELF_EVENT_RESET];
  my $event_idle    = \$self->[SELF_EVENT_IDLE];

  $self->[SELF_STATE_READ] = ref($self) . "($unique_id) -> handle timer read";
  my $state_read    = \$self->[SELF_STATE_READ];

  $poe_kernel->state(
    $$state_read,
    sub {

      # Protects against coredump on older perls.
      0 && CRIMSON_SCOPE_HACK('<');

      # The actual code starts here.
      my ($k, $ses) = @_[KERNEL, SESSION];

      # File isn't open?  We're done.
      unless (defined $$handle and fileno $$handle) {
        TRACE_POLL and warn " ", time, " $$handle closed";
        $$event_error and
          $k->call($ses, $$event_error, 'read', 0, "", $unique_id);
        return;
      }

      # Reset position.
      eval { sysseek($$handle, 0, SEEK_CUR) };
      $! = 0;

      # Read the next chunk, and return its data.  Go around again.
      if (defined(my $raw_input = $driver->get($$handle))) {
        TRACE_POLL and warn " " . time . " raw input";
        $filter->get_one_start($raw_input);
        while (1) {
          my $next_rec = $filter->get_one();
          last unless @$next_rec;
          foreach my $cooked_input (@$next_rec) {
            TRACE_POLL and warn " " . time . " cooked input";
            $k->call($ses, $$event_input, $cooked_input, $unique_id);
          }
        }

        # Clear the filehandle's EOF status, if any.
        IO::Handle::clearerr($$handle);

        # Must use a timer so that it can be cleared in DESTROY.
        $k->delay($$state_read, 0) if defined $$state_read;
        return;
      }

      # Some kind of important error?
      if ($!) {
        TRACE_POLL and warn " ", time, " $$handle error: $!";
        $$event_error and
          $k->call($ses, $$event_error, 'read', ($!+0), "$!", $unique_id);
        return;
      }

      # Merely EOF.  Check for file rotation.

      my @new_stat = (
        (defined $$filename)
        ? ((stat $$filename)[0..7])
        : ((stat $$handle)[0..7])
      );

      unless (@new_stat) {
        TRACE_POLL and warn " ", time, " $$handle stat error";
        $$event_error and
          $k->call($ses, $$event_error, 'stat', ($!+0), "$!", $unique_id);
        return;
      }

      TRACE_STAT_VERBOSE and do {
        my @test_new = @new_stat;
        my @test_old = @$last_stat;
        warn " from: @test_old\n to  : @test_new" if (
          "@test_new" ne "@test_old"
        );
      };

      # Ignore rdev changes for non-device files
      eval {
        if (!S_ISBLK($new_stat[2]) and !S_ISCHR($new_stat[2])) {
          $last_stat->[6] = $new_stat[6];
        }
      };

      # Something fundamental about the file changed.
      # Consider it a reset, and try to rewind to the top of the file.
      if (
        $new_stat[1] != $last_stat->[1] or # inode's number
        $new_stat[0] != $last_stat->[0] or # inode's device
        $new_stat[6] != $last_stat->[6] or # device type
        $new_stat[3] != $last_stat->[3] or # number of links
        $new_stat[7] <  $last_stat->[7]    # size reduced
      ) {
        TRACE_STAT and do {
          warn " inode $new_stat[1] != old $last_stat->[1]"
            if $new_stat[1] != $last_stat->[1];
          warn " inode device $new_stat[0] != old $last_stat->[0]"
            if $new_stat[0] != $last_stat->[0];
          warn " device type $new_stat[6] != old $last_stat->[6]"
            if $new_stat[6] != $last_stat->[6];
          warn " link count $new_stat[3] != old $last_stat->[3]"
            if $new_stat[3] != $last_stat->[3];
          warn " file size $new_stat[7] < old $last_stat->[7]"
            if $new_stat[7] < $last_stat->[7];
        };

        # File has reset.
        TRACE_RESET and warn " filehandle has reset";
        $$event_reset and $k->call($ses, $$event_reset, $unique_id);

        sysseek($$handle, 0, SEEK_SET);
      }
      elsif (defined $$event_idle) {
        $k->call($ses, $$event_idle, $unique_id);
      }

      # The file didn't roll.  Try again shortly.
      @$last_stat = @new_stat;
      IO::Handle::clearerr($$handle);
      $k->delay($$state_read, $poll_interval) if defined $$state_read;
      return;
    }
  );
}

sub _generate_filename_timer {
  my $self = shift;

  my $filter        = $self->[SELF_FILTER];
  my $driver        = $self->[SELF_DRIVER];
  my $unique_id     = $self->[SELF_UNIQUE_ID];
  my $poll_interval = $self->[SELF_INTERVAL];
  my $filename      = $self->[SELF_FILENAME];
  my $last_stat     = $self->[SELF_LAST_STAT];

  my $handle        = \$self->[SELF_HANDLE];
  my $event_input   = \$self->[SELF_EVENT_INPUT];
  my $event_error   = \$self->[SELF_EVENT_ERROR];
  my $event_reset   = \$self->[SELF_EVENT_RESET];
  my $event_idle    = \$self->[SELF_EVENT_IDLE];

  $self->[SELF_STATE_READ] = ref($self) . "($unique_id) -> name timer read";
  my $state_read    = \$self->[SELF_STATE_READ];

  $poe_kernel->state(
    $$state_read,
    sub {

      # Protects against coredump on older perls.
      0 && CRIMSON_SCOPE_HACK('<');

      # The actual code starts here.
      my ($k, $ses) = @_[KERNEL, SESSION];

      # File isn't open?  Try to open it.
      unless ($$handle) {
        $$handle = _open_file($filename);

        # Couldn't open yet.
        unless ($$handle) {
          $k->call($ses, $$event_idle, $unique_id) if defined $$event_idle;
          $k->delay($$state_read, $poll_interval) if defined $$state_read;
          return;
        }

        # File has reset.
        TRACE_RESET and warn " file name has reset";
        $$event_reset and $k->call($ses, $$event_reset, $unique_id);

        @$last_stat = (stat $filename)[0..7];
      }
      else {
        # Reset position.
        eval { sysseek($$handle, 0, SEEK_CUR) };
        $! = 0;
      }

      # Read the next chunk, and return its data.  Go around again.
      if (defined(my $raw_input = $driver->get($$handle))) {
        TRACE_POLL and warn " " . time . " raw input";
        $filter->get_one_start($raw_input);
        my $cooked_array;
        while (@{$cooked_array = $filter->get_one()}) {
          foreach my $cooked_input (@$cooked_array) {
            TRACE_POLL and warn " " . time . " cooked input";
            $k->call($ses, $$event_input, $cooked_input, $unique_id);
          }
        }

        # Clear the filehandle's EOF status, if any.
        IO::Handle::clearerr($$handle);

        # Must use a timer so that it can be cleared in DESTROY.
        $k->delay($$state_read, 0) if defined $$state_read;
        return;
      }

      # Some kind of important error?
      if ($!) {
        TRACE_POLL and warn " ", time, " $$handle error: $!";
        $$event_error and
          $k->call($ses, $$event_error, 'read', ($!+0), "$!", $unique_id);
        return;
      }

      # Merely EOF.  Check for file rotation.
      my @new_stat = (stat $filename)[0..7];
      unless (@new_stat) {
        TRACE_POLL and warn " ", time, " $filename stat error: $!";
        if ($! != ENOENT) {
          $$event_error and
            $k->call($ses, $$event_error, 'stat', ($!+0), "$!", $unique_id);
          return;
        }
        @new_stat = (-1) x 8;
      }

      TRACE_STAT_VERBOSE and do {
        my @test_new = @new_stat;
        my @test_old = @$last_stat;
        warn " from: @test_old\n to  : @test_new" if (
          "@test_new" ne "@test_old"
        );
      };

      # Ignore rdev changes for non-device files
      eval {
        if (!S_ISBLK($new_stat[2]) and !S_ISCHR($new_stat[2])) {
          $last_stat->[6] = $new_stat[6];
        }
      };

      # Something fundamental about the file changed.
      # Consider it a reset, and close the file.
      if (
        $new_stat[1] != $last_stat->[1] or # inode's number
        $new_stat[0] != $last_stat->[0] or # inode's device
        $new_stat[6] != $last_stat->[6] or # device type
        $new_stat[3] != $last_stat->[3] or # number of links
        $new_stat[7] <  $last_stat->[7]    # size reduced
      ) {
        TRACE_STAT and do {
          warn " inode $new_stat[1] != old $last_stat->[1]"
            if $new_stat[1] != $last_stat->[1];
          warn " inode device $new_stat[0] != old $last_stat->[0]"
            if $new_stat[0] != $last_stat->[0];
          warn " device type $new_stat[6] != old $last_stat->[6]"
            if $new_stat[6] != $last_stat->[6];
          warn " link count $new_stat[3] != old $last_stat->[3]"
            if $new_stat[3] != $last_stat->[3];
          warn " file size $new_stat[7] < old $last_stat->[7]"
            if $new_stat[7] < $last_stat->[7];
        };

        $$handle = undef;
        @$last_stat = @new_stat;

        # Must use a timer so that it can be cleared in DESTROY.
        $k->delay($$state_read, 0) if defined $$state_read;
        return;
      }
      elsif (defined $$event_idle) {
        $k->call($ses, $$event_idle, $unique_id);
      }

      # The file didn't roll.  Try again shortly.
      @$last_stat = @new_stat;
      IO::Handle::clearerr($$handle);
      $k->delay($$state_read, $poll_interval) if defined $$state_read;
      return;
    }
  );
}

#------------------------------------------------------------------------------

sub event {
  my $self = shift;
  push(@_, undef) if (scalar(@_) & 1);

  while (@_) {
    my ($name, $event) = splice(@_, 0, 2);

    if ($name eq 'InputEvent') {
      if (defined $event) {
        $self->[SELF_EVENT_INPUT] = $event;
      }
      else {
        carp "InputEvent requires an event name.  ignoring undef";
      }
    }
    elsif ($name eq 'ErrorEvent') {
      $self->[SELF_EVENT_ERROR] = $event;
    }
    elsif ($name eq 'ResetEvent') {
      $self->[SELF_EVENT_RESET] = $event;
    }
    elsif ($name eq 'IdleEvent') {
      $self->[SELF_EVENT_IDLE] = $event;
    }
    else {
      carp "ignoring unknown FollowTail parameter '$name'";
    }
  }
}

#------------------------------------------------------------------------------

sub DESTROY {
  my $self = shift;

  # Remove our tentacles from our owner.
  $poe_kernel->select_read($self->[SELF_HANDLE] => undef) if (
    defined $self->[SELF_HANDLE]
  );

  if ($self->[SELF_STATE_READ]) {
    $poe_kernel->delay($self->[SELF_STATE_READ]);
    $poe_kernel->state($self->[SELF_STATE_READ]);
    undef $self->[SELF_STATE_READ];
  }

  &POE::Wheel::free_wheel_id($self->[SELF_UNIQUE_ID]);
}

#------------------------------------------------------------------------------

sub ID {
  return $_[0]->[SELF_UNIQUE_ID];
}

sub tell {
  my $self = shift;
  return sysseek($self->[SELF_HANDLE], 0, SEEK_CUR);
}

sub _open_file {
  my $filename = shift;

  my $handle = gensym();

  # FIFOs (named pipes) are opened R/W so they don't report EOF.
  # Everything else is opened read-only.
  if (-p $filename) {
    return $handle if open $handle, "+<", $filename;
    return;
  }

  return $handle if open $handle, "<", $filename;
  return;
}

1;

__END__