AC-Yenta

 view release on metacpan or  search on metacpan

lib/AC/Yenta/Client.pm  view on Meta::CPAN

    my $addr = shift;
    my $port = shift;
    my $req  = shift;
    my $file = shift;	# reference

    my $ipn = inet_aton($addr);
    $req .= $$file if $file;

    $me->{debug}->("trying to contact yenta server $addr:$port");
    my $res;
    eval {
        $res = $me->{proto}->send_request($ipn, $port, $req, $me->{debug}, $me->{timeout});
        $res->{data} = $me->{proto}->decode_reply( $res ) if $res;
    };
    if(my $e = $@){
        $me->{debug}->("yenta request failed: $e");
        $res = undef;
    }
    return $res;
}

lib/AC/Yenta/Customize.pm  view on Meta::CPAN

use strict;

sub customize {
    my $class  = shift;
    my $implby = shift;

    (my $default = $class) =~ s/(.*)::([^:]+)$/$1::Default::$2/;

    # load user's implemantation + default
    for my $p ($implby, $default){
        eval "require $p" if $p;
        die $@ if $@;
    }

    # import/export
    no strict;
    no warnings;
    for my $f ( @{$class . '::CUSTOM'} ){
        *{$class . '::' . $f} = ($implby && $implby->can($f)) || $default->can($f);
    }
}

lib/AC/Yenta/Kibitz/Status.pm  view on Meta::CPAN

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

# do not believe a client that says it is up
# put it on the sceptical queue, and check for ourself
sub update_sceptical {
    my $gpb = shift;
    my $io  = shift;

    return unless $gpb;
    my $c;
    eval {
        $c = ACPYentaStatusRequest->decode( $gpb );
        $c = $c->{myself};
    };
    if(my $e = $@){
        problem("cannot decode status data: $e");
        return;
    }

    my $id = $c->{server_id};

lib/AC/Yenta/Kibitz/Status.pm  view on Meta::CPAN

    return if AC::Yenta::Status->my_server_id() eq $id;

    AC::Yenta::Status->update_sceptical($id, $c, $io);
}

