File Coverage

blib/lib/Data/Queue/Persistent.pm

Criterion	Covered	Total	%
statement	15	135	11.1
branch	0	64	0.0
condition	0	21	0.0
subroutine	5	24	20.8
pod	7	18	38.8
total	27	262	10.3

package Data::Queue::Persistent;

use 5.008004;
use strict;
use warnings;
use Carp qw / croak /;
use DBI;

our $VERSION = '0.13';

our $schema = q{
    CREATE TABLE %s (
                     qkey VARCHAR(255) NOT NULL,
                     idx INTEGER UNSIGNED NOT NULL,
                     value BLOB,
                     PRIMARY KEY (qkey, idx)
                     )
    };

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

    my $dsn = delete $opts{dsn};
    my $dbh = delete $opts{dbh};

    croak "No DSN or database handle passed to Data::Queue::Persistent->new"
        unless $dsn || $dbh;

    my $username = delete $opts{username};
    my $pass = delete $opts{pass};

    my $cache = delete $opts{cache} || 0;

    my $key = delete $opts{id} or croak "No queue id defined";

    my $table = delete $opts{table} || 'persistent_queue';

    my $noload = delete $opts{noload};
    my $max_size = delete $opts{max_size};

    # connect to db
    if ($dsn) {
        $dbh = DBI->connect($dsn, $username, $pass)
            or croak "Could not connect to database";
    }

    my $self = {
        cache      => $cache,
        dbh        => $dbh,
        q          => [],
        key        => $key,
        table_name => $table,
        max_size   => $max_size,
    };

    bless $self, $class;
    $self->init;
    $self->load if $self->caching && ! $noload;

    return $self;
}

sub table_name {
    my $self = shift;
    return $self->dbh->quote_identifier($self->{table_name});
}

sub dbh { $_[0]->{dbh} }
sub key { $_[0]->{key} }
sub q { $_[0]->{q} }
sub caching { $_[0]->{cache} }
sub max_size { $_[0]->{max_size} }

# returns how many items are in the queue
sub length {
    my $self = CORE::shift();

    return (scalar @{$self->{q}}) if $self->caching;

    my $table = $self->table_name;
    my ($length) = $self->dbh->selectrow_array("SELECT COUNT(idx) FROM $table WHERE qkey=?",
                                               undef, $self->key);
    die $self->dbh->errstr if $self->dbh->err;

    return $length || 0;
}

sub _max_idx {
    my $self = CORE::shift();

    # TODO: cache max index

    my $table = $self->table_name;
    my ($idx) = $self->dbh->selectrow_array("SELECT MAX(idx) FROM $table WHERE qkey=?",
                                               undef, $self->key);
    die $self->dbh->errstr if $self->dbh->err;

    return defined $idx ? $idx + 1 : 0;
}

# do a sql statement and die if it fails
sub do {
    my ($self, $sql, @vals) = @_;

    $self->dbh->do($sql, undef, @vals);
    croak $self->dbh->errstr if $self->dbh->err;
}

# initialize the storage
sub init {
    my ($self) = @_;

    croak "No table name defined" unless $self->table_name;

    # don't do anything if table already exists
    return if $self->table_exists;

    # table doesn't exist, create it
    my $sql = sprintf($schema, $self->table_name);
    $self->do($sql);
}

# load data from db
sub load {
    my $self = CORE::shift();

    my $table = $self->table_name or croak "No table name defined";
    die "Table $table does not exist." unless $self->table_exists;

    my $rows = $self->dbh->selectall_arrayref("SELECT value FROM $table WHERE qkey=? ORDER BY idx", undef, $self->key);
    die $self->dbh->errstr if $self->dbh->err;

    return unless $rows && @$rows;
    $self->absorb_rows(@$rows);
}

sub absorb_rows {
    my ($self, @rows) = @_;
    push @{$self->{q}}, map { $_->[0] } @rows;
}

# delete everything from the queue
sub empty {
    my ($self) = @_;

    my $table = $self->table_name;
    $self->do("DELETE FROM $table WHERE qkey=?", $self->key);

    $self->{q} = [] if $self->caching;
}

sub table_exists {
    my $self = CORE::shift();
    # get table info, see if our table exists
    my @tables = $self->dbh->tables(undef, undef, $self->{table_name}, "TABLE");
    my $table = $self->{table_name};

    $table = $self->dbh->quote_identifier($self->{table_name})
        if $self->dbh->get_info(29); # quote if the db driver uses table name quoting

    return grep { $_ =~ m/$table$/ } @tables;
}

# add @vals to the queue
*add = \&unshift;
sub unshift {
    my ($self, @vals) = @_;

    my $idx = $self->_max_idx;

    my $key = $self->dbh->quote($self->key);
    my $table = $self->table_name;
    my $dbh = $self->dbh;

    $dbh->begin_work;
    my $sth = $dbh->prepare(qq[ INSERT INTO $table (qkey, idx, value) VALUES ($key, ?, ?) ]);

    foreach my $val (@vals) {
        push @{$self->{q}}, $val if $self->caching;

        $sth->execute($idx++, $val);
        if ($dbh->err) {
            die $dbh->errstr;
            $dbh->rollback;
        }
    }

    $dbh->commit;

    # truncate queue to max_size
    my $max_size = $self->max_size;
    my $length = $self->length;
    $self->shift($length - $max_size) if defined $max_size && $length > $max_size;
}

# shift $count elements off the queue
*remove = \&shift;
sub shift {
    my ($self, $_count) = @_;

    my $count = defined $_count ? $_count : 1;
    $count += 0;

    if ($self->caching) {
        CORE::shift(@{$self->{q}}) for 1 .. $count;
    }

    my $table = $self->table_name;

    # begin transaction
    $self->dbh->begin_work;

    # get $count elements
    my $rows = $self->dbh->selectall_arrayref("SELECT idx, value FROM $table WHERE qkey = ? ORDER BY idx LIMIT $count",
                                              undef, $self->key);
    die $self->dbh->errstr if $self->dbh->err;

    my @idx = map { $_->[0] } @$rows;
    my @vals = map { $_->[1] } @$rows;

    return () unless @vals;

    # remove the retreived elements
    my $bindstr = join(',', map { '?' } @idx);
    $self->do("DELETE FROM $table WHERE qkey=? AND idx BETWEEN ? AND ?", $self->key, $idx[0], $idx[-1]);

    # commit transaction
    $self->dbh->commit;

    # return first element if no $count defined, otherwise return array of values
    return $vals[0] unless defined $_count;
    return @vals;
}

# retreive elements at an index
sub get {
    my ($self, $offset, $length, %opts) = @_;

    return $self->all unless defined $offset || defined $length;

    $length = -1 unless defined $length; # need to specify a limit when selecting an offset, this is wack
    $offset += 0;

    my $direction = $opts{reverse} ? "DESC" : '';

    my $table = $self->table_name;
    my $rows = $self->dbh->selectcol_arrayref("SELECT value FROM $table WHERE qkey = ? ORDER BY idx $direction LIMIT $length OFFSET $offset",
                                              undef, $self->key);
    die $self->dbh->errstr if $self->dbh->err;

    return wantarray ? @$rows : $rows->[0];
}

# returns all elements of the queue
sub all {
    my $self = CORE::shift();

    return @{$self->{q}} if $self->caching;

    my $valsref = $self->dbh->selectall_arrayref("SELECT value FROM " . $self->table_name . " WHERE qkey = ?",
                                                 undef, $self->key);
    return map { @$_ } @$valsref;
}

1;

__END__