File Coverage

blib/lib/Data/Monad/CondVar.pm

Criterion	Covered	Total	%
statement	202	208	97.1
branch	27	32	84.3
condition	9	12	75.0
subroutine	55	59	93.2
pod	6	8	75.0
total	299	319	93.7

package Data::Monad::CondVar;
use strict;
use warnings;
use 5.012;
use AnyEvent;
use Scalar::Util;
use Exporter qw/import/;

our $VERSION = "0.06";
our @EXPORT = qw/as_cv cv_unit cv_zero cv_fail cv_flat_map_multi cv_map_multi
                 cv_sequence call_cc/;

sub _assert_cv($) {
    $_[0]->ready and die "[BUG]It already has been ready";
    $_[0];
}

sub as_cv(&) {
    my $code = shift;
    $code->(my $cv = AE::cv);
    $cv;
}

sub cv_unit { AnyEvent::CondVar->unit(@_) }
sub cv_zero { AnyEvent::CondVar->zero(@_) }
sub cv_fail { AnyEvent::CondVar->fail(@_) }
sub cv_flat_map_multi(&@) { AnyEvent::CondVar->flat_map_multi(@_) }
sub cv_map_multi(&@) { AnyEvent::CondVar->map_multi(@_) }
sub cv_sequence { AnyEvent::CondVar->sequence(@_) }

sub call_cc(&) {
    my $f = shift;
    my $ret_cv = AE::cv;

    my $skip = sub {
        my @v = @_;
        _assert_cv $ret_cv;
        $ret_cv->send(@v);

        return AE::cv; # nop
    };

    my $branch_cv = $f->($skip)->map(sub {
        _assert_cv $ret_cv;
        $ret_cv->send(@_);
    })->catch(sub {
        _assert_cv $ret_cv;
        $ret_cv->croak(@_);
        cv_unit; # void
    })->catch(sub { warn @_ });
    $ret_cv->canceler(sub { $branch_cv->cancel });

    return $ret_cv;
}


package Data::Monad::CondVar::Mixin;
use strict;
use warnings;
use Carp ();
use Scalar::Util ();
use AnyEvent ();

# extends AE::cv directly
require Data::Monad::Base::MonadZero;
for my $mixin (__PACKAGE__, 'Data::Monad::Base::MonadZero') {
    next if grep { $_ eq $mixin } @AnyEvent::CondVar::ISA;
    push @AnyEvent::CondVar::ISA, $mixin;
}

our $ZERO = "[ZERO of ${\ __PACKAGE__}]";

*_assert_cv = \&Data::Monad::CondVar::_assert_cv;

sub _cv_or_die($) {
    Scalar::Util::blessed $_[0] && $_[0]->isa('AnyEvent::CondVar')
        or Carp::croak "You must use CondVar object here.";
}

sub unit {
    my $class = shift;
    (my $cv = AE::cv)->send(@_);
    return $cv;
}

sub fail {
    my $class = shift;

    # XXX cv's croak doesn't throw the error if the message is empty.
    my $msg = $_[0] || $ZERO;
    (my $cv = AE::cv)->croak($msg);

    return $cv;
}

sub zero { $_[0]->fail($ZERO) }

sub any {
    my ($class, @cvs) = @_;

    my $result_cv = AE::cv;
    my @any_cv;
    for (@cvs) {
        push @any_cv, $_->map(sub {
            return unless $result_cv;
            _assert_cv $result_cv;
            $result_cv->send(@_);
            $result_cv->cancel;
        })->catch(sub {
            return unless $result_cv;
            _assert_cv $result_cv;
            $result_cv->croak(@_);
            $result_cv->cancel;
            return $class->unit;
        })->catch(sub { warn @_ });
    }
    $result_cv->canceler(sub { $_->cancel for @any_cv });

    $result_cv;
}

sub all {
    my ($class, @cvs) = @_;

    my @result;
    (my $result_cv = AE::cv)->begin(sub {
        my ($cv) = @_;
        _assert_cv $cv;
        $cv->send(@result);
    });
    for my $i (0 .. $#cvs) {
        $result_cv->begin;

        $cvs[$i]->map(sub {
            $result[$i] = [@_];
            $result_cv->end;
        })->catch(sub {
            _assert_cv $result_cv;
            $result_cv->croak(@_);
            $result_cv->cancel;
            return $class->unit;
        })->catch(sub { warn @_ });
    }
    $result_cv->end;

    $result_cv->canceler(sub { $_->cancel for @cvs });

    $result_cv;
}

