view release on metacpan or  search on metacpan

lib/Hadoop/HDFS/Command.pm  view on Meta::CPAN

    if ( $cb ) {
        die "callback needs to be a CODE" if ! Ref::Util::is_coderef $cb;
        if ( defined wantarray ) {
            Carp::croak "You need to call this function in void context when callback is specified";
        }
    }

    my @response = $self->_capture(
        $options,
        $self->cmd_hdfs,
        qw( dfs -ls ),
        ( map { '-' . $_ } grep { $arg->{ $_ } } @flags ),
        @{ $paths },
    );

    # directory is empty
    #
    return if ! @response;

    if ( $response[0] && $response[0] =~ m{ \A Found \s+ [0-9] }xms ) {
        shift @response; # junk
    }

    my $space = q{ };

    my @rv;
    for my $line ( @response ) {
        my($mode, $replication, $user, $group, @unknown) = split m{ \s+ }xms, $line, 5;
        my @rest = map { split $space, $_ } @unknown;
        my $size;
        if ( $arg->{h}) {
            if ( $rest[0] eq '0' || $rest[1] !~ m{ [a-zA-Z_] }xms ) {
                $size = shift @rest;
            }
            else {
                $size = join $space, shift @rest, shift @rest;
            }
        }
        else {
            $size = shift @rest;
        }
        my $date   = join ' ', shift @rest, shift @rest;
        my $path   = shift( @rest ) || die "Unable to parse $line to gather the path";
        my $is_dir = $mode =~ m{ \A [d] }xms ? 1 : 0;

        my %record = (
            mode        => $mode,
            replication => $replication,
            user        => $user,
            group       => $group,
            size        => $size,
            date        => $date,
            path        => $path,
            type        => $is_dir ? 'dir' : 'file',
        );

        if ( $want_epoch ) {
            $strp ||= DateTime::Format::Strptime->new(
                            pattern   => '%Y-%m-%d %H:%M',
                            time_zone => 'CET',
                            on_error  => 'croak',
                        );
            eval {
                $record{epoch} = $strp->parse_datetime( $date )->epoch;
                1;
            } or do {
                my $eval_error = $@ || 'Zombie error';
                $self->_log( debug => 'Failed to convert %s into an epoch: %s',
                                $date,
                                $eval_error,
                );
            };
        }

        if ( @rest ) {
            # interpret as the rest of the path as spaces in paths are ok
            # possibly this will need to be revisited in the future.
            #
            $record{path} = join $space, $record{path}, @rest;
        }

        if ( $cb ) {
            # control the flow from the callback
            # So, the return value matters.
            #
            if ( ! $cb->( \%record ) ) {
                $self->_log( info => 'Terminating the ls processing as the user callback did not return a true value.');
                last;
            }
            next;
        }

        push @rv, { %record };
    }

    return if $cb;
    return @rv;
}

sub _dfs_du {
    my $self = shift;
    my $options = shift;
    my @params  = @_;
    my @flags   = qw( h s );
    my($arg, $paths) = $self->_parse_options(
                            \@params,
                            \@flags,
                            undef,
                            {
                                require_params => 1,
                            },
                        );

    my @rv = $self->_capture(
        $options,
        $self->cmd_hdfs,
        qw( dfs -du ),
        ( map { '-' . $_ } grep { $arg->{ $_ } } @flags ),
        @{ $paths },
    ) or die "No output collected from -du command";