File Coverage

blib/lib/Directory/Queue/Simple.pm

Criterion	Covered	Total	%
statement	128	161	79.5
branch	29	70	41.4
condition	3	21	14.2
subroutine	23	27	85.1
pod	13	13	100.0
total	196	292	67.1

#+##############################################################################
#                                                                              #
# File: Directory/Queue/Simple.pm                                              #
#                                                                              #
# Description: object oriented interface to a simple directory based queue     #
#                                                                              #
#-##############################################################################

#
# module definition
#

package Directory::Queue::Simple;
use strict;
use warnings;
our $VERSION  = "2.1";
our $REVISION = sprintf("%d.%02d", q$Revision: 1.19 $ =~ /(\d+)\.(\d+)/);

#
# used modules
#

use Directory::Queue qw(_create _name _touch SYSBUFSIZE /Regexp/ /special/);
use No::Worries::Die qw(dief);
use No::Worries::File qw(file_read file_write);
use No::Worries::Stat qw(ST_MTIME);
use No::Worries::Warn qw(warnf);
use POSIX qw(:errno_h);

#
# inheritance
#

our(@ISA) = qw(Directory::Queue);

#
# constants
#

# suffix indicating a temporary element
use constant TEMPORARY_SUFFIX => ".tmp";

# suffix indicating a locked element
use constant LOCKED_SUFFIX => ".lck";

#
# object constructor
#

sub new : method {
    my($class, %option) = @_;
    my($self);

    # default object
    $self = __PACKAGE__->SUPER::_new(%option);
    foreach my $name (qw(path maxlock maxtemp rndhex umask)) {
        delete($option{$name});
    }
    # check granularity
    if (defined($option{granularity})) {
        dief("invalid granularity: %s", $option{granularity})
            unless $option{granularity} =~ /^\d+$/;
        $self->{granularity} = delete($option{granularity});
    } else {
        $self->{granularity} = 60; # default
    }
    # check unexpected options
    foreach my $name (keys(%option)) {
        dief("unexpected option: %s", $name);
    }
    # so far so good...
    return($self);
}

#
# helpers for the add methods
#

sub _add_dir ($) {
    my($self) = @_;
    my($time);

    $time = time();
    $time -= $time % $self->{granularity} if $self->{granularity};
    return(sprintf("%08x", $time));
}

sub _add_data ($$) {
    my($self, $dataref) = @_;
    my($dir, $name, $path, $fh);

    $dir = _add_dir($self);
    while (1) {
        $name = _name($self->{rndhex});
        $path = $self->{path}."/".$dir."/".$name . TEMPORARY_SUFFIX;
        $fh = _create($path, $self->{umask});
        last if $fh;
        _special_mkdir($self->{path}."/".$dir, $self->{umask})
            if $! == ENOENT;
    }
    file_write($path, handle => $fh, data => $dataref);
    return($dir, $path);
}

sub _add_path ($$$) {
    my($self, $tmp, $dir) = @_;
    my($name, $new);

    while (1) {
        $name = _name($self->{rndhex});
        $new = $self->{path}."/".$dir."/".$name;
        # N.B. we use link() + unlink() to make sure $new is never overwritten
        if (link($tmp, $new)) {
            unlink($tmp) or dief("cannot unlink(%s): %s", $tmp, $!);
            return($dir."/".$name);
        }
        dief("cannot link(%s, %s): %s", $tmp, $new, $!) unless $! == EEXIST;
    }
}

#
# add a new element to the queue and return its name
#

sub add : method {
    my($self, $data) = @_;
    my($dir, $path);

    ($dir, $path) = _add_data($self, \$data);
    return(_add_path($self, $path, $dir));
}

sub add_ref : method {
    my($self, $dataref) = @_;
    my($dir, $path);

    ($dir, $path) = _add_data($self, $dataref);
    return(_add_path($self, $path, $dir));
}

sub add_path : method {
    my($self, $path) = @_;
    my($dir);

    $dir = _add_dir($self);
    _special_mkdir($self->{path}."/".$dir, $self->{umask});
    return(_add_path($self, $path, $dir));
}

#
# get a locked element
#

sub get : method {
    my($self, $name) = @_;

    return(file_read($self->{path}."/".$name . LOCKED_SUFFIX));
}

sub get_ref : method {
    my($self, $name) = @_;
    my($data);

    return(file_read($self->{path}."/".$name . LOCKED_SUFFIX, data => \$data));
}

sub get_path : method {
    my($self, $name) = @_;

    return($self->{path}."/".$name . LOCKED_SUFFIX);
}

#
# lock an element:
#  - return true on success
#  - return false in case the element could not be locked (in permissive mode)
#

