view release on metacpan or  search on metacpan

lib/Perl6/GatherTake.pm  view on Meta::CPAN

use Data::Dumper;
use base 'Exporter';
use Perl6::GatherTake::LazyList;
use Coro;
use Coro::Channel;
use Carp qw(confess);
use Scalar::Util qw(refaddr);
our @EXPORT = qw(gather take);

our %_coro_to_queue;

sub gather(&@) {
    my $code = shift;
    # cheat prototype by prepending '&' to method call:
    my $coro = &async($code, @_);
    my @result = ();
    my $queue = Coro::Channel->new(1);
#    print "Initialized coro $coro\n";
    $_coro_to_queue{refaddr($coro)} = $queue;
    tie @result, 'Perl6::GatherTake::LazyList', $coro, $queue;
    return \@result;
}

sub take {
    my $c = Coro::current;
#    print "Take: $c\n";
    for (@_){
        $_coro_to_queue{refaddr($c)}->put($_);
    }
}

1;

lib/Perl6/GatherTake/LazyList.pm  view on Meta::CPAN

You shouldn't use this module. C<Perl6::GatherTake> does that transparently
for you.

    use Coro;
    use Coro::Channel;
    use Perl6::GatherTake::LazyList;

    my $queue = Coro::Channel->new(1);

    my $coro = async {
        for (1 .. 100){
            my $result;
            # do some heavy computations here
            $queue->put($result);
        }
    };

    my @results;
    tie @results, 'Perl6::GatherTake::LazyList', $coro, $queue;

=head1 DESCRIPTION

Tied array implementation for C<Perl6::GatherTake>. Again: don't use this
yourself unless you really know what you're doing (and you don't).

=head1 LICENSE

Same as C<Perl6::GatherTake>.

lib/Perl6/GatherTake/LazyList.pm  view on Meta::CPAN

our %_ties;

our @ISA;

BEGIN {
    require Tie::Array;
    @ISA = qw(Tie::Array);
}

sub TIEARRAY {
    my ($class, $coro, $queue) = @_;
    my $self = bless {
        coro        => $coro,
        queue       => $queue,
        computed    => [],
        exhausted   => 0,
    }, $class;
    $_ties{$coro} = $self;

    $coro->on_destroy( sub { 
        #print "Exhausted iterator\n";
        $self->{exhausted} = 1 ;
        # this is tricky: the coro will not put another item into
        # the queue when it end, but _compute calls ->get(), thus
        # waits for one - which is a deadlock.
        # so we have to put another value, which _computed will remove
        $self->{queue}->put(undef);
    });

    return $self;
}

sub FETCH {