view release on metacpan or  search on metacpan

lib/AC/MrGamoo/Xfer.pm  view on Meta::CPAN

                                );

    return unless $me;
    $REGISTRY{$req->{copyid}} = $me;
    debug("xfer requesting $loc:$srcname => $dstname, id $req->{copyid}");

    return $me;
}

sub start {
    my $me = shift;

    my $nrun = 0;
    for my $t (values %REGISTRY){
        $nrun ++ if $t->{fd};
    }

    if( $nrun >= $MAXRUNNING ){
        $me->{_queueprio}    = $^T;
        debug("queue xfer $me->{request}{copyid}");
        return 1;
    }

    $me->_start();

}

sub _start {
    my $me = shift;

    debug("start xfer $me->{request}{copyid}");

    $me->timeout_rel($TIMEOUT);
    $me->set_callback('timeout',  \&timeout);
    $me->set_callback('shutdown', \&shutdown);

    $me->SUPER::start();
}

sub _run_child {
    my $srcname = shift;	# on remote system
    my $dstname = shift;	# on local system
    my $tmpfile = shift;
    my $loc     = shift;
    my $req     = shift;


    exit(0) if -f $dstname;
    my $con = AC::MrGamoo::EUConsole->new( $req->{jobid}, $req->{console} );
    $con->send_msg('debug', "retrieving file $loc:$srcname");

    # RSN - remove scriblr::client
    my $ok;
    if( get_peer_addr_from_id($loc) ){
        $ok = _get_file( $req, $loc, $srcname, $tmpfile );
    }else{
        verbose("cannot locate server: $loc");
    }

    if( $ok ){
        rename $tmpfile, $dstname;
        exit 0;
    }
    exit 1;
}

sub timeout {
    my $me = shift;
    debug("xfer timeout");
    $me->shut();
}

sub shutdown {
    my $me = shift;

    delete $REGISTRY{$me->{request}{copyid}};

    debug("exitval = $me->{exitval}");
    if( !$me->{exitval} ){
        $me->run_callback('on_success');
    }else{
        $me->run_callback('on_failure');
    }

    periodic(1);	# try to start another xfer
}

sub _send_status_update {
    my $req = shift;

    debug('send xfer status');
    AC::MrGamoo::API::Xfer::tell_master( $req, 100, 'Working...' );
}


sub _get_file {
    my $oreq    = shift;
    my $loc     = shift;
    my $srcname = shift;	# on remote system
    my $tmpfile = shift;


    my($addr, $port) = get_peer_addr_from_id( $loc );
    unless( $addr ){
        debug("cannot find addr for $loc");
        return;
    }

    debug("connecting to $addr:$port");

    my $req = AC::MrGamoo::Protocol->encode_request( {
        type		=> 'scribl_get',
        msgidno		=> $$,
        want_reply	=> 1,
    }, { filename => $srcname } );

    my $p;
    eval {
        # connect
        my $s = AC::MrGamoo::Protocol->connect_to_server( inet_aton($addr), $port );
        return unless $s;