File Coverage

blib/lib/Async.pm

Criterion	Covered	Total	%
statement	55	76	72.3
branch	20	26	76.9
condition	0	3	0.0
subroutine	8	13	61.5
pod	0	4	0.0
total	83	122	68.0

use 5.006; use strict; no warnings;

package Async;

our $VERSION = '0.13';

our $ERROR;

sub new {
	my ( $class, $task ) = ( shift, @_ );

	my ( $r, $w );
	unless ( pipe $r, $w ) {
		$ERROR = "Couldn't make pipe: $!";
		return;
	}

	my $pid = fork;
	unless ( defined $pid ) {
		$ERROR = "Couldn't fork: $!";
		return;
	}

	if ( $pid ) { # parent
		close $w;
		my $self = {
			TASK => $task,
			PID  => $pid,
			PPID => $$,
			PIPE => $r,
			FD   => fileno $r,
			DATA => '',
		};
		bless $self, $class;
	} else { # child
		close $r;
		my $result = $task->();
		print $w $result;
		exit 0;
	}
}

# return true iff async process is complete
# with true `$force' argmuent, wait until process is complete before returning
sub ready {
	my ( $self, $force ) = ( shift, @_ );

	my $timeout;
	$timeout = 0 unless $force;

	return 1 if $self->{'FINISHED'};

	my $fdset = '';
	vec( $fdset, $self->{'FD'}, 1 ) = 1;

	while ( select $fdset, undef, undef, $timeout ) {
		my $buf;
		my $nr = read $self->{'PIPE'}, $buf, 8192;
		if ( $nr ) {
			$self->{'DATA'} .= $buf;
		} elsif ( defined $nr ) { # EOF
			$self->{'FINISHED'} = 1;
			return 1;
		} else {
			$self->{'ERROR'} = "Read error: $!";
			$self->{'FINISHED'} = 1;
			return 1;
		}
	}

	return 0;
}

# Return error message if an error occurred
# Return false if no error occurred
sub error { $_[0]{'ERROR'} }

# Return resulting data if async process is complete
# return undef if it is incopmplete
# a true $force argument waits for the process to complete before returning
sub result {
	my ( $self, $force ) = ( shift, @_ );
	if ( $self->{'FINISHED'} ) {
		$self->{'DATA'};
	} elsif ( $force ) {
		$self->ready( $force );
		$self->{'DATA'};
	} else {
		return;
	}
}

sub DESTROY {
	my $self = shift;
	return if $self->{'PPID'} != $$; # created in a different process
	my $pid = $self->{'PID'};
	local ( $., $@, $!, $^E, $? );
	kill 9, $pid; # I don't care.
	waitpid $pid, 0;
}

package AsyncTimeout;
our @ISA = 'Async';

our $VERSION = '0.13';

sub new {
	my ( $class, $task, $timeout, $msg ) = ( shift, @_ );
	$msg = "Timed out\n" unless defined $msg;
	my $newtask = sub {
		local $SIG{'ALRM'} = sub { die "TIMEOUT\n" };
		alarm $timeout;
		my $s = eval { $task->() };
		return $msg if not defined $s and $@ eq "TIMEOUT\n";
		return $s;
	};
	$class->SUPER::new( $newtask );
}

package AsyncData;
our @ISA = 'Async';

our $VERSION = '0.13';

sub new {
	require Storable;
	my ( $class, $task ) = ( shift, @_ );
	my $newtask = sub {
		my $v = $task->();
		return Storable::freeze( $v );
	};
	$class->SUPER::new( $newtask );
}

sub result {
	require Storable;
	my $self = shift;
	my $rc = $self->SUPER::result( @_ );
	return defined $rc ? Storable::thaw( $rc ) : $rc;
}

1;

__END__