view release on metacpan or  search on metacpan

lib/Alvis/Pipeline/Read.pm  view on Meta::CPAN

# $Id: Read.pm,v 1.12 2006/08/14 17:24:04 mike Exp $

package Alvis::Pipeline::Read;
use vars qw(@ISA);
@ISA = qw(Alvis::Pipeline);

use strict;
use warnings;
use IO::File;
use IO::Socket::INET;
use Fcntl qw(:flock);


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

    my $this = bless {}, $class;
    $this->{spooldir} = delete $opts{spooldir}
	or die "new($class) with no spooldir";
    $this->{port} = delete $opts{port}
	or die "new($class) with no port";

    $this->_setopts(sleep => 10, %opts);

    # Asynchronicity: server process accepts pushes and stores them
    $this->log(1, "forking");
    my $pid = fork();
    die "couldn't fork: $!" if !defined $pid;
    if ($pid == 0) {
	# Child process
	$this->_start_server();
	die "_start_server() returned!  It should never do that";
    }

    # Automatic reaper prevents the child going zombie when we kill
    # it.  (Yes, "IGNORE" has a special-case meaning for SIGCHLD.)
    $SIG{CHLD} = 'IGNORE';

    $this->{pid} = $pid;
    $this->log(1, "parent $$ spawned harvesting child, pid=$pid");
    return $this;
}


sub read {
    my $this = shift();
    my($block) = @_;

    $this->log(2, "parent reading from spooldir");
    my $dir = $this->{spooldir};
    my($fh, $lastread, $lastwrite) = $this->_lock_and_read($dir);
    while ($lastread == $lastwrite) {
	$this->_write_and_unlock($dir, $fh, $lastread, $lastwrite);
	return undef if !$block;
	$this->log(2, "no document yet, sleeping");
	sleep $this->option("sleep");
	($fh, $lastread, $lastwrite) = $this->_lock_and_read($dir);
    }

    $lastread++;
    my $filename = "$dir/$lastread";
    my $f2 = new IO::File("<$filename")
	or die "can't read file '$filename': $!";
    binmode $f2, ":utf8";
    my $doc = join("", <$f2>);
    $f2->close();
    unlink($filename);

    $this->_write_and_unlock($dir, $fh, $lastread, $lastwrite);
    return $doc;
}


sub close {
    my $this = shift();

    # We need to kill the child process that is running the OAI server
    # so that its Internet port is cleared for subsequent invocations.
    # Also so that the parent can exit cleanly.
    my $pid = $this->{pid};

    kill 15, $pid;
    sleep 1;

    if (kill 0, $pid) {