sub lock : method {  ## no critic 'ProhibitBuiltinHomonyms'
    my($self, $name, $permissive) = @_;
    my($path, $lock, $time, $ignored);

    $permissive = 1 unless defined($permissive);
    $path = $self->{path}."/".$name;
    $lock = $path . LOCKED_SUFFIX;
    unless (link($path, $lock)) {
        return(0) if $permissive and ($! == EEXIST or $! == ENOENT);
        dief("cannot link(%s, %s): %s", $path, $lock, $!);
    }
    # we also touch the element to indicate the lock time
    $time = time();
    unless (utime($time, $time, $path)) {
        if ($permissive and $! == ENOENT) {
            # RACE: the element file does not exist anymore
            # (this can happen if an other process locked & removed the element
            #  while our link() was in progress... yes, this can happen!
            #  we do our best and ignore what unlink() returns)
            $ignored = unlink($lock);
            return(0);
        }
        # otherwise this is unexpected...
        dief("cannot utime(%d, %d, %s): %s", $time, $time, $path, $!);
    }
    # so far so good
    return(1);
}

#
# unlock an element:
#  - return true on success
#  - return false in case the element could not be unlocked (in permissive mode)
#

sub unlock : method {
    my($self, $name, $permissive) = @_;
    my($path, $lock);

    $permissive = 0 unless defined($permissive);
    $path = $self->{path}."/".$name;
    $lock = $path . LOCKED_SUFFIX;
    return(1) if unlink($lock);
    return(0) if $permissive and $! == ENOENT;
    dief("cannot unlink(%s): %s", $lock, $!);
}

#
# touch an element to indicate that it is still being used
#

sub touch : method {
    my($self, $element) = @_;

    _touch($self->{"path"}."/".$element);
}

#
# remove a locked element from the queue
#

sub remove : method {
    my($self, $name) = @_;
    my($path, $lock);

    $path = $self->{path}."/".$name;
    $lock = $path . LOCKED_SUFFIX;
    unlink($path) or dief("cannot unlink(%s): %s", $path, $!);
    unlink($lock) or dief("cannot unlink(%s): %s", $lock, $!);
}

#
# return the number of elements in the queue, locked or not (but not temporary)
#

sub count : method {
    my($self) = @_;
    my($count, @list);

    $count = 0;
    # get the list of directories
    foreach my $name (_special_getdir($self->{path}, "strict")) {
        push(@list, $1) if $name =~ /^($_DirectoryRegexp)$/o; # untaint
    }
    # count the elements inside
    foreach my $name (@list) {
        $count += grep(/^(?:$_ElementRegexp)$/o,
                       _special_getdir($self->{path}."/".$name));
    }
    # that's all
    return($count);
}

#
# purge an intermediate directory
#

sub _purge_dir ($$$) {
    my($dir, $oldtemp, $oldlock) = @_;
    my($path, @stat);

    foreach my $name (grep(/\./, _special_getdir($dir))) {
        $path = $dir."/".$name;
        @stat = stat($path);
        unless (@stat) {
            dief("cannot stat(%s): %s", $path, $!) unless $! == ENOENT;
            next;
        }
        next if substr($name, -4) eq TEMPORARY_SUFFIX
            and $stat[ST_MTIME] >= $oldtemp;
        next if substr($name, -4) eq LOCKED_SUFFIX
            and $stat[ST_MTIME] >= $oldlock;
        warnf("removing too old volatile file: %s", $path);
        next if unlink($path);
        dief("cannot unlink(%s): %s", $path, $!) unless $! == ENOENT;
    }
}

#
# purge the queue
#

sub purge : method {
    my($self, %option) = @_;
    my(@list, $path, $oldtemp, $oldlock);

    # check options
    $option{maxtemp} = $self->{maxtemp} unless defined($option{maxtemp});
    $option{maxlock} = $self->{maxtemp} unless defined($option{maxlock});
    foreach my $name (keys(%option)) {
        dief("unexpected option: %s", $name)
            unless $name =~ /^(maxtemp|maxlock)$/;
        dief("invalid %s: %s", $name, $option{$name})
            unless $option{$name} =~ /^\d+$/;
    }
    # get the list of intermediate directories
    @list = ();
    foreach my $name (_special_getdir($self->{path}, "strict")) {
        push(@list, $1) if $name =~ /^($_DirectoryRegexp)$/o; # untaint
    }
    # remove the old temporary or locked elements
    $oldtemp = $oldlock = 0;
    $oldtemp = time() - $option{maxtemp} if $option{maxtemp};
    $oldlock = time() - $option{maxlock} if $option{maxlock};
    if ($oldtemp or $oldlock) {
        foreach my $name (@list) {
            _purge_dir($self->{path}."/".$name, $oldtemp, $oldlock);
        }
    }
    # try to purge all but the last intermediate directory
    if (@list > 1) {
        @list = sort(@list);
        pop(@list);
        foreach my $name (@list) {
            $path = $self->{path}."/".$name;
            _special_rmdir($path) unless _special_getdir($path);
        }
    }
}

1;

__END__