File Coverage

blib/lib/Pg/Queue.pm

Criterion	Covered	Total	%
statement	15	43	34.8
branch	0	2	0.0
condition	0	3	0.0
subroutine	5	10	50.0
pod	5	5	100.0
total	25	63	39.6

package Pg::Queue;

use strict;
use warnings;

use DBI;
use Moo;

use version; our $VERSION = qv('1.0');

has dbh => (
	is => 'rw',
	required => 1,
);

has queuename => (
	is => 'rw',
	default => sub { "queuetest" },
);


sub create_queue_table {
	my( $self ) = @_;
	my $dbh = $self->dbh;
	my $name = $self->queuename;

	$dbh->do("DROP TABLE IF EXISTS $name");

	$dbh->do(<
	CREATE TABLE $name (
		id SERIAL PRIMARY KEY,
		available BOOLEAN DEFAULT TRUE,
		processed TIMESTAMP,
		item TEXT
	)
SQL

	$dbh->do("CREATE INDEX ${name}_available_idx ON $name (id) WHERE available");
}

sub add_work_item {
	my( $self, $item ) = @_;

	$self->dbh->do("INSERT INTO ${\$self->queuename} (item) VALUES (?)",undef, $item);
}

sub pull_work_item {
	my( $self, $callback ) = @_;
	my $dbh = $self->dbh;
	my $name = $self->queuename;

	$dbh->do("BEGIN");

	my $row = $dbh->selectrow_arrayref(
		"UPDATE $name SET available=false
		WHERE id = (
			SELECT id FROM $name
			WHERE available
			ORDER BY id
			LIMIT 1
			FOR UPDATE SKIP LOCKED
		) RETURNING id,item"
	);

	if( $row and $callback->(@$row) ) {
		$dbh->do("UPDATE $name SET processed=NOW() where id = ?", undef, $row->[0]);
		$dbh->do("COMMIT");
		return 1; #TRUE
	}
	else {
		$dbh->do("ROLLBACK");
		return 0; #FALSE
	}
}

sub count_total {
	my( $self ) = @_;

	my $row = $self->dbh->selectrow_arrayref( "SELECT count(*) FROM ${\$self->queuename}" );

	return $row->[0];
}

sub count_available {
	my( $self ) = @_;

	my $row = $self->dbh->selectrow_arrayref( "SELECT count(*) FROM ${\$self->queuename} WHERE available" );

	return $row->[0];
}


1;
__END__