view release on metacpan or  search on metacpan

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

package Hadoop::HDFS::Command;
$Hadoop::HDFS::Command::VERSION = '0.007';
use 5.010;
use strict;
use warnings;
use Capture::Tiny   ();
use Carp            ();
use Data::Dumper;
use DateTime::Format::Strptime;
use DateTime;
use Getopt::Long    ();
use IPC::Cmd        ();
use Ref::Util       ();
use Time::HiRes   qw( time );
use Types::Standard qw(Bool Str);

{ use Moo; }

has cmd_hdfs => (
    is  => 'rw',
    isa => sub {
        my $val = shift;
        return if $val && -e $val && -x _;
        Carp::confess sprintf "The command `%s` either does not exist or not an executable!",
                        $val,
        ;
    },
    default => sub { '/usr/bin/hdfs' },
    lazy    => 1,
);

has enable_log => (
    is      => 'rw',
    isa     => Bool,
    default => sub { 0 },
    lazy    => 1,
);

has trace_logs => (
    is      => 'rw',
    isa     => Bool,
    default => sub { 0 },
    lazy    => 1,
);

has runas => (
    is      => 'rw',
    isa     => Str,
    default => scalar getpwuid $<,
    lazy    => 1,
);

before ['_capture', '_capture_with_stdin'] => sub {
    my ($self, $options, @cmd) = @_;
    unshift @cmd, 'sudo', '-u', $self->runas
        unless $self->runas eq getpwuid $<;
    @_ = ($self, $options, @cmd);
};

sub dfs {
    my $self = shift;
    my $options = Ref::Util::is_hashref $_[0] ? shift( @_ ) : {};
    (my $cmd    = shift || die "No dfs command specified") =~ s{ \A [-]+ }{}xms;
    my $method  = '_dfs_' . $cmd;
    Carp::croak "'$cmd' is not implemented!" if ! $self->can( $method );
    $self->$method( $options, @_ );
}

sub _dfs_ls {
    my $self = shift;
    state $strp;

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

    my $want_epoch = $options->{want_epoch};
    my $cb = delete $options->{callback};

    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;