File Coverage

blib/lib/IO/Lambda.pm

Criterion	Covered	Total	%
statement	660	750	88.0
branch	257	424	60.6
condition	49	102	48.0
subroutine	121	145	83.4
pod	60	77	77.9
total	1147	1498	76.5

# $Id: Lambda.pm,v 1.191 2012/01/13 06:41:28 dk Exp $
package IO::Lambda;

use Carp qw(croak);
use strict;
use warnings;
use Exporter;
use Sub::Name;
use Scalar::Util qw(weaken);
use Time::HiRes qw(time);
use vars qw(
	$LOOP %EVENTS @LOOPS
	$VERSION @ISA
	@EXPORT_OK %EXPORT_TAGS	@EXPORT_CONSTANTS @EXPORT_LAMBDA @EXPORT_STREAM
	@EXPORT_DEV @EXPORT_MISC @EXPORT_FUNC
	$THIS @CONTEXT $METHOD $CALLBACK $AGAIN $SIGTHROW
	$DEBUG_IO $DEBUG_LAMBDA $DEBUG_CALLER %DEBUG
);
$VERSION     = '1.24';
@ISA         = qw(Exporter);
@EXPORT_CONSTANTS = qw(
	IO_READ IO_WRITE IO_EXCEPTION
	WATCH_OBJ WATCH_DEADLINE WATCH_LAMBDA WATCH_CALLBACK
	WATCH_IO_HANDLE WATCH_IO_FLAGS WATCH_CALLER WATCH_CANCEL
);
@EXPORT_STREAM = qw(
	sysreader syswriter getline readbuf writebuf
);
@EXPORT_LAMBDA = qw(
	this context lambda again state restartable catch autocatch
	io readable writable rwx timeout tail tails tailo any_tail
);
@EXPORT_FUNC = qw(
	seq par mapcar filter fold curry
);
@EXPORT_MISC    = qw(
	set_frame get_frame swap_frame sigthrow
);
@EXPORT_DEV    = qw(
	_subname _o _t
);
@EXPORT_OK   = (
	@EXPORT_LAMBDA, @EXPORT_CONSTANTS, @EXPORT_STREAM,
	@EXPORT_DEV, @EXPORT_MISC, @EXPORT_FUNC
);
%EXPORT_TAGS = (
	func      => \@EXPORT_FUNC,
	lambda    => \@EXPORT_LAMBDA,
	stream    => \@EXPORT_STREAM,
	constants => \@EXPORT_CONSTANTS,
	dev       => \@EXPORT_DEV,
	all       => [ @EXPORT_LAMBDA, @EXPORT_STREAM, @EXPORT_CONSTANTS, @EXPORT_FUNC ],
);

if ( exists $ENV{IO_LAMBDA_DEBUG}) {
	for my $p ( split ',', $ENV{IO_LAMBDA_DEBUG}) {
		if ( $p =~ /^([^=]+)=(.*)$/) {
			$DEBUG{lc $1}=$2;
		} else {
			$DEBUG{lc $p}++;
		}
	}
	$DEBUG_IO     = $DEBUG{io}     || 0;
	$DEBUG_LAMBDA = $DEBUG{lambda} || 0;
	$DEBUG_CALLER = $DEBUG{caller} || 0;
	$IO::Lambda::Loop::DEFAULT = $DEBUG{loop} if $DEBUG{loop};
	$SIG{__DIE__} = sub {
		return if $^S;
		Carp::confess(@_);
	} if $DEBUG{die};
}

use constant IO_READ         => 4;
use constant IO_WRITE        => 2;
use constant IO_EXCEPTION    => 1;

use constant WATCH_OBJ       => 0;
use constant WATCH_CANCEL    => 1;

use constant WATCH_DEADLINE  => 2;
use constant WATCH_LAMBDA    => 2;
use constant WATCH_CALLBACK  => 3;

use constant WATCH_CALLER    => 4;
use constant WATCH_IO_HANDLE => 4;
use constant WATCH_IO_FLAGS  => 5;

sub new
{
	IO::Lambda::Loop-> new unless $LOOP;
	return bless {
		in      => [],    # events we wait for
		last    => [],    # result of the last state
		stopped => 0,     # initial state
		start   => $_[1], # kick-start coderef
	}, $_[0];
}

sub DESTROY
{
	my $self = $_[0];
	$self-> cancel_all_events;
}

my  $_doffs = 0;
sub _d_in  { $_doffs++ }
sub _d_out { $_doffs-- if $_doffs }
sub _d     { ('  ' x $_doffs), _obj(shift), ': ', @_, "\n" }
sub _o     { $_[0] =~ /0x([\w]+)/; $1 }
sub _obj   { "lambda(". _o($_[0]) . ")." . ( $_[0]->{caller} || '()' ) }
sub _t     { defined($_[0]) ? ( "time(", (($_[0] < 1_000_000) ? $_[0] : $_[0]-time()), ")" ) : () }
sub _ev
{
	$_[0] =~ /0x([\w]+)/;
	"event($1) ",
	(($#{$_[0]} == WATCH_IO_FLAGS) ? (
		'fd=',
		fileno($_[0]->[WATCH_IO_HANDLE]),
		' ',
		( $_[0]->[WATCH_IO_FLAGS] ? (
			join('/',
				(($_[0]->[WATCH_IO_FLAGS] & IO_READ)      ? 'read'  : ()),
				(($_[0]->[WATCH_IO_FLAGS] & IO_WRITE)     ? 'write' : ()),
				(($_[0]->[WATCH_IO_FLAGS] & IO_EXCEPTION) ? 'exc'   : ()),
			)) :
			'timeout'
		),
		' ', _t($_[0]->[WATCH_DEADLINE]),
	) : (
		ref($_[0]-> [WATCH_LAMBDA]) ?
			_obj($_[0]-> [WATCH_LAMBDA]) :
			_t($_[0]->[WATCH_DEADLINE])
	))
}
sub _msg
{
	my $self = shift;
	_d(
		$self,
		"@_ >> (",
		join(',',
			map {
				defined($_) ? $_ : 'undef'
			}
			@{$self->{last}}
		),
		')'
	)
}

