File Coverage

blib/lib/IPC/Queue/Duplex.pm

Criterion	Covered	Total	%
statement	52	84	61.9
branch	10	30	33.3
condition	2	12	16.6
subroutine	9	11	81.8
pod	5	5	100.0
total	78	142	54.9

# -*-cperl-*-
#
# IPC::Queue::Duplex - Filesystem based request / response queue
# Copyright (c) 2017 Ashish Gulhati
#
# $Id: lib/IPC/Queue/Duplex.pm v1.008 Mon Mar 12 22:18:20 PDT 2018 $

package IPC::Queue::Duplex;

use warnings;
use strict;
use File::Temp;
use File::Copy qw(cp);
use IPC::Queue::Duplex::Job;
use Fcntl qw(:flock);

our ( $VERSION ) = '$Revision: 1.008 $' =~ /\s+([\d\.]+)/;

sub new {
  my ($class, %args) = @_;
  return unless $args{Dir} and -d $args{Dir};
  bless { Dir => $args{Dir} }, $class;
}

sub add {
  my ($self, $jobstr) = @_;
  return unless $jobstr;
  my $job = File::Temp->new( DIR => $self->{Dir}, SUFFIX => '.iqd');
  print $job "$jobstr\n";
#  $job->unlink_on_destroy(0);
  $job->close; my $filename = $job->filename; $filename =~ s/\.iqd/.job/;
  cp($job->filename,$filename);
  bless { File => $filename }, 'IPC::Queue::Duplex::Job';
}

sub addfile {
  my ($self, $filename, $jobstr) = @_;
  return unless $filename and $jobstr;
  my $filetemp = $filename; $filetemp .= '.iqdtmp';
  open JOBFILE, ">$filetemp";
  print JOBFILE "$jobstr\n";
  close JOBFILE;
  rename $filetemp, $filename;
  bless { File => $filename }, 'IPC::Queue::Duplex::Job';
}

sub get {
  my $self = shift;
  my $filework = '';
  while (1) {
    my $filename; my $oldestage;
    foreach my $file (glob("$self->{Dir}/*.job")) {
      if( !defined($filename) or -M $file > $oldestage ) {
	$filename = $file;
	$oldestage = -M $file;
      }
    }
    last unless $filename;
    next unless -e $filename;
    open(my $jobfh, "+<", $filename);
    close $jobfh, next unless flock($jobfh, LOCK_EX | LOCK_NB);
    $filework = $filename; $filework =~ s/\.job/.wrk/;
    close $jobfh, last if rename $filename,$filework;
  }
  if ($filework) {
    open (JOB, $filework);
    my $jobstr = ; chomp $jobstr;
    close JOB;
    return bless { File => $filework, Request => $jobstr }, 'IPC::Queue::Duplex::Job';
  }
  return undef;
}

sub getresponse {
  my $self = shift;
  my ($filename, $filefin);
  while (1) {
    my $oldestage;
    foreach my $file (glob("$self->{Dir}/*.fin")) {
      if( !defined($filename) or -M $file > $oldestage ) {
	$filename = $file;
	$oldestage = -M $file;
      }
    }
    last unless $filename;
    next unless -e $filename;
    open(my $jobfh, "+<", $filename);
    close $jobfh, next unless flock($jobfh, LOCK_EX | LOCK_NB);
    $filefin = $filename; $filefin =~ s/\.fin/.rsp/;
    close $jobfh, last if rename $filename,$filefin;
  }
  if ($filefin) {
    open (JOBFIN, $filefin);
    my $response = ; chomp $response;
    close JOBFIN; unlink $filefin;
    return bless { File => $filename, Response => $response }, 'IPC::Queue::Duplex::Job';
  }
  return undef;
}

1; # End of IPC::Queue::Duplex

=head1 NAME

IPC::Queue::Duplex - Filesystem based request / response queue

=head1 VERSION

 $Revision: 1.008 $
 $Date: Mon Mar 12 22:18:20 PDT 2018 $

=head1 SYNOPSIS

    (Enqueuer)

    use IPC::Queue::Duplex;

    my $client = new IPC::Queue::Duplex (Dir => $dir);
    my $job = $client->add($jobstr);
    my $response = $job->response;

    (Worker)

    use IPC::Queue::Duplex;

    my $server = new IPC::Queue::Duplex (Dir => $dir);
    my $job = $server->get:
    process_job($job);
    $job->finish($result);

=head1 METHODS

=head2 new

Creates and returns a new IPC::Queue::Duplex object. Requires one
named parameter:

=over

Dir - The directory that will contain the queue for this object. It's
important to use a unique directory per queue,

=back

=head2 add

Adds a job to the queue and returns an IPC::Queue::Duplex::Job
object. A single aregument is required, the job request as a string.

=head2 get

Gets a job from the queue and returns an IPC::Queue::Duplex::Job
object. Returns undef if there is no job waiting.

=head2 addfile

Adds a job to the queue, with an explicitly provided filename, and
returns an IPC::Queue::Duplex::Job object. Two arguments are required:
the filename to use for the job, and the job request string, in that
order.

=head2 getresponse

Get a response from the queue. Normally a requester would hold on to
their job object and get the response by calling the response method
on that object. However, a requester can can also handle responses
asynchronously and call this method to get the next waiting response
instead. Returns an IPC::Queue::Duplex::Job object, or undef if there
is no response on the queue.

=head1 AUTHOR

Ashish Gulhati, C<<  >>

=head1 BUGS

Please report any bugs or feature requests to C, or through
the web interface at L.  I will be notified, and then you'll
automatically be notified of progress on your bug as I make changes.

=head1 SUPPORT

You can find documentation for this module with the perldoc command.

    perldoc IPC::Queue::Duplex

You can also look for information at:

=over 4

=item * RT: CPAN's request tracker

L

=item * AnnoCPAN: Annotated CPAN documentation

L

=item * CPAN Ratings

L

=item * Search CPAN

L

=back

=head1 LICENSE AND COPYRIGHT

Copyright (c) 2017 Ashish Gulhati.

This program is free software; you can redistribute it and/or modify it
under the terms of the Artistic License 2.0.

See L for the full
license terms.