sub cancel { (delete $_[0]->{_monad_canceler} || sub {})->() }

sub canceler {
    my $cv = shift;
    @_ and $cv->{_monad_canceler} = shift;
    $cv->{_monad_canceler};
}

sub _add_cb {
    my ($self, $cb) = @_;
    if(my $old_cb = $self->cb) {
        $self->cb(sub {
            $old_cb->(@_);
            $cb->(@_);
        });
    } else {
        $self->cb($cb);
    }
}

sub flat_map {
    my ($self, $f) = @_;

    my $cv_bound = AE::cv;
    my $cv_current = $self;
    $self->_add_cb(sub {
        my $cv = $cv_current = eval {
            my ($cv) = $f->($_[0]->recv);
            _cv_or_die $cv;
            $cv;
        };

        if ($@) {
            _assert_cv $cv_bound;
            return $cv_bound->croak($@);
        }
        $cv->_add_cb(sub {
            my @v = eval { $_[0]->recv };
            _assert_cv $cv_bound;
            $@ ? $cv_bound->croak($@) : $cv_bound->send(@v);
        });
    });
    $cv_bound->canceler(sub {
        $cv_current->cancel;
        $cv_current->cb(undef); # release the callback function
    });

    return $cv_bound;
}

sub or {
    my ($self, $alter) = @_;

    my $cv_mixed = AE::cv;
    $self->_add_cb(sub {
        my @v = eval { $_[0]->recv };
        unless ($@) {
            $cv_mixed->(@v);
        } elsif ($@ =~ /\Q$ZERO\E/) {
            $alter->_add_cb(sub {
                my @v = eval { $_[0]->recv };
                _assert_cv $cv_mixed;
                $@ ? $cv_mixed->croak($@) : $cv_mixed->(@v);
            });
        } else {
            _assert_cv $cv_mixed;
            $cv_mixed->croak($@);
        }
    });
    $cv_mixed->canceler(sub { $_->cancel for $self, $alter });

    $cv_mixed;
}

sub catch {
    my ($self, $f) = @_;

    my $result_cv = AE::cv;
    my $active_cv = $self;
    $self->_add_cb(sub {
        my @v = eval { $_[0]->recv };
        my $exception = $@ or return $result_cv->(@v);

        my $cv = $active_cv = eval {
            my $cv = $f->($exception);
            _cv_or_die $cv;
            $cv;
        };
        $@ and return (_assert_cv $result_cv)->croak($@);

        $cv->_add_cb(sub {
            my @v = eval { $_[0]->recv };
            _assert_cv $result_cv;
            $@ ? $result_cv->croak($@) : $result_cv->send(@v);
        });
    });
    $result_cv->canceler(sub {
        $active_cv->cancel;
        $active_cv->cb(undef); # release the callback function
    });

    return $result_cv;
}

sub sleep {
    my ($self, $sec) = @_;
    $self->flat_map(sub {
        my @v = @_;
        my $cv = AE::cv;
        my $t; $t = AE::timer $sec, 0, sub { $cv->(@v); undef $cv };
        $cv->canceler(sub { undef $t });
        return $cv;
    });
}

sub timeout {
    my ($self, $sec) = @_;
    my $class = ref $self;

    $class->any($class->unit->sleep($sec), $self);
}

sub retry {
    my $self = shift;
    my $f = pop;
    my ($retry, $pace) = @_;
    $retry ||= 1;
    $pace ||= 0;
    my $class = ref $self;

    $self->flat_map(sub {
        my @args = @_;

        $class->unit($retry, undef, undef)->while(
            sub {
                my ($retry, $ok, $fail) = @_;
                ! $ok and $retry > 0;
            }, sub {
                my ($retry, $ok, $fail) = @_;
                $f->(@args)->flat_map(sub {
                    $class->unit($retry - 1, [@_], undef);
                })->catch(sub {
                    $class->unit($retry - 1, undef, [@_])
                          ->sleep($pace);
                });
            }
        )->flat_map(sub {
            my ($retry, $ok, $fail) = @_;
            $fail ? $class->fail(@$fail) : $class->unit(@$ok);
        });
    });
}

1;

__END__