#
# Part I - Object interface to callback and
# messaging interface with event loop and lambdas
#
#########################################################

# register an IO event
sub watch_io
{
	my ( $self, $flags, $handle, $deadline, $callback, $cancel) = @_;

	croak "can't register events on a stopped lambda" if $self-> {stopped};
	croak "bad io flags" if 0 == ($flags & (IO_READ|IO_WRITE|IO_EXCEPTION));

	$deadline += time if defined($deadline) and $deadline < 1_000_000_000;

	my $rec = [
		$self,
		$cancel,
		$deadline,
		$callback,
		$handle,
		$flags,
	];
	weaken $rec->[0];
	push @{$self-> {in}}, $rec;

	warn _d( $self, "> ", _ev($rec)) if $DEBUG_IO;

	$LOOP-> watch( $rec );

	return $rec;
}

# register a timeout
sub watch_timer
{
	my ( $self, $deadline, $callback, $cancel) = @_;

	croak "can't register events on a stopped lambda" if $self-> {stopped};
	croak "$self: time is undefined" unless defined $deadline;

	$deadline += time if $deadline < 1_000_000_000;
	my $rec = [
		$self,
		$cancel,
		$deadline,
		$callback,
	];
	weaken $rec->[0];
	push @{$self-> {in}}, $rec;

	warn _d( $self, "> ", _ev($rec)) if $DEBUG_IO;

	$LOOP-> after( $rec);

	return $rec;
}

# register a callback when another lambda exits
sub watch_lambda
{
	my ( $self, $lambda, $callback, $cancel) = @_;
	@_ = (); # perl bug http://rt.perl.org/rt3//Public/Bug/Display.html?id=70974

	croak "can't register events on a stopped lambda" if $self-> {stopped};
	croak "bad lambda" unless $lambda and $lambda->isa('IO::Lambda');

	croak "won't watch myself" if $self == $lambda;
	# XXX check cycling

	$lambda-> reset if $lambda-> is_stopped;

	my $rec = [
		$self,
		$cancel,
		$lambda,
		$callback,
	];
	weaken $rec->[0];
	$rec-> [WATCH_CALLER] = Carp::shortmess if $DEBUG_CALLER;
	push @{$self-> {in}}, $rec;
	push @{$EVENTS{"$lambda"}}, $rec;

	$lambda-> start if $lambda-> is_passive;

	warn _d( $self, "> ", _ev($rec)) if $DEBUG_LAMBDA;

	return $rec;
}

# watch the watchers
sub override
{
	my ( $self, $method, $state, $cb) = ( 4 == @_) ? @_ : (@_[0,1],'*',$_[2]);

	if ( $cb) {
		$self-> {override}->{$method} ||= [];
		push @{$self-> {override}->{$method}}, [ $state, $cb ];
	} else {
		my $p;
		return unless $p = $self-> {override}->{$method};
		for ( my $i = $#$p; $i >= 0; $i--) {
			if (
				(
					not defined ($state) and
					not defined ($p->[$i]-> [0])
				) or (
					defined($state) and
					defined($p->[$i]-> [0]) and
					$p->[$i]->[0] eq $state
				)
			) {
				my $ret = splice( @$p, $i, 1);
				delete $self-> {override}->{$method} unless @$p;
				return $ret->[1];
			}
		}

		return undef;
	}
}

sub override_handler
{
	my ( $self, $method, $sub, $cb) = @_;

	my $o = $self-> {override}-> {$method}-> [-1];

	# check state match
	my ($a, $b) = ( $self-> {state}, $o-> [0]);
	unless (
		( not defined($a) and not defined ($b)) or
		( defined $a and defined $b and $a eq $b) or
		( defined $b and $b eq '*')
	) {
		# state not matched
		if ( 1 == @{$self-> {override}->{$method}}) {
			local $self-> {override}->{$method} = undef;
			return $sub-> ($cb);
		} else {
			pop @{$self-> {override}->{$method}};
			my $ret = $sub-> ($cb);
			push @{$self->{override}->{$method}}, $o;
			return $ret;
		}
	} else {
		# state matched
		local $self-> {super} = [ $sub, $cb ];
		if ( 1 == @{$self-> {override}->{$method}}) {
			local $self-> {override}->{$method} = undef;
			return $o-> [1]-> ( $self, $sub, $cb);
		} else {
			pop @{$self-> {override}->{$method}};
			my $ret = $o-> [1]-> ( $self, $sub, $cb);
			push @{$self->{override}->{$method}}, $o;
			return $ret;
		}
	}
}

