view release on metacpan or  search on metacpan

local/lib/perl5/IO/Async/FileStream.pm  view on Meta::CPAN

   foreach (qw( interval )) {
      $self->{file}->configure( $_ => delete $params{$_} ) if exists $params{$_};
   }
   if( exists $params{filename} ) {
      $self->{file}->configure( filename => delete $params{filename} );
      $params{read_handle} = $self->{file}->handle;
   }
   elsif( exists $params{handle} or exists $params{read_handle} ) {
      my $handle = delete $params{handle};
      defined $handle or $handle = delete $params{read_handle};

      $self->{file}->configure( handle => $handle );
      $params{read_handle} = $self->{file}->handle;
   }

   croak "Cannot have a write_handle in a ".ref($self) if defined $params{write_handle};

   $self->SUPER::configure( %params );

   if( $self->read_handle and !defined $self->{last_size} ) {
      my $size = (stat $self->read_handle)[7];

      $self->{last_size} = $size;

      local $self->{running_initial} = 1;
      $self->maybe_invoke_event( on_initial => $size );
   }
}

=head1 METHODS

=cut

# Replace IO::Async::Handle's implementation
sub _watch_read
{
   my $self = shift;
   my ( $want ) = @_;

   if( $want ) {
      $self->{file}->start if !$self->{file}->is_running;
   }
   else {
      $self->{file}->stop;
   }
}

sub _watch_write
{
   my $self = shift;
   my ( $want ) = @_;

   croak "Cannot _watch_write in " . ref($self) if $want;
}

sub on_devino_changed
{
   my $self = shift or return;

   $self->{renamed} = 1;
   $self->debug_printf( "read tail of old file" );
   $self->read_more;
}

sub on_size_changed
{
   my $self = shift or return;
   my ( $size ) = @_;

   if( $size < $self->{last_size} ) {
      $self->maybe_invoke_event( on_truncated => );
      $self->{last_pos} = 0;
   }

   $self->{last_size} = $size;

   $self->debug_printf( "read_more" );
   $self->read_more;
}

sub read_more
{
   my $self = shift;

   sysseek( $self->read_handle, $self->{last_pos}, SEEK_SET ) if defined $self->{last_pos};

   $self->on_read_ready;

   $self->{last_pos} = sysseek( $self->read_handle, 0, SEEK_CUR ); # == systell

   if( $self->{last_pos} < $self->{last_size} ) {
      $self->loop->later( sub { $self->read_more } );
   }
   elsif( $self->{renamed} ) {
      $self->debug_printf( "reopening for rename" );

      $self->{last_size} = 0;

      if( $self->{last_pos} ) {
         $self->maybe_invoke_event( on_truncated => );
         $self->{last_pos} = 0;
         $self->loop->later( sub { $self->read_more } );
      }

      $self->configure( read_handle => $self->{file}->handle );
      undef $self->{renamed};
   }
}

sub write
{
   carp "Cannot ->write from a ".ref($_[0]);
}

=head2 seek

   $filestream->seek( $offset, $whence )

Callable only during the C<on_initial> event. Moves the read position in the
filehandle to the given offset. C<$whence> is interpreted as for C<sysseek>,
should be either C<SEEK_SET>, C<SEEK_CUR> or C<SEEK_END>. Will be set to
C<SEEK_SET> if not provided.

Normally this would be used to seek to the end of the file, for example

 on_initial => sub {
    my ( $self, $filesize ) = @_;
    $self->seek( $filesize );
 }

=cut

sub seek
{
   my $self = shift;
   my ( $offset, $whence ) = @_;

   $self->{running_initial} or croak "Cannot ->seek except during on_initial";

   defined $whence or $whence = SEEK_SET;

   sysseek( $self->read_handle, $offset, $whence );
}

=head2 seek_to_last

   $success = $filestream->seek_to_last( $str_pattern, %opts )

Callable only during the C<on_initial> event. Attempts to move the read
position in the filehandle to just after the last occurance of a given match.
C<$str_pattern> may be a literal string or regexp pattern. 

Returns a true value if the seek was successful, or false if not. Takes the
following named arguments: