Port of when.rb

when.pl

package Promise;
use strict;
use warnings;

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

    bless {
        deferred => $deferred,
    }, $class;
}

sub then {
    my ($self, $cb) = @_;
    $self->{deferred}->then($cb);
}

sub callback {
    my ($self, $cb) = @_;
    $self->{deferred}->callback($cb);
}

sub errback {
    my ($self, $cb) = @_;
    $self->{deferred}->errback($cb);
}

1;

package Resolver;
use strict;
use warnings;

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

    bless {
        deferred => $deferred,
    }, $class;
}

sub resolve {
    my $self = shift;
    $self->{deferred}->resolve(@_);
}

sub reject {
    my $self = shift;
    $self->{deferred}->reject(@_);
}

sub is_resolved {
    my $self = shift;
    $self->{deferred}->resolved;
}

1;

package Deferred;
use strict;
use warnings;

sub new {
    my $class = shift;

    my $self = bless {
        resolution   => [],
        callbacks    => {
            resolved => [],
            rejected => [],
        },
    }, $class;

    $self->{resolver} = Resolver->new($self);
    $self->{promise}  = Promise->new($self);

    $self;
}

sub resolve {
    my $self = shift;
    $self->_mark_resolved('resolved', @_);
}

sub reject {
    my $self = shift;
    $self->_mark_resolved('rejected', @_);
}

sub callback {
    my ($self, $cb) = @_;
    $self->_add_callback(resolved => $cb);
}

sub then {
    my ($self, $cb) = @_;
    $self->_add_callback(resolved => $cb);
}

sub errback {
    my ($self, $cb) = @_;
    $self->_add_callback(rejected => $cb);
}

sub is_resolved {
    my $self = shift;
    scalar @{$self->{resolution}} != 0;
}

sub _add_callback {
    my ($self, $type, $cb) = @_;
    return $self->_notify_callbacks({$type => $cb}) if $self->is_resolved;
    push @{$self->{callbacks}->{$type}}, $cb;
}

sub _mark_resolved {
    my ($self, $state, @args) = @_;
    die "Already resolved" if $self->is_resolved;

    $self->{resolution} = [$state, \@args];
    $self->_notify_callbacks($self->{callbacks});
}

sub _notify_callbacks {
    my ($self, $callbacks) = @_;
    my $type = $self->{resolution}->[0];
    my $blocks = $callbacks->{$type} || [];

    for my $block (@{$blocks}) {
        my $args_ref = $self->{resolution}->[1];
        $block->(@{$args_ref});
    }
}

# accessor
sub resolver { $_[0]->{resolver} }
sub promise  { $_[0]->{promise}  }

1;

package When;
use strict;
use warnings;

sub defer {
    my ($cb) = @_;

    my $deferred = Deferred->new;
    $cb->($deferred) if defined $cb;
    return $deferred;
}

sub resolve {
    my $val = shift;
    my $deferred = Deferred->new;
    $deferred->resolve($val);
    return $deferred->promise;
}

sub all {
    my @promises = @_;

    my $resolved = 0;
    my @results;

    my $deferred = Deferred->new;
    my $attempt_resolution = sub {
        my ($err, $results_ref) = @_;
        return if $deferred->is_resolved;

        if (!$err) {
             if (scalar(@promises) == $resolved) {
                 $deferred->resolve($results_ref);
             }
        } else {
            $deferred->reject($err);
        }
    };

    _wait_for_all(\@promises => sub {
        my ($err, $result, $index) = @_;
        $resolved += 1;
        $results[$index] = $result;
        $attempt_resolution->($err, \@results);
    });
}

sub _wait_for_all {
    my ($promises_ref, $cb) = @_;

    my $index = 0;
    for my $promise (@{$promises_ref}) {
        $promise->callback->(sub {
            my $result = shift;
            $cb->(undef, $result, $index);
        });
        $promise->errback(sub {
            my $err = shift;
            $cb->($err, undef, $index);
        });
        $index++;
    }
}

1;

package main;
use strict;
use warnings;

use AnyEvent;

my $w;
sub async {
    my $deferred = When::defer();

    $w = AnyEvent->timer(
        after => 0,
        cb    => sub {
            $deferred->resolver->resolve(42);
        },
    );
    $deferred->promise;
}

my $cb = AnyEvent->condvar;

my $a = async();
$a->then(sub {
    my $num = shift;
    print "Got number $num\n";

    $cb->send;
});

$cb->recv;