# Insert a new callback to be called before original callback.
# Needs to insert callbacks in {override} stack in reverse order,
# because direct order serves LIFO order for override() callbacks, --
# and that means FIFO for intercept() callbacks. But we also want LIFO.
sub intercept
{
	my ( $self, $method, $state, $cb) = ( 4 == @_) ? @_ : (@_[0,1],'*',$_[2]);

	return $self-> override( $method, $state, undef) unless $cb;

	_subname("intercept($method:$state)" => $cb);

	$self-> {override}->{$method} ||= [];
	unshift @{$self-> {override}->{$method}}, [ $state, sub {
		# this is called when lambda calls $method with $state
		my ( undef, $sub, $orig_cb) = @_;
		# $sub is a condition, like readable(&) or tail(&)
		$sub->( sub {
		# that (&) is finally called when IO event is there
			local $self-> {super} = [$orig_cb];
			&$cb;
		});
	} ];
}

sub super
{
	croak "super() call outside overridden condition" unless $_[0]-> {super};
	my $data = $_[0]-> {super};
	if ( defined $data-> [1]) {
		# override() super
		return $data-> [0]-> ($data-> [1]);
	} else {
		# intercept() super
		my $self = shift;
		return defined($data->[0]) ?
			$data-> [0]-> (@_) :
			( wantarray ? @_ : $_[0] );
	}
}


