File Coverage

blib/lib/Jonk.pm

Criterion	Covered	Total	%
statement	15	132	11.3
branch	0	30	0.0
condition	0	29	0.0
subroutine	5	32	15.6
pod	5	5	100.0
total	25	228	10.9

package Jonk;
use strict;
use warnings;
use Jonk::Job;
use Try::Tiny;
use Carp ();

our $VERSION = '0.10_02';

sub new {
    my ($class, $dbh, $opts) = @_;

    unless ($dbh) {
        Carp::croak('missing job queue database handle.');
    }

    my $functions  = _parse_functions($opts);
    my $table_name = $opts->{table_name} || 'job';
    my $driver     = _verify_driver($dbh);

    bless {
        dbh           => $dbh,
        table_name    => $table_name,
        functions     => $functions,
        driver        => $driver,
        has_func      => scalar(keys %{$functions}) ? 1 : 0,

        _errstr       => undef,

        insert_time_callback => ($opts->{insert_time_callback}||sub{
            my ( $sec, $min, $hour, $mday, $mon, $year, undef, undef, undef ) = localtime(time);
            return sprintf('%04d-%02d-%02d %02d:%02d:%02d', $year + 1900, $mon + 1, $mday, $hour, $min, $sec);
        }),

        insert_query => sprintf(
            'INSERT INTO %s (func, arg, enqueue_time, grabbed_until, run_after, retry_cnt, priority) VALUES (?,?,?,0,?,0,?)'
            ,$table_name
        ),

        grab_query => sprintf('UPDATE %s SET grabbed_until = ? WHERE id = ? AND grabbed_until = ?', $table_name),

        lookup_job_query => sprintf('SELECT * FROM %s WHERE id = ? AND grabbed_until <= ? AND run_after <= ?', $table_name),

        find_job_query => sprintf(
            'SELECT * FROM %s WHERE func IN (%s) AND grabbed_until <= ? AND run_after <= ? ORDER BY priority DESC LIMIT %s',
            $table_name, (join(', ', map { "'$_'" } keys %{$functions})), ($opts->{job_find_size} || 50)
        ),

        delete_query => sprintf('DELETE FROM %s WHERE id = ?', $table_name),

        failed_query => sprintf('UPDATE %s SET retry_cnt = retry_cnt + 1, run_after = ?, grabbed_until = 0, priority = ? WHERE id = ?', $table_name),

        unixtime_query => _settled_unixtime_query($driver),

    }, $class;
}

sub _parse_functions {
    my $opts = shift;

    my $functions = $opts->{functions} || [];
    my $default_grab_for = $opts->{default_grab_for} || (60*60);

    my $funcs = +{};
    for (my $i = 0; $i < @{$functions}; $i++) {
        my $func = $functions->[$i];

        my $value;
        if    (not defined $functions->[$i+1]) {$i++                       }
        elsif (ref $functions->[$i+1])         {$value = $functions->[++$i]}

        $value->{grab_for}     ||= $default_grab_for;
        $value->{serializer}   ||= ($opts->{default_serializer}   || sub {$_[0]});
        $value->{deserializer} ||= ($opts->{default_deserializer} || sub {$_[0]});

        $funcs->{$func} = $value;
    }
    $funcs;
}

sub _verify_driver {
    my $dbh = shift;
    my $driver = $dbh->{Driver}{Name};
    $driver =~ /(mysql|SQLite|Pg)/ ? $driver : Carp::croak('Jonk support only mysql,SQLite,Pg');
}

sub _settled_unixtime_query {
    my $driver = shift;

    if ($driver eq 'Pg') {
        return "SELECT TRUNC(EXTRACT('epoch' from NOW()))";
    } elsif ($driver eq 'mysql') {
        return 'SELECT UNIX_TIMESTAMP()';
    }
}

sub errstr {$_[0]->{_errstr}}

