File Coverage

blib/lib/Hadoop/Streaming/Reducer/Input.pm

Criterion	Covered	Total	%
statement	6	27	22.2
branch	0	10	0.0
condition			n/a
subroutine	2	7	28.5
pod	5	5	100.0
total	13	49	26.5

package Hadoop::Streaming::Reducer::Input;
$Hadoop::Streaming::Reducer::Input::VERSION = '0.143060';
use Moo;
use Hadoop::Streaming::Reducer::Input::Iterator;

#ABSTRACT: Parse input stream for reducer

has handle => (
    is       => 'ro',
    does     => 'FileHandle',
    required => 1,
);

has buffer => (
    is   => 'rw',
);


sub next_key
{
    my $self = shift;
    my $line = $self->buffer ? $self->buffer : $self->next_line;
    return if not defined $line;
    my ( $key, $value ) = split /\t/, $line, 2;
    return $key;
}


sub next_line {
    my $self = shift;
    return if $self->handle->eof;
    $self->buffer( $self->handle->getline );
    $self->buffer;
}


sub getline {
    my $self = shift;
    if (defined $self->buffer) {
        my $buf = $self->buffer;
        $self->buffer(undef);
        return $buf;
    } else {
        return $self->next_line;
    }
}


sub iterator {
    my $self = shift;
    Hadoop::Streaming::Reducer::Input::Iterator->new( input => $self );
}


sub each
{
    my $self = shift;
    my $line = $self->getline or return;
    chomp $line;
    split /\t/, $line, 2;
}

__PACKAGE__->meta->make_immutable;

1;

__END__