# handle incoming asynchronous events
sub io_handler
{
	my ( $self, $rec) = @_;

	warn _d( $self, '< ', _ev($rec)) if $DEBUG_IO;

	my $in = $self-> {in};
	my $nn = @$in;
	@$in = grep { $rec != $_ } @$in;
	die _d($self, 'stray ', _ev($rec))
		if $nn == @$in or $self != $rec->[WATCH_OBJ];

	_d_in if $DEBUG_IO;
	local $self-> {cancel} = $rec-> [WATCH_CANCEL];
	@{$self->{last}} = $rec-> [WATCH_CALLBACK]-> (
		$self,
		(($#$rec == WATCH_IO_FLAGS) ? $rec-> [WATCH_IO_FLAGS] : ()),
		@{$self->{last}}
	) if $rec-> [WATCH_CALLBACK];

	_d_out if $DEBUG_IO;
	warn $self-> _msg('io') if $DEBUG_IO;

	unless ( @$in) {
		warn _d( $self, 'stopped') if $DEBUG_LAMBDA;
		$self-> {stopped}++;
	}
}

# handle incoming synchronous events
sub lambda_handler
{
	my ( $self, $rec) = @_;

	warn _d( $self, '< ', _ev($rec)) if $DEBUG_LAMBDA;

	my $in = $self-> {in};
	my $nn = @$in;
	@$in = grep { $rec != $_ } @$in;
	die _d($self, 'stray ', _ev($rec))
		if $nn == @$in or $self != $rec->[WATCH_OBJ];

	my $lambda = $rec-> [WATCH_LAMBDA];
	die _d($self,
		'handler called but ', _obj($lambda),
		' is not finished yet') unless $lambda-> {stopped};

	my $arr = $EVENTS{"$lambda"};
	@$arr = grep { $_ != $rec } @$arr;
	delete $EVENTS{"$lambda"} unless @$arr;

	_d_in if $DEBUG_LAMBDA;

	local $self-> {cancel} = $rec-> [WATCH_CANCEL];
	@{$self->{last}} =
		$rec-> [WATCH_CALLBACK] ?
			$rec-> [WATCH_CALLBACK]-> (
				$self,
				@{$rec-> [WATCH_LAMBDA]-> {last}}
			) :
			@{$rec-> [WATCH_LAMBDA]-> {last}};

	_d_out if $DEBUG_LAMBDA;
	warn $self-> _msg('tail') if $DEBUG_LAMBDA;

	unless ( @$in) {
		warn _d( $self, 'stopped') if $DEBUG_LAMBDA;
		$self-> {stopped} = 1;
	}
}

# Removes one event from queue
sub cancel_event
{
	my ( $self, $rec) = @_;

	return unless @{$self-> {in}};

	@{$self->{last}} = $rec-> [WATCH_CANCEL]->($self, @{$self->{last}})
		if $rec-> [WATCH_CANCEL];

	$LOOP-> remove_event($rec) if $LOOP;
	@{$self-> {in}} = grep { $_ != $rec } @{$self-> {in}};

	if ($rec->[WATCH_LAMBDA] and ref($rec->[WATCH_LAMBDA])) {
		my $arr = $EVENTS{$rec->[WATCH_LAMBDA]};
		if ( $arr) {
			@$arr = grep { $_ != $rec } @$arr;
			delete $EVENTS{$rec->[WATCH_LAMBDA]} unless @$arr;
		}
	}
	@$rec = ();

	return if @{$self->{in}};
	# that was the last event

	warn _d( $self, 'stopped') if $DEBUG_LAMBDA;
	$self-> {stopped} = 1;
	$_-> remove( $self) for @LOOPS;
}

# Removes all events bound to the object, notifies the interested objects.
# The object becomes stopped immediately, so no new events will be allowed to register.
sub cancel_all_events
{
	my $self = shift;

	$self-> {stopped} = 1;

	return unless @{$self-> {in}};

	for ( grep { $_-> [WATCH_CANCEL] } reverse @{$self-> {in}}) {
		my $wc = $_-> [WATCH_CANCEL];
		$_-> [WATCH_CANCEL] = undef;
		@{$self->{last}} = $wc-> ($self, $_, @{$self->{last}})
	}

	$LOOP-> remove( $self) if $LOOP;
	$_-> remove($self) for @LOOPS;

	for my $rec ( @{$self->{in}}) {
		if ( ref($rec->[WATCH_LAMBDA])) {
			my $arr = $EVENTS{$rec->[WATCH_LAMBDA]};
			if ( $arr) {
				@$arr = grep { $_ != $rec } @$arr;
				delete $EVENTS{$rec->[WATCH_LAMBDA]} unless @$arr;
			}
		}
		@$rec = ();
	}

	@{$self-> {in}} = ();
}

sub autorestart
{
	$#_ ?
		$_[0]-> {autorestart} = $_[1] :
		( exists($_[0]-> {autorestart}) ?
			$_[0]-> {autorestart} : 1)
}
sub is_stopped  { $_[0]-> {stopped}  }
sub is_waiting  { not($_[0]->{stopped}) and @{$_[0]->{in}} }
sub is_passive  { not($_[0]->{stopped}) and not(@{$_[0]->{in}}) }
sub is_active   { $_[0]->{stopped} or @{$_[0]->{in}} }

# reset the state machine
sub reset
{
	my $self = shift;

	$self-> cancel_all_events;
	@{$self-> {last}} = ();
	delete $self-> {stopped};
	warn _d( $self, 'reset') if $DEBUG_LAMBDA;
}

# start the state machine
sub start
{
	my $self = shift;

	croak "can't start active lambda, call reset() first" if $self-> is_active;

	warn _d( $self, 'started') if $DEBUG_LAMBDA;
	@{$self->{last}} = $self-> {start}-> ($self, @{$self->{last}})
		if $self-> {start};
	warn $self-> _msg('initial') if $DEBUG_LAMBDA;

	unless ( @{$self->{in}}) {
		warn _d( $self, 'stopped') if $DEBUG_LAMBDA;
		$self-> {stopped} = 1;
	}
}

# peek into the current state
sub peek { wantarray ? @{$_[0]->{last}} : $_[0]-> {last}-> [0] }

# pass initial parameters to lambda
sub call
{
	my $self = shift;

	croak "can't call active lambda" if $self-> is_active;

	@{$self-> {last}} = @_;
	$self;
}

# abandon all states and stop with constant message
sub terminate
{
	my ( $self, @error) = @_;

	$self-> {last} = \@error;
	$self-> cancel_all_events;
	warn $self-> _msg('terminate') if $DEBUG_LAMBDA;
}

# propagate event destruction on all levels
sub destroy
{
	shift-> cancel_all_events( cascade => 1);
}

# synchronisation

# drives objects dependant on the other objects until all of them
# are stopped
sub drive
{
	my $changed = 1;
	my $executed = 0;
	warn "IO::Lambda::drive --------\n" if $DEBUG_LAMBDA;
	while ( $changed) {
		$changed = 0;

		# dispatch
		for my $rec ( map { @$_ } values %EVENTS) {
			next unless $rec->[WATCH_LAMBDA]-> {stopped};
			$rec->[WATCH_OBJ]-> lambda_handler( $rec);
			$changed = 1;
			$executed++;
		}
	warn "IO::Lambda::drive .........\n" if $DEBUG_LAMBDA and $changed;
	}
	warn "IO::Lambda::drive +++++++++\n" if $DEBUG_LAMBDA;

	return $executed;
}

# do one quant
sub yield
{
	my $nonblocking = shift;
	my $more_events = 0;

	# custom loops must not wait
	for ( @LOOPS) {
		next if $_-> empty;
		$_-> yield;
		$more_events = 1;
	}

	if ( drive) {
		# some callbacks we called, don't let them wait in sleep
		return 1;
	}

	# main loop waits, if anything
	unless ( $LOOP-> empty) {
		$LOOP-> yield( $nonblocking);
		$more_events = 1;
	}

	$more_events = 1 if keys %EVENTS;
	return $more_events;
}


# wait for one lambda to stop
sub wait
{
	my $self = shift;
	if ( $self-> is_passive) {
		$self-> call(@_);
		$self-> start;
	}
	yield while not $self-> {stopped};
	return $self-> peek;
}

# wait for all lambdas to stop
sub wait_for_all
{
	my @objects = @_;
	return unless @objects;
	$_-> start for grep { $_-> is_passive } @objects;
	my @ret;
	while ( 1) {
		push @ret, map { $_-> peek } grep { $_-> {stopped} } @objects;
		@objects = grep { not $_-> {stopped} } @objects;
		last unless @objects;
		yield;
	}
	return @ret;
}

# wait for at least one lambda to stop, return those that stopped
sub wait_for_any
{
	my @objects = @_;
	return unless @objects;
	$_-> start for grep { $_-> is_passive } @objects;
	while ( 1) {
		my @n = grep { $_-> {stopped} } @objects;
		return @n if @n;
		yield;
	}
}

# run the event loop until no lambdas are left in the blocking state
sub run { do {} while yield }

#
# Part II - Procedural interface to the lambda-style programming
#
#################################################################

sub _lambda_restart { die "lambda() is not restartable" }
sub lambda(&)
{
	my $cb  = _subname(lambda => $_[0]);
	my $l   = __PACKAGE__-> new( sub {
		# initial lambda code is usually executed by tail/tails inside another lambda,
		# so protect the upper-level context
		local *__ANON__ = "IO::Lambda::lambda::callback";
		local $THIS     = shift;
		local @CONTEXT  = ();
		local $CALLBACK = $cb;
		local $METHOD   = \&_lambda_restart;
		$cb ? $cb-> (@_) : @_;
	});
	if ( $DEBUG_CALLER) {
		if ( $DEBUG_CALLER > 1) {
			$l-> {caller} = Carp::longmess;
			chomp $l-> {caller};
			$l-> {caller} =~ s/^ at //;
		} else {
			$l-> {caller} = join(':', (caller)[1,2]);
		}
	}
	$l;
}

sub _subname
{
	subname(
		caller(1 + ($_[2] || 0)) .  '::_'.  $_[0],
		$_[1]
	) if $DEBUG_CALLER and $_[1] and not $AGAIN;
	return $_[1];
}

*io = \λ

# re-enter the latest (or other) frame
sub again
{
	( $METHOD, $CALLBACK) = @_ if 2 == @_;
	local $AGAIN = 1;
	defined($METHOD) ?
		$METHOD-> ($CALLBACK) :
		croak "again() outside of a restartable call"
}

# define context
sub this        { @_ ? ($THIS, @CONTEXT)    = @_ : $THIS }
sub context     { @_ ? (@CONTEXT)           = @_ : @CONTEXT }
sub restartable { @_ ? ($METHOD, $CALLBACK) = @_ : ( $METHOD, $CALLBACK) }
sub set_frame   { ( $THIS, $METHOD, $CALLBACK, @CONTEXT) = @_ }
sub get_frame   { ( $THIS, $METHOD, $CALLBACK, @CONTEXT) }
sub swap_frame  { my @f = get_frame; set_frame(@_); @f }
sub clear       { set_frame(); undef $AGAIN; }

END { ( $THIS, $METHOD, $CALLBACK, @CONTEXT) = (); }

sub state($)
{
	my $this = ($_[0] && ref($_[0])) ? shift(@_) : this;
	@_ ? $this-> {state} = $_[0] : return $this-> {state};
}

# exceptions and backtracing
sub catch(&$)
{
	my ( $cb, $event) = @_;
	my $who = (caller(1))[3];
	my @ctx = @CONTEXT;
	croak "catch callback already defined" if $event-> [WATCH_CANCEL];
	$event->[WATCH_CANCEL] = $cb ? sub {
		local *__ANON__ = "$who\:\:catch" if $DEBUG_CALLER;
		$THIS     = shift;
		local $THIS-> {cancelled_event} = shift;
		local $THIS-> {cancelling} = 1;
		@CONTEXT  = @ctx;
		$METHOD   = undef;
		$CALLBACK = undef;
		$cb-> (@_);
	} : undef;

	# if throw() happened before we even get here
	$event->[WATCH_CALLBACK] = $event->[WATCH_CANCEL]
		if $event->[WATCH_CALLBACK] && $event->[WATCH_CALLBACK] == \&_throw;

	return $event;
}

sub call_again
{
	my $self = shift;
	croak "called outside catch()" unless $self-> {cancelled_event};
	my $cb = $self-> {cancelled_event}->[WATCH_CALLBACK];
	$cb->($self, @_) if $cb;
}

sub autocatch($)
{
	catch {
		this-> call_again;
		this-> throw(@_);
	} $_[0]
}

sub is_cancelling { $_[0]-> {cancelling} }

sub _throw
{
	my $self = shift;
	warn _d( $self, 'throw') if $DEBUG_LAMBDA;
	$self-> throw(@_);
}

sub throw
{
	my ( $self, @error) = @_;
	my @c = $self-> callers;
	$_-> [WATCH_CALLBACK] = $_->[WATCH_CANCEL] || \&_throw for @c;
	$self-> terminate(@error);
	$SIGTHROW->($self, @error) if $SIGTHROW and not @c;
	return @error;
}

sub sigthrow
{
	shift if defined($_[0]) and (not(ref $_[0]) or ref($_[0]) ne 'CODE');
	$SIGTHROW = $_[0] if @_;
	return $SIGTHROW;
}

sub callees      { @{ $EVENTS{ "$_[0]" } || [] } }
sub callers      { grep { $_[0] == $_-> [WATCH_LAMBDA] } map { @$_ } values %EVENTS }

sub backtrace
{
	require IO::Lambda::Backtrace;
	IO::Lambda::Backtrace-> new($_[0], Carp::shortmess);
}

#
# Conditions:
#

# common wrapper for declaration of handle-watching user conditions
sub add_watch
{
	my ($self, $cb, $method, $flags, $handle, $deadline, @ctx) = @_;
	my $who;
	$who = (caller(1))[3] if $DEBUG_CALLER;
	$self-> watch_io(
		$flags, $handle, $deadline,
		sub {
			local *__ANON__ = "$who\:\:callback" if $DEBUG_CALLER;
			$THIS     = shift;
			@CONTEXT  = @ctx;
			$METHOD   = $method;
			$CALLBACK = $cb;
			$cb ? $cb-> (@_) : @_;
		},
		($AGAIN ? delete($self-> {cancel}) : undef),
	)
}

# rwx($flags,$handle,$deadline)
sub rwx(&)
{
	return $THIS-> override_handler('rwx', \&rwx, shift)
		if $THIS-> {override}->{rwx};

	$THIS-> add_watch(
		_subname(rwx => shift), \&rwx,
		@CONTEXT[0,1,2,0,1,2]
	)
}

# readable($handle,$deadline)
sub readable(&)
{
	return $THIS-> override_handler('readable', \&readable, shift)
		if $THIS-> {override}->{readable};

	$THIS-> add_watch(
		_subname(readable => shift), \&readable, IO_READ,
		@CONTEXT[0,1,0,1]
	)
}

# writable($handle,$deadline)
sub writable(&)
{
	return $THIS-> override_handler('writable', \&writable, shift)
		if $THIS-> {override}->{writable};

	$THIS-> add_watch(
		_subname(writable => shift), \&writable, IO_WRITE,
		@CONTEXT[0,1,0,1]
	)
}

# common wrapper for declaration of time-watching user conditions
sub add_timer
{
	my ($self, $cb, $method, $deadline, @ctx) = @_;
	my $who;
	$who = (caller(1))[3] if $DEBUG_CALLER;
	$self-> watch_timer(
		$deadline,
		sub {
			local *__ANON__ = "$who\:\:callback" if $DEBUG_CALLER;
			$THIS     = shift;
			@CONTEXT  = @ctx;
			$METHOD   = $method;
			$CALLBACK = $cb;
			$cb ? $cb-> (@_) : @_;
		},
		($AGAIN ? delete($self-> {cancel}) : undef),
	)
}

# timeout($deadline)
sub timeout(&)
{
	return $THIS-> override_handler('timeout', \&timeout, shift)
		if $THIS-> {override}->{timeout};
	$THIS-> add_timer( _subname(timeout => shift), \&timeout, @CONTEXT[0,0])
}

# common wrapper for declaration of single lambda-watching user conditions
sub add_tail
{
	my ($self, $cb, $method, $lambda, @ctx) = @_;
	my $who;
	$who = (caller(1))[3] if $DEBUG_CALLER;
	$self-> watch_lambda(
		$lambda,
		($cb ? sub {
			local *__ANON__ = "$who\:\:callback" if $DEBUG_CALLER;
			$THIS     = shift;
			@CONTEXT  = @ctx;
			$METHOD   = $method;
			$CALLBACK = $cb;
			$cb-> (@_);
		} : undef),
		($AGAIN ? delete($self-> {cancel}) : undef),
	);
}

# convert constant @param into a lambda
sub add_constant
{
	my ( $self, $cb, $method, @param) = @_;
	$self-> add_tail (
		_subname(constant => $cb), $method,
		lambda { @param },
		@CONTEXT
	);
}

# handle default condition logic given a lambda
sub condition
{
	my ( $self, $cb, $method, $name) = @_;

	return $THIS-> override_handler($name, $method, $cb)
		if defined($name) and $THIS-> {override}->{$name};

	my @ctx = @CONTEXT;

	my $who;
	if ( $DEBUG_CALLER) {
		$who = defined($name) ? $name : (caller(1))[3];
		_subname($who, $cb, 2);
	}

	$THIS-> watch_lambda(
		$self,
		$cb ? sub {
			local *__ANON__ = "$who\:\:callback" if $DEBUG_CALLER;
			$THIS     = shift;
			@CONTEXT  = @ctx;
			$METHOD   = $method;
			$CALLBACK = $cb;
			$cb-> (@_);
		} : undef
	);
}

# dummy sub for empty calls for tails() family
sub _empty
{
	my ($name, $method, $cb) = @_;
	my @ctx = context;
	$THIS-> watch_lambda( IO::Lambda-> new, sub {
		local *__ANON__ = "IO::Lambda::".$name."::callback";
		@CONTEXT  = @ctx;
		$METHOD   = $method;
		$CALLBACK = $cb;
		$cb-> ();
	}) if $cb;
}

# tail( $lambda, @param) -- initialize $lambda with @param, and wait for it
sub tail(&)
{
	return $THIS-> override_handler('tail', \&tail, shift)
		if $THIS-> {override}->{tail};

	my ( $lambda, @param) = context;
	return _empty(tail => \&tail, shift) unless $lambda;

	$lambda-> reset
		if $lambda-> is_stopped and $lambda-> autorestart;
	if ( @param) {
		$lambda-> call( @param);
	} else {
		$lambda-> call unless $lambda-> is_active;
	}
	$THIS-> add_tail( _subname(tail => shift), \&tail, $lambda, $lambda, @param);
}


# tails(@lambdas) -- wait for all lambdas to finish
sub tails(&)
{
	return $THIS-> override_handler('tails', \&tails, shift)
		if $THIS-> {override}->{tails};

	my $cb = _subname tails => $_[0];
	my @lambdas = context;
	my $n = $#lambdas;
	return _empty(tails => \&tails, $cb) unless @lambdas;

	my @ret;
	my $watcher;
	$watcher = sub {
		$THIS     = shift;
		push @ret, @_;
		return if $n--;

		local *__ANON__ = "IO::Lambda::tails::callback";
		@CONTEXT  = @lambdas;
		$METHOD   = \&tails;
		$CALLBACK = $cb;
		$cb ? $cb-> (@ret) : @ret;
	};
	my $this = $THIS;
	$this-> watch_lambda( $_, $watcher) for @lambdas;
}

# tailo(@lambdas) -- wait for all lambdas to finish, return ordered results
sub tailo(&)
{
	return $THIS-> override_handler('tailo', \&tailo, shift)
		if $THIS-> {override}->{tailo};

	my $cb = _subname tailo => $_[0];
	my @lambdas = context;
	my $n = $#lambdas;
	return _empty(tailo => \&tailo, $cb) unless @lambdas;

	my @ret;
	my $watcher;
	$watcher = sub {
		my $curr  = shift;
		$THIS     = shift;
		$ret[ $curr ] = \@_;
		return if $n--;

		local *__ANON__ = "IO::Lambda::tailo::callback";
		@CONTEXT  = @lambdas;
		$METHOD   = \&tailo;
		$CALLBACK = $cb;
		@ret = map { @$_ } @ret;
		$cb ? $cb-> (@ret) : @ret;
	};
	my $this = $THIS;
	for ( my $i = 0; $i < @lambdas; $i++) {
		my $d = $i;
		$this-> watch_lambda(
			$lambdas[$i],
			sub { $watcher->($d, @_) }
		);
	};
}

# any_tail($deadline,@lambdas) -- wait for any lambda to finish within time
sub any_tail(&)
{
	return $THIS-> override_handler('any_tail', \&any_tail, shift)
		if $THIS-> {override}->{any_tail};

	my $cb = _subname any_tail => $_[0];
	my ( $deadline, @lambdas) = context;
	my $n = $#lambdas;
	return _empty(any_tail => \&any_tail, $cb) unless @lambdas;

	my ( @ret, @watchers);
	my $timer;

	$timer = $THIS-> watch_timer( $deadline, sub {
		local *__ANON__ = "IO::Lambda::any_tail::callback1";
		$THIS     = shift;
		@CONTEXT  = ($deadline, @lambdas);
		$METHOD   = \&any_tail;
		$CALLBACK = $cb;
		@ret = $cb-> (@ret) if $cb;
		$THIS-> cancel_event($_) for @watchers;
		return @ret;
	}) if defined $deadline;

	my $watcher;
	$watcher = sub {
		push @ret, shift;
		return if $n--;

		local *__ANON__ = "IO::Lambda::any_tail::callback2";
		$THIS = shift;
		@CONTEXT  = ($deadline, @lambdas);
		$METHOD   = \&any_tail;
		$CALLBACK = $cb;
		@ret = $cb-> (@ret) if $cb;
		$THIS-> cancel_event( $timer) if $timer;
		return @ret;
	};

	@watchers = map {
		my $l = $_;
		$THIS-> watch_lambda( $l, sub {
			$watcher->($l, @_);
		})
	} @lambdas;
}

#
# Part III - High order lambdas
#
################################################################

# fold($l) :: @b -> @c
# $l :: ($a,@b) -> @c
sub fold($)
{
	my $l = shift;
	lambda {
		my @q = @_;
		context $l, shift(@q), shift(@q);
	tail {
		return @_ unless @q;
		context $l, shift(@q), @_;
		again;
	}}
}

# mapcar($l) :: (@p) -> @r
sub mapcar($)
{
	my $lambda = shift;
	lambda {
		my @ret;
		my @p = @_;
		return unless @p;
		context $lambda, shift @p;
	tail {
		push @ret, @_;
		return @ret unless @p;
		context $lambda, shift @p;
		again;
	}}
}

# filter($l) :: (@p) -> @r
sub filter($)
{
	my $lambda = shift;
	lambda {
		my @ret;
		return unless @_;
		my @p = @_;
		my $p = shift @p;
		context $lambda, $p;
	tail {
		push @ret, $p if shift;
		return @ret unless @p;
		$p = shift @p;
		context $lambda, $p;
		again;
	}}
}

# curry(@a -> $l) :: @a -> @b
sub curry(&)
{
	my $cb = $_[0];
	lambda {
		context $cb->(@_);
		&tail();
	}
}

# seq() :: (@l) -> @m
sub seq { mapcar curry { shift } }

# par($max = 0) :: (@l) -> @m
sub par
{
	my $max = $_[0] || 0;

	lambda {
		my @q = @_;
		my @ret;
		$max = @q if $max < 1 or $max > @q;
		context map {
			lambda {
				return unless @q;
				context shift @q;
			tail {
				push @ret, @_;
				return unless @q;
				context shift @q;
				again;
			}}
		} 1 .. $max;
		tails { @ret }
	}
}


# sysread lambda wrapper
#
# ioresult    :: ($result, $error)
# sysreader() :: ($fh, $buf, $length, $deadline) -> ioresult
sub sysreader (){ lambda
{
	my ( $fh, $buf, $length, $deadline) = @_;
	$$buf = '' unless defined $$buf;

	this-> watch_io( IO_READ, $fh, $deadline, subname _sysreader => sub {
		return undef, 'timeout' unless $_[1];
                local $SIG{PIPE} = 'IGNORE';
		my $n = sysread( $fh, $$buf, $length, length($$buf));
		if ( $DEBUG_IO) {
			warn "fh(", fileno($fh), ") read ", ( defined($n) ? "$n bytes" : "error $!"), "\n";
			warn substr( $$buf, length($$buf) - $n), "\n" if $DEBUG_IO > 1 and $n > 0;
		}
		return undef, $! unless defined $n;
		return $n;
	})
}}

# syswrite() lambda wrapper
#
# syswriter() :: ($fh, $buf, $length, $offset, $deadline) -> ioresult
sub syswriter (){ lambda
{
	my ( $fh, $buf, $length, $offset, $deadline) = @_;

	this-> watch_io( IO_WRITE, $fh, $deadline, subname _syswriter => sub {
		return undef, 'timeout' unless $_[1];
                local $SIG{PIPE} = 'IGNORE';
		my $n = syswrite( $fh, $$buf, $length, $offset);
		if ( $DEBUG_IO) {
			warn "fh(", fileno($fh), ") wrote ", ( defined($n) ? "$n bytes out of $length" : "error $!"), "\n";
			warn substr( $$buf, $offset, $n), "\n" if $DEBUG_IO > 1 and $n > 0;
		}
		return undef, $! unless defined $n;
		return $n;
	});
}}

sub _match
{
	my ( $cond, $buf) = @_;

	return unless defined $cond;

	return ($$buf =~ /($cond)/)[0] if ref($cond) eq 'Regexp';
	return $cond->($buf) if ref($cond) eq 'CODE';
	return length($$buf) >= $cond;
}

# read from stream until condition is met
#
# readbuf($reader) :: ($fh, $$buf, $cond, $deadline) -> ioresult
sub readbuf
{
	my $reader = shift || sysreader;

	lambda {
		my ( $fh, $buf, $cond, $deadline) = @_;

		$$buf = "" unless defined $$buf;

		my $match = _match( $cond, $buf);
		return $match if $match;

		my ($maxbytes, $bufsize);
		$maxbytes = $cond - length($$buf)
			if defined($cond) and not ref($cond) and $cond > length($$buf);
		$bufsize = defined($maxbytes) ? $maxbytes : 65536;

		my $savepos = pos($$buf); # useful when $cond is a regexp

		context $reader, $fh, $buf, $bufsize, $deadline;
	tail {
		pos($$buf) = $savepos;

		my $bytes = shift;
		return undef, shift unless defined $bytes;

		unless ( $bytes) {
			return 1 unless defined $cond;
			return undef, 'eof';
		}

		# got line? return it
		my $match = _match( $cond, $buf);
		return $match if $match;

		# otherwise, just wait for more data
		$bufsize -= $bytes if defined $maxbytes;

		context $reader, $fh, $buf, $bufsize, $deadline;
		again;
	}}
}

# curry readbuf()
#
# getline($reader) :: ($fh, $$buf, $deadline) -> ioresult
sub getline
{
	my $reader = shift;
	lambda {
		my ( $fh, $buf, $deadline) = @_;
		croak "getline() needs a buffer! ( f.ex getline,\$fh,\\(my \$buf='') )"
			unless ref($buf);
		context readbuf($reader), $fh, $buf, qr/^[^\n]*\n/, $deadline;
	tail {
		substr( $$buf, 0, length($_[0]), '') unless defined $_[1];
		@_;
	}}
}

# write whole buffer to stream
#
# writebuf($writer) :: syswriter
sub writebuf
{
	my $writer = shift || syswriter;

	lambda {
		my ( $fh, $buf, $len, $offs, $deadline) = @_;

		my ( $written, $recheck_length, $olen) = (0);
		$$buf = "" unless defined $$buf;
		$offs = 0 unless defined $offs;
		unless ( defined $len) {
			$olen = $len = length $$buf;
			$recheck_length++;
		}

		context $writer, $fh, $buf, $len, $offs, $deadline;
	tail {
		my $bytes = shift;
		return undef, shift unless defined $bytes;

		$offs    += $bytes;
		$written += $bytes;
		$len     -= $bytes;
		if ( $recheck_length) {
			my $l = length $$buf;
			if ( $l > $olen) {
				$len  += $l - $olen;
				$olen = $l;
			}
		}
		return $written if $len <= 0;

		context $writer, $fh, $buf, $len, $offs, $deadline;
		again;
	}}
}

#
# Part IV - Developer API for custom condvars and event loops
#
################################################################

# register condvar listener
sub bind
{
	my $self = shift;

	# create new condition
	croak "can't register events on a stopped lambda" if $self-> {stopped};

	my $rec = [ $self, @_ ];
	push @{$self-> {in}}, $rec;

	return $rec;
}

# stop listening on a condvar
sub resolve
{
	my ( $self, $rec) = @_;

	my $in = $self-> {in};
	my $nn = @$in;
	@$in = grep { $rec != $_ } @$in;
	die _d($self, "stray condvar event $rec (@$rec)")
		if $nn == @$in or $self != $rec->[WATCH_OBJ];

	undef $rec-> [WATCH_OBJ]; # unneeded references

	unless ( @$in) {
		warn _d( $self, 'stopped') if $DEBUG_LAMBDA;
		$self-> {stopped} = 1;
	}
}

sub callout
{
	my ( $self, $cb, @param) = @_;
	@{$self->{last}} = $cb ? $cb-> (@param) : @param;
}

sub add_loop     { push @LOOPS, shift }
sub remove_loop  { @LOOPS = grep { defined and $_ != $_[0] } @LOOPS }

sub __end        { clear(); undef %EVENTS; undef @LOOPS; } # for threads

package IO::Lambda::Loop;
use vars qw($DEFAULT);
use strict;
use warnings;

$DEFAULT = 'Select' unless defined $DEFAULT;
sub default { $DEFAULT = shift }

sub new
{
	return $IO::Lambda::LOOP if $IO::Lambda::LOOP;

	my ( $class, %opt) = @_;

	$opt{type} ||= $DEFAULT;
	$class .= "::$opt{type}";
	eval "use $class;";
	die $@ if $@;

	return $IO::Lambda::LOOP = $class-> new();
}

1;

__DATA__