sub update {
    my $gpb = shift;

    return unless $gpb;
    my $c;
    eval {
        $c = ACPYentaStatusReply->decode( $gpb );
    };
    if(my $e = $@){
        problem("cannot decode status data: $e");
        return;
    }

    debug("rcvd update");
    my $myself = AC::Yenta::Status->my_server_id();

lib/AC/Yenta/Kibitz/Store/Client.pm  view on Meta::CPAN

    my $evt = shift;

    debug("recvd reply");

    my $yp = AC::Yenta::Protocol->new( secret => conf_value('secret') );
    my($proto, $data, $content) = $yp->read_protocol( $me, $evt );
    $me->timeout_rel($TIMEOUT) if $evt->{data} && !$proto;
    return unless $proto;
    $proto->{data}    = $data;
    $proto->{content} = $content;
    eval {
        my $yp = AC::Yenta::Protocol->new();
        $proto->{data} = $yp->decode_reply($proto) if $data;
    };
    if(my $e = $@){
        problem("cannot decode reply: $e");
    }

    # process
    $me->{_store_ok} = 1;
    if( $proto->{is_error} || $@ ){

lib/AC/Yenta/Kibitz/Store/Server.pm  view on Meta::CPAN

    my $gpb     = shift;
    my $content = shift;	# not used

    unless( $proto->{want_reply} ){
        $io->shut();
        return;
    }

    # decode request
    my $req;
    eval {
        $req = ACPYentaGetSet->decode( $gpb );
    };
    if(my $e = $@){
        problem("cannot decode request: $e");
        $io->shut();
        return;
    }

    # process requests
    my @res;

lib/AC/Yenta/Kibitz/Store/Server.pm  view on Meta::CPAN

    my $gpb     = shift;
    my $content = shift;	# not used

    unless( $proto->{want_reply} ){
        $io->shut();
        return;
    }

    # decode request
    my $req;
    eval {
        $req = ACPYentaCheckRequest->decode( $gpb );
    };
    if(my $e = $@){
        problem("cannot decode request: $e");
        $io->shut();
        return;
    }

    debug("check request: $req->{map}, $req->{version}, $req->{level}");

lib/AC/Yenta/Kibitz/Store/Server.pm  view on Meta::CPAN

}

sub api_distrib {
    my $io      = shift;
    my $proto   = shift;
    my $gpb     = shift;
    my $content = shift;	# reference

    # decode request
    my $req;
    eval {
        $req = ACPYentaDistRequest->decode( $gpb );
        die "invalid k/v for put request\n" unless $req->{datum}{key} && $req->{datum}{version};
    };
    if(my $e = $@){
        my $enc = $proto->{data_encrypted} ? ' (encrypted)' : '';
        problem("cannot decode request: peer: $io->{peerip} $enc, $e");
        $io->shut();
        return;
    }

lib/AC/Yenta/Kibitz/Store/Server.pm  view on Meta::CPAN

    $io->write_and_shut( $response );

}

sub _check_content {
    my $meta = shift;
    my $cont = shift;

    return 1 unless $meta && $meta =~ /^\{/;

    eval {
        $meta = decode_json($meta);
    };
    return 1 if $@;

    if( $meta->{sha1} ){
        my $chk = sha1_base64( $$cont );
        return unless $chk eq $meta->{sha1};
    }
    if( $meta->{size} ){
        my $len = length($$cont);

lib/AC/Yenta/Monitor.pm  view on Meta::CPAN

        debug("monitor $id down too long. removing from report");
        delete $MON{$id};
    }
}

sub update {
    my $id = shift;
    my $gb = shift;	# ACPHeartbeat

    my $up;
    eval {
        $up = ACPHeartBeat->decode( $gb );
        $up->{id} = $id;
    };
    if(my $e = $@){
        problem("cannot decode hb data: $e");
        return isdown($id, 0);
    }
    unless( $up->{status_code} == 200 ){
        return isdown($id, $up->{status_code});
    }

lib/AC/Yenta/Protocol.pm  view on Meta::CPAN

    my $me  = shift;
    my $io  = shift;
    my $evt = shift;

    $io->{rbuffer} .= $evt->{data};

    return _read_http($io, $evt) if $io->{rbuffer} =~ /^GET/;

    if( length($io->{rbuffer}) >= $HDRSIZE && !$io->{proto_header} ){
        # decode header
        eval {
            $io->{proto_header} = $me->decode_header( $io->{rbuffer} );
        };
        if(my $e=$@){
            verbose("cannot decode protocol header: $e");
            $io->run_callback('error', {
                cause	=> 'read',
                error	=> "cannot decode protocol: $e",
            });
            $io->shut();
            return;

lib/AC/Yenta/Protocol.pm  view on Meta::CPAN

}

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

sub _decrypt_data {
    my $me   = shift;
    my $io   = shift;
    my $auth = shift;
    my $data = shift;

    eval {
        $data = $me->decrypt( $auth, $data );
    };
    if(my $e=$@){
        verbose("cannot decrypt protocol data: $e");
        $io->run_callback('error', {
            cause	=> 'read',
            error	=> "cannot decrypt protocol: $e",
        });
        $io->shut();
        return;

lib/AC/Yenta/Store/File.pm  view on Meta::CPAN

    my $cf   = $me->{conf};
    my $base = $cf->{basedir};
    return 1 unless $base;

    # split name into dir / file
    my($dir, $file) = $name =~ m|(.*)/([^/]+)$|;

    # create directory
    debug("mkpath: $base/$dir");
    my $mask = umask 0;
    eval { mkpath("$base/$dir", undef, 0777); };
    umask $mask;

    # save file
    my $f;
    unless( open($f, "> $base/$name.tmp") ){
        problem("cannot save file '$base/$name.tmp': $!");
        return;
    }

    debug("saving file '$base/$name'");

lib/AC/Yenta/Store/LevelDB.pm  view on Meta::CPAN

package AC::Yenta::Store::LevelDB;
use AC::Yenta::Debug 'ldb';
use strict;

# does not support concurrent access
# does not work on sparc


BEGIN {
    # only if we have it (no LevelDB on sparc)
    eval {
        require Tie::LevelDB;
        AC::Yenta::Store::Map->add_backend( ldb 	=> 'AC::Yenta::Store::LevelDB' );
        AC::Yenta::Store::Map->add_backend( leveldb 	=> 'AC::Yenta::Store::LevelDB' );
    };
};

my %OPEN;

sub new {
    my $class = shift;

lib/AC/Yenta/Store/SQLite.pm  view on Meta::CPAN

    my $r = $st->fetchall_arrayref({});

    return @$r;
}

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

sub _init {
    my $db = shift;

    eval {
        for my $sql (split /;/, $initsql){
            $sql =~ s/--\s.*$//gm;              # remove comments
            next unless $sql !~ /^\s*$/;
            _do($db, $sql);
        }
    };
    if(my $e=$@){
        # QQQ?
        problem("error initializing sqlite db: $e");
    }
}

sub _do {
    my $db  = shift;
    my $sql = shift;

    my( $st, $nrow );
    eval {
        debug("sql: $sql");
        $st   = $db->prepare( $sql );
        $nrow = $st->execute( @_ );
    };
    my $e = $@;
    die $e if $e;

    return $st;
}

lib/AC/Yenta/Store/Tokyo.pm  view on Meta::CPAN

use strict;

# does not work on sparc (tests sigbus)
# new version does not compile with gcc 3.4.3
#
# faster average performance than BDB
# worse worst-case performance than BDB

BEGIN {
    # only if we have it (not on sparc)
    eval {
        require TokyoCabinet;

        AC::Yenta::Store::Map->add_backend( tcb 	=> 'AC::Yenta::Store::Tokyo' );
        AC::Yenta::Store::Map->add_backend( tokyo 	=> 'AC::Yenta::Store::Tokyo' );
    };
};

sub new {
    my $class = shift;
    my $name  = shift;