sub insert {
    my ($self, $func, $arg, $opt) = @_;

    my $job_id;
    try {
        $self->{_errstr} = undef;
        local $self->{dbh}->{RaiseError} = 1;
        local $self->{dbh}->{PrintError} = 0;

        my $serializer = $self->{functions}->{$func}->{serializer} || sub {$_[0]};
        my $sth = $self->{dbh}->prepare_cached($self->{insert_query});
        $sth->bind_param(1, $func);
        $sth->bind_param(2, $serializer->($arg), _bind_param_attr($self->{driver}));
        $sth->bind_param(3, $self->{insert_time_callback}->());
        $sth->bind_param(4, $opt->{run_after}||0);
        $sth->bind_param(5, $opt->{priority} ||0);
        $sth->execute();

        $job_id = $self->{dbh}->last_insert_id("","",$self->{table_name},"");
        $sth->finish;
    } catch {
        $self->{_errstr} = "can't insert for job queue database: $_"
    };

    $job_id;
}

sub _bind_param_attr {
    my $driver = shift;

    if ( $driver eq 'Pg' ) {
        return { pg_type => DBD::Pg::PG_BYTEA() };
    } elsif ( $driver eq 'SQLite' ) {
        return DBI::SQL_BLOB();
    }
    return;
}

sub _server_unixitime {
    my $self = shift;
    return time() if $self->{driver} eq 'SQLite';
    $self->{dbh}->selectrow_array($self->{unixtime_query});
}

sub _grab_job {
    my ($self, $callback, $opt) = @_;

    my $job;
    try {
        $self->{_errstr} = undef;
        local $self->{dbh}->{RaiseError} = 1;
        local $self->{dbh}->{PrintError} = 0;

        my $time = $self->_server_unixitime;
        my $sth = $callback->($time);

        while (my $row = $sth->fetchrow_hashref) {
            $job = $self->_grab_a_job($row, $time, $opt);
            last if $job;
        }

        $sth->finish;
    } catch {
        $self->{_errstr} = "can't grab job from job queue database: $_";
    };

    $job;

}

sub _grab_a_job {
    my ($self, $row, $time) = @_;

    my $sth = $self->{dbh}->prepare_cached($self->{grab_query});
    $sth->execute(
        ($time + ($self->{functions}->{$row->{func}}->{grab_for})),
        $row->{id},
        $row->{grabbed_until}
    );
    my $grabbed = $sth->rows;
    $sth->finish;
    $grabbed ? Jonk::Job->new($self => $row) : undef;
}

sub lookup_job {
    my ($self, $job_id) = @_;

    $self->_grab_job(
        sub {
            my $time = shift;
            my $sth = $self->{dbh}->prepare_cached($self->{lookup_job_query});
            $sth->execute($job_id, $time, $time);
            $sth;
        }
    );
}

sub find_job {
    my ($self, $opts) = @_;

    unless ($self->{has_func}) {
        Carp::croak('missin find_job functions.');
    }

    $self->_grab_job(
        sub {
            my $time = shift;
            my $sth = $self->{dbh}->prepare_cached($self->{find_job_query});
            $sth->execute($time, $time);
            $sth;
        }
    );
}

sub _delete {
    my ($self, $job_id) = @_;

    try {
        my $sth = $self->{dbh}->prepare_cached($self->{delete_query});
        $sth->execute($job_id);
        $sth->finish;
        return $sth->rows;
    } catch {
        $self->{_errstr} = "can't dequeue job from job queue database: $_";
        return;
    };
}

sub _failed {
    my ($self, $job_id, $opt) = @_;

    my $retry_delay = $self->_server_unixitime + (defined($opt->{retry_delay}) ? $opt->{retry_delay} : 60);
    my $priority    = (defined($opt->{priority}) ? $opt->{priority} : 0);

    try {
        my $sth = $self->{dbh}->prepare_cached($self->{failed_query});
        $sth->execute($retry_delay, $priority, $job_id);
        $sth->finish;
        return $sth->rows;
    } catch {
    warn 'ababaaaba';
        $self->{_errstr} = "can't update job from job queue database: $_";
        return;
    };
}

1;

__END__