IO::Lambda::Poll - emulate asynchronous behavior by polling

    use IO::Lambda qw(:lambda);
    use IO::Lambda::Poll qw(poller);

    lambda {
       context 
          poller { check_if_ready }, 
	  timeout   => 5,
	  frequency => 0.1;
    tail {
       print shift() ? "ok\n" : "timeout\n";
    }}

# $Id: Poll.pm,v 1.7 2009/11/30 13:25:06 dk Exp $
package IO::Lambda::Poll;
use vars qw(
	@ISA @EXPORT_OK %EXPORT_TAGS 
	$DEBUG @RECORDS @TIMER $TIMER_ACTIVE $MASTER
);

$DEBUG = $IO::Lambda::DEBUG{poll} || 0;
@ISA = qw(Exporter);
@EXPORT_OK  = qw(poll_event poll_cancel poller);
%EXPORT_TAGS = ( all => \@EXPORT_OK);

use strict;
use warnings;
use Time::HiRes qw(time);
use IO::Lambda qw(:all :dev set_frame get_frame);

$MASTER = bless {}, __PACKAGE__;

# register yield handler
IO::Lambda::add_loop($MASTER);
END {
	@RECORDS = ();
	IO::Lambda::remove_loop($MASTER);
};

# There'll also be a single timer as we need timeouts
$TIMER[WATCH_OBJ] = bless {}, "IO::Lambda::Poll::Timer";
sub IO::Lambda::Poll::Timer::io_handler
{
	warn "poll.timer < expired\n" if $DEBUG;
	$TIMER_ACTIVE = 0;
}

sub empty { 0 == @RECORDS }

sub remove
{
	my $lambda = $_[1];
	my $n = @RECORDS;
	@RECORDS = grep { $_-> {this} ne $lambda } @RECORDS;
	return if $n == @RECORDS;
	warn "poll.remove $lambda\n" if $DEBUG;
	reset_timer();
}

sub yield
{
	warn "poll.yield\n" if $DEBUG > 1;
	my $time = time;

	my @new;
	my @frame = get_frame;
	for my $rec ( @RECORDS) {
		my ( $ok, @result) = $rec-> {poller}-> (
			defined($rec->{deadline}) && $rec->{deadline} <= $time,
			@{ $rec-> {param}}
		);
		unless ($ok) {
			push @new, $rec;
			next;
		}
		warn "poll.resolve($rec)\n" if $DEBUG;
		my $this = $rec-> {this};
		$this-> set_frame($rec-> {method}, $rec->{callback}, @{ $rec->{context} });
		$this-> callout( $rec-> {callback}, @result);
		$this-> resolve( $rec-> {bind});
	}
	set_frame(@frame);
	return if @RECORDS == @new;

	@RECORDS = @new;
	reset_timer();
}

sub reset_timer
{
	my ( $expires, $frequency);
	for my $rec (@RECORDS) {
		my ($f,$d) = @{$rec}{qw(frequency deadline)};
		$frequency = $f if not defined($frequency) or (defined($f) and $frequency > $f);
		$expires   = $d if not defined($expires)   or (defined($d) and $expires   > $d); 
	}

	if ( defined $frequency) {
		$frequency += time;
		if ( defined $expires) {
			$expires = $frequency if $expires > $frequency;
		} elsif ( @RECORDS) {
			$expires = $frequency;
		}
	}

	if ( defined $expires) {
		if ( $TIMER_ACTIVE) {
			if ( abs( $expires - $TIMER[WATCH_DEADLINE]) > 0.001) {
				# restart the active timer
				warn "poll.timer > restart $expires/$TIMER[WATCH_DEADLINE]\n"
					if $DEBUG;
				$IO::Lambda::LOOP-> remove_event( \@TIMER);
				$TIMER[WATCH_DEADLINE] = $expires;
				$IO::Lambda::LOOP-> after( \@TIMER);
			}
			# else, same timeout, on already active timer - do nothing
		} else {
			# resubmit
			warn "poll.timer > submit $expires\n" if $DEBUG;
			$TIMER[WATCH_DEADLINE] = $expires;
			$IO::Lambda::LOOP-> after( \@TIMER);
			$TIMER_ACTIVE = 1;
		}
	} elsif ( $TIMER_ACTIVE) {
		warn "poll.timer > stop\n" if $DEBUG;
		# stop timer
		$IO::Lambda::LOOP-> remove_event( \@TIMER);
		$TIMER_ACTIVE = 0;
	}
}

sub poll_event
{
	my ( $cb, $method, $poller, $deadline, $frequency, @param ) = @_;

	$deadline += time if defined($deadline) and $deadline < 1_000_000_000;
	
	push @RECORDS, {
		this      => this,
		bind      => this-> bind,
		method    => $method,
		callback  => $cb,
		context   => [ context ],
		poller    => $poller,
		deadline  => $deadline,
		param     => \@param,
		frequency => $frequency,
	};

	reset_timer;
	warn "poll.new($RECORDS[-1]) on ", this, "\n" if $DEBUG;

	return $RECORDS[-1];
}

# don't call this, use lambda-> cancel_event( $record->{bind} )
sub poll_cancel
{
	my $rec = shift;
	my $n = @RECORDS;
	@RECORDS = grep { $rec != $_ } @RECORDS;
	return if $n == @RECORDS;
	warn "poll.cancel($rec)\n" if $DEBUG;
	reset_timer;
}

sub poll_handler
{
	my ( $expired, $cb, @opt) = @_;
	my @res = $cb->(@opt);
	return 1,@res if $res[0];
	return 1,0 if $expired;
	return 0;
}

sub poller(&)
{
	my $cb = _subname poller => shift;

	lambda {
		my %opt = @_;
		poll_event(
			undef, undef, \&poll_handler, 
			exists($opt{timeout}) ? $opt{timeout} : $opt{deadline},
			$opt{frequency}, 
			$cb, %opt
		);
	}
}

1;

__DATA__