package File::Queue; use strict; use IO::File; use Fcntl 'SEEK_END', 'SEEK_SET', 'O_CREAT', 'O_RDWR'; use Carp qw(carp croak); sub new { my $class = shift; my $mi = $class . '->new()'; croak "$mi requires an even number of parameters" if (@_ & 1); my %params = @_; croak "$mi needs an File parameter" unless exists $params{File}; my $queue_file = delete $params{File}; my $idx_file = $queue_file . '.idx'; my $self; my $mode = delete $params{Mode} || '0666'; $self->{block_size} = delete $params{BlockSize} || 64; $self->{delimiter} = delete $params{Delimiter} || "\n"; $self->{del_length} = length $self->{delimiter}; croak "Delimiter length cannot be greater than BlockSize" if $self->{self_length} > $self->{block_size}; $self->{queue} = new IO::File $queue_file, O_CREAT | O_RDWR, $mode or die $!; $self->{idx} = new IO::File $queue_file . '.idx', O_CREAT | O_RDWR, $mode or die $!; $self->{idx}->sysread($self->{ptr}, 1024); if($self->{ptr} > -s $queue_file) { warn "Ptr is greater than queue file size, resetting ptr to '0'"; $self->{idx}->truncate(0) or croak "Could not truncate idx: $!"; $self->{idx}->sysseek(0, SEEK_SET); $self->{idx}->syswrite('0') or croak "Could not syswrite to idx: $!"; } bless $self, $class; return $self; } sub enq { my $self = shift; $self->{queue}->sysseek(0, SEEK_END); if($_[0] =~ s/$self->{delimiter}//g) { warn "Removed illegal delimiter(s) from $_[0]"; } $self->{queue}->syswrite("$_[0]$self->{delimiter}") or croak "Could not syswrite to queue: $!"; } sub deq { my $self = shift; my $item; $self->{queue}->sysseek($self->{ptr}, SEEK_SET); my $i; while($self->{queue}->sysread($_, $self->{block_size})) { $i = index($_, $self->{delimiter}); if($i != -1) { $item .= substr($_, 0, $i); $self->{ptr} += $i + $self->{del_length}; last; } else { ## If delimiter isn't found, go back 'del_length' spaces to ensure we don't miss it between reads $item .= substr($_, 0, -$self->{del_length}, ''); $self->{ptr} += $self->{block_size} - $self->{del_length}; $self->{queue}->sysseek($self->{ptr}, SEEK_SET); } } if($self->{queue}->sysread($_, 1) == 0) { $self->{queue}->truncate(0) or croak "Could not truncate queue: $!"; $self->{queue}->sysseek($self->{ptr} = 0, SEEK_SET); } $self->{idx}->truncate(0) or croak "Could not truncate idx: $!"; $self->{idx}->sysseek(0, SEEK_SET); $self->{idx}->syswrite($self->{ptr}) or croak "Could not syswrite to idx: $!"; return $item; } sub peek { my ($self, $count) = @_; croak "Invalid argument to peek ($count)" unless $count > 0; my $item; $self->{queue}->sysseek($self->{ptr}, SEEK_SET); my (@items, $rem); GATHER: while($self->{queue}->sysread($_, $self->{block_size})) { @items = split /$self->{delimiter}/, $rem . $_, -1; $rem = pop @items; foreach (@items) { push @$item, $_; last GATHER if $count == @$item; } } return $item; } sub reset { my $self = shift; $self->{idx}->truncate(0) or croak "Could not truncate idx: $!"; $self->{idx}->sysseek(0, SEEK_SET); $self->{idx}->syswrite('0') or croak "Could not syswrite to idx: $!"; $self->{queue}->sysseek($self->{ptr} = 0, SEEK_SET); } 1;