view release on metacpan or  search on metacpan

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

# Copyright (c) 2009 AdCopy
# Author: Jeff Weisberg
# Created: 2009-Dec-14 17:49 (EST)
# Function: get file from remote scriblr
#
# $Id: Xfer.pm,v 1.7 2011/01/14 22:24:30 jaw Exp $

package AC::MrGamoo::Xfer;
use AC::MrGamoo::Debug 'xfer';
use AC::MrGamoo::Config;
use AC::MrGamoo::Protocol;
use AC::MrGamoo::EUConsole;
use AC::MrGamoo::PeerList;
use Digest::SHA1;
use File::Path;
use Socket;
use strict;

require 'AC/protobuf/std_reply.pl';
require 'AC/protobuf/scrible.pl';

our @ISA = 'AC::DC::IO::Forked';

my $TIMEOUT    = 3600;	# QQQ
my $BUFSIZ     = 65536;
my $STATUSTIME = 5;
my $MAXRUNNING = 4;	# QQQ - configurable?
my %REGISTRY;

################################################################

# schedule periodic "cronjob"
AC::DC::Sched->new(
    info	=> "xfer periodic",
    freq	=> 5,
    func	=> \&periodic,
   );

################################################################


sub new {
    my $class   = shift;
    my $srcname = shift;	# on remote system
    my $dstname = shift;	# on local system
    my $loc     = shift;
    my $req     = shift;	# APCMRMFileXfer

    if( $REGISTRY{$req->{copyid}} ){
        verbose("ignoring duplicate xfer $req->{copyid}");
        return $REGISTRY{$req->{copyid}};
    }

    $dstname = conf_value('basedir') . '/' . $dstname;
    my $tmpfile = $dstname . ".$$";

    # mkpath
    my($dir) = $dstname =~ m|^(.+)/[^/]+$|;
    eval{ mkpath($dir, undef, 0755) };

    my $me = $class->SUPER::new( \&_run_child,
                                 [ $srcname, $dstname, $tmpfile, $loc, $req ],
                                 info     => "xfer $loc:$srcname",
                                 request  => $req,
                                 rbufsize => 65536,
                                );

    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;