File Coverage

blib/lib/File/Set/Writer.pm

Criterion	Covered	Total	%
statement	7	9	77.7
branch			n/a
condition			n/a
subroutine	3	3	100.0
pod			n/a
total	10	12	83.3

package File::Set::Writer;

use Moo;
use MooX::Types::MooseLike::Base qw( Str );
use MooX::Types::MooseLike::Numeric qw( PositiveInt );

our $VERSION = '0.000001'; # 0.0.1
$VERSION = eval $VERSION;

has max_lines => ( is => 'rw', default => sub { 500 }, isa => PositiveInt );

has max_files => ( is => 'rw', default => sub { 100 }, isa => PositiveInt );

has max_handles => ( is => 'rw', required => 1, isa => PositiveInt );

has line_join => ( is => 'rw', default => sub { "\n" }, isa => Str );

has expire_files_batch_size => ( is => 'rw', isa => PositiveInt );

has expire_handles_batch_size => ( is => 'rw', isa => PositiveInt );

# If the user doesn't set a batch_size for files or handles
# we will use 20% of max_(files|handles).  This will be updated
# if max_files or max_handles is updated _unless_ the user explictly
# sets the batch_size, at which point it becomes their responsiblity
# to manage the values.

around expire_files_batch_size => sub {
    my ( $orig, $self ) = ( shift, shift );

    return $self->$orig( @_ ) || int( $self->max_files / 5 );
};

around expire_handles_batch_size => sub {
    my ( $orig, $self ) = ( shift, shift );

    return $self->$orig( @_ ) || int( $self->max_handles / 5 );
};

sub print {
    my ( $self, $file, @lines ) = @_;

    push @{$self->{queue}->{$file}}, @lines;

    $self->_write_files( $file )
        if @{$self->{queue}->{$file}} >= $self->max_lines;

    $self->_write_pending_files
        if $self->_files >= $self->max_files;

    return $self;
}

# Write $self->expire_files_batch_size amount of files to disk,
# in the order of files with the most lines of content.  This
# is used when ->_files >= ->max_files in ->print.

sub _write_pending_files {
    my ( $self ) = @_;

    my @files = sort {
        scalar @{$self->{queue}->{$b} || []} <=> scalar @{$self->{queue}->{$a} || []}
    } keys %{$self->{queue}};

    $self->_write_files( splice @files, 0, $self->expire_files_batch_size );
}

# Given names of files with queued lines, write the lines to the
# file handle with $self->_write(), joining the lines together with
# $self->line_join.

sub _write_files {
    my ( $self, @files ) = @_;

    foreach my $file ( @files ) {
        die "Error _write_files called with invalid argument \"$file\""
            unless defined $file and exists $self->{queue}->{$file};

        $self->_write(
            $file,
            join( $self->line_join, @{$self->{queue}->{$file}}, '' )
        );
        delete $self->{queue}->{$file};
    }
}


# Given a filename and a message, write the message to the file.
#
# This function implements a Least Recently Used (LRU) algorithm to cache file
# handles for repeated use.
# $self->max_handles is the limit of open file descriptors at any given time,
# while $self->expires_handles_batch_size handles will be closed when max_handles
# has been reached.

sub _write {
    my ( $self, $file, @contents ) = @_;

    if ( $self->_handles >= $self->max_handles ) {
        my @files = sort {
            $self->{fcache}->{$a}->{stamp} <=> $self->{fcache}->{$b}->{stamp}
        } keys %{$self->{fcache}};

        foreach my $i ( 0 .. $self->expire_handles_batch_size ) {
            last unless $files[$i];
            delete $self->{fcache}->{$files[$i]};
        }
    }

    if ( ! exists $self->{fcache}->{$file} ) {
        open my $new_fh, ">>", $file
            or die "Failed to open $file for writing: $!";
        $self->{fcache}->{$file} = {
            fh          => $new_fh,
            name        => $file,
            stamp       => time(),
        };
    }

    my $wfh = $self->{fcache}->{$file}->{fh};
    my $content = join ("", @contents);
    print $wfh $content
        or die "Failed to write $file: $!";
    $self->{fcache}->{$file}->{stamp} = time;
}

# Write all staged data to disk and closes all currently-open
# file handles.  This happens automatically at the objects
# destruction.

sub _sync {
    my ( $self ) = @_;

    $self->_write_files( keys %{$self->{queue}} );
}

# Return the count of open file handles currently in the cache.

sub _handles {
    return scalar keys %{ shift->{fcache} || {} };
}

# Return the count of files currently staged for being written.

sub _files {
    return scalar keys %{ shift->{queue} || {} };
}

# $self->_lines( "filename" );
#
# Return the count of lines staged for the given filename.

sub _lines {
    return scalar @{ shift->{queue}->{ shift() } || [] };
}

# Push our buffered arrays into the file handles before
# we close the file handles.
sub DESTROY { shift->_sync; }

1;


__END__