File Coverage

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

Criterion	Covered	Total	%
statement	9	26	34.6
branch	0	8	0.0
condition			n/a
subroutine	3	6	50.0
pod	3	3	100.0
total	15	43	34.8

package Hadoop::Streaming::Reducer::Input::Iterator;
$Hadoop::Streaming::Reducer::Input::Iterator::VERSION = '0.143060';
use Moo;
use Safe::Isa;
with 'Hadoop::Streaming::Role::Iterator';

use Hadoop::Streaming::Reducer::Input::ValuesIterator;

#ABSTRACT: Collects values for each key together with an iterator interface

has input => (
    is       => 'ro',
    isa      => sub {
        die 'not a Hadoop::Streaming::Reducer::Input object'
            unless $_[0]->$_isa('Hadoop::Streaming::Reducer::Input');
    },
    required => 1,
);

has current_key => (
    is   => 'rw',
    does => 'Str'
);


sub has_next {
    my $self = shift;
    return if not defined $self->input->next_key;
    1;
}


sub next {
    my $self = shift;

    if ( not defined $self->current_key ) {
        $self->current_key($self->input->next_key);
        return $self->retval( $self->current_key );
    }

    if ($self->current_key ne $self->input->next_key) {
        $self->current_key($self->input->next_key);
        return $self->retval( $self->current_key );
    }

    my ($key, $value);
    do {
        ($key, $value) = $self->input->each or return;
    } while ($self->current_key eq $key);
    $self->current_key( $key );

    return $self->retval($key, $value);
}


sub retval {
    my ($self, $key, $value) = @_;
    return (
        $key,
        Hadoop::Streaming::Reducer::Input::ValuesIterator->new(
            input_iter => $self,
            first      => $value,
        ),
    );
}

__PACKAGE__->meta->make_immutable;

1;

__END__