File Coverage

blib/lib/Async/Defer.pm

Criterion	Covered	Total	%
statement	331	335	98.8
branch	127	130	97.6
condition	52	58	89.6
subroutine	44	44	100.0
pod	16	16	100.0
total	570	583	97.7

package Async::Defer;
use 5.012;
use warnings;
use strict;
use Carp;
no if $] >= 5.017011, warnings => 'experimental::smartmatch';

our $VERSION = 'v1.0.1';

use Scalar::Util qw( refaddr );
use List::Util qw( any );

## no critic (ProhibitBuiltinHomonyms)

use constant NOT_RUNNING=> -1;
use constant OP_CODE    => 1;
use constant OP_DEFER   => 2;
use constant OP_IF      => 3;
use constant OP_ELSE    => 4;
use constant OP_ENDIF   => 5;
use constant OP_WHILE   => 6;
use constant OP_ENDWHILE=> 7;
use constant OP_TRY     => 8;
use constant OP_CATCH   => 9;
use constant OP_FINALLY => 10;
use constant OP_ENDTRY  => 11;

my %SELF;


sub new {
    my ($class) = @_;
    my $this = bless {}, $class;
    $SELF{refaddr $this} = {
        parent  => undef,      # parent Defer object, if any
        opcode  => [],         # [[OP_CODE,$sub], [OP_TRY], ...]
        pc      => NOT_RUNNING,# point to _CURRENT_ opcode, if any
        iter    => [],         # [[1,$outer_while_pc], [8,$inner_while_pc], ...]
        findone => undef,      # undef or ['continue'] or ['break'] or ['throw',$err]
    };
    return $this;
}

sub DESTROY {
    my ($this) = @_;
    delete $SELF{refaddr $this};
    return;
}

sub clone {
    my ($this) = @_;
    my $self = $SELF{refaddr $this};

    my $clone = __PACKAGE__->new();
    my $clone_self = $SELF{refaddr $clone};

    $clone_self->{opcode} = [ @{ $self->{opcode} } ];
    %{$clone} = %{$this};
    return $clone;
}

sub iter {
    my ($this) = @_;
    my $self = $SELF{refaddr $this};

    if (!@{ $self->{iter} }) {
        croak 'iter() can be used only inside while';
    }

    return $self->{iter}[-1][0];
}

sub _add {
    my ($this, $op, @params) = @_;
    my $self = $SELF{refaddr $this};

    if ($self->{pc} != NOT_RUNNING) {
        croak 'unable to modify while running';
    }

    push @{ $self->{opcode} }, [ $op, @params ];
    return $this;
}

sub do {
    my ($this, $task, @more_tasks) = @_;
    if(@more_tasks){
        for ($task, @more_tasks) {
            $this->do($_);
        }
        return $this;
    }
    given (ref $task) {
        when ('CODE') {
            return $this->_add(OP_CODE, $task);
        }
        when (__PACKAGE__) {
            return $this->_add(OP_DEFER, $task);
        }
        when ('ARRAY') {
            my %task = map { $_ => $task->[$_] } 0 .. $#{ $task };
            return $this->_add(OP_CODE, _do_batch(1, %task));
        }
        when ('HASH') {
            return $this->_add(OP_CODE, _do_batch(0, %{ $task }));
        }
        default {
            croak 'require CODE/Defer object or ARRAY/HASH in first param'
        }
    }
}

sub _do_batch {
    my ($is_array, %task) = @_;

    # Isolate each task in own Defer object to guarantee they won't be
    # surprised by shared state.
    for my $key (keys %task) {
        my $task;
        given (ref $task{$key}) {
            when ('CODE') {
                $task = __PACKAGE__->new();
                $task->do( $task{$key} );
            }
            when (__PACKAGE__) {
                $task = $task{$key}->clone();
            }
            default {
                my $pos = $is_array ? $key+1 : "{$key}";
                croak 'require CODE/Defer object in param '.$pos;
            }
        }
        $task{$key} = $task;
    }

    return sub{
        my ($d, @taskparams) = @_;
        my %taskparams
            = !$is_array ? (@taskparams)
            :              (map { ($_ => $taskparams[$_]) } 0 .. $#taskparams);

        if (!keys %task) {
            return $d->done();
        }

        my %taskresults = map { $_ => undef } keys %task;
        for my $key (sort keys %task) {     # sort just to simplify testing
            my $t = __PACKAGE__->new();
            $t->try();
                $t->do( $task{$key} );
            $t->catch(
                qr/.*/ms => sub{
                    my ($t,$err) = @_;      ## no critic (ProhibitReusedNames)
                    $t->{err} = $err;
                    $t->done();
                },
                FINALLY => sub{
                    my ($t, @result) = @_;  ## no critic (ProhibitReusedNames)
                    $taskresults{$key} = $t->{err} // \@result;
                    if (!any {!defined} values %taskresults) {
                        my @taskresults
                            = !$is_array ? %taskresults
                            :              map { $taskresults{$_-1} } 1 .. keys %taskresults;
                        $d->done(@taskresults);
                    }
                    return $t->done();
                },
            );
            $t->run( undef, @{ $taskparams{$key} || [] } );
        }
    };
}

sub if {
    my ($this, $code) = @_;
    if (!$code || ref $code ne 'CODE') {
        croak 'require CODE in first param';
    }
    return $this->_add(OP_IF, $code);
}

sub else {
    my ($this) = @_;
    return $this->_add(OP_ELSE);
}

sub end_if {
    my ($this) = @_;
    return $this->_add(OP_ENDIF);
}

sub while {
    my ($this, $code) = @_;
    if (!$code || ref $code ne 'CODE') {
        croak 'require CODE in first param';
    }
    return $this->_add(OP_WHILE, $code);
}

sub end_while {
    my ($this) = @_;
    return $this->_add(OP_ENDWHILE);
}

sub try {
    my ($this) = @_;
    return $this->_add(OP_TRY);
}

sub catch {
    my ($this, @param) = @_;
    if (2 > @param) {
        croak 'require at least 2 params';
    } elsif (@param % 2) {
        croak 'require even number of params';
    }

    my ($finally, @catch);
    while (my ($cond, $code) = splice @param, 0, 2) {
        if ($cond eq 'FINALLY') {
            $finally ||= $code;
        } else {
            push @catch, $cond, $code;
        }
    }

    if (@catch) {
        $this->_add(OP_CATCH, @catch);
    }
    if ($finally) {
        $this->_add(OP_FINALLY, $finally);
    }
    return $this->_add(OP_ENDTRY);
}

sub _check_stack {
    my ($self) = @_;
    my @stack;
    my %op_open  = (
        OP_IF()         => 'end_if()',
        OP_WHILE()      => 'end_while()',
        OP_TRY()        => 'catch()',
    );
    my %op_close = (
        OP_ENDIF()      => [ OP_IF,     'end_if()'      ],
        OP_ENDWHILE()   => [ OP_WHILE,  'end_while()'   ],
        OP_ENDTRY()     => [ OP_TRY,    'catch()'       ],
    );
    my $extra = 0;
    for (my $i = 0; $i < @{ $self->{opcode} }; $i++) {
        my ($op) = @{ $self->{opcode}[ $i ] };

        if ($op == OP_CATCH || $op == OP_FINALLY) {
            $extra++;
        }

        if ($op_open{$op}) {
            push @stack, [$op,0];   # second number is counter for seen OP_ELSE
        }
        elsif ($op_close{$op}) {
            my ($close_op, $close_func) = @{ $op_close{$op} };
            if (@stack && $stack[-1][0] == $close_op) {
                pop @stack;
            } else {
                croak 'unexpected '.$close_func.' at operation '.($i+1-$extra);
            }
        }
        elsif ($op == OP_ELSE) {
            if (!(@stack && $stack[-1][0] == OP_IF)) {
                croak 'unexpected else() at operation '.($i+1-$extra);
            }
            elsif ($stack[-1][1]) {
                croak 'unexpected double else() at operation '.($i+1-$extra);
            }
            $stack[-1][1]++;
        }
    }
    if (@stack) {
        croak 'expected '.$op_open{ $stack[-1][0] }.' at end';
    }
    return;
}

sub run {
    my ($this, $d, @result) = @_;
    my $self = $SELF{refaddr $this};

    my %op_stmt = map {$_=>1} OP_CODE, OP_DEFER, OP_FINALLY;
    if (!any {$op_stmt{ $_->[0] }} @{ $self->{opcode} }) {
        croak 'no operations to run, use do() first';
    }
    if ($self->{pc} != NOT_RUNNING) {
        croak 'already running';
    }
    _check_stack($self);

    if(ref($d) eq 'CODE') {
        my $callback = $d;
        $d = __PACKAGE__->new();
        $d->do(
            sub {
                my ($defer, @results) = @_;
                $callback->(@results);
                $defer->done;
            }
        );
    }

    $self->{parent} = $d;
    $this->done(@result);
    return;
}

sub _op {
    my ($self) = @_;
    my ($op, @params) = @{ $self->{opcode}[ $self->{pc} ] };
    return wantarray ? ($op, @params) : $op;
}

sub done {
    my ($this, @result) = @_;
    my $self = $SELF{refaddr $this};

    # If OP_FINALLY was called while processing continue(), break() or throw(),
    # and it has finished with done() - continue with continue/break/throw by
    # calling them _again_ instead of done().
    if ($self->{findone}) {
        my ($method, @param) = @{ $self->{findone} };
        return $this->$method(@param);
    }

    while (++$self->{pc} <= $#{ $self->{opcode} }) {
        my ($opcode, @param) = _op($self);

        # @result received from previous opcode will be available to next
        # opcode only if these opcodes stay one-after-one without any
        # other opcodes between them (like OP_IF, for example).
        # Only exception is (no-op) OP_TRY, OP_CATCH and OP_ENDTRY.
        # This limitation should help user to avoid subtle bugs.
        given ($opcode) {
            when (OP_CODE) {
                return $param[0]->($this, @result);
            }
            when (OP_DEFER) {
                return $param[0]->run($this, @result);
            }
            when (OP_FINALLY) {
                return $param[0]->($this, @result);
            }
            when ([OP_TRY,OP_CATCH,OP_ENDTRY]) {
                next;
            }
        }
        @result = ();

        given ($opcode) {
            when (OP_IF) {
                # true  - do nothing (i.e. just move to next opcode)
                # false - skip to nearest OP_ELSE or OP_ENDIF
                if (!$param[0]->( $this )) {
                    my $stack = 0;
                    while (++$self->{pc} <= $#{ $self->{opcode} }) {
                        my $op = _op($self);
                          $op == OP_ELSE  && !$stack    ? last
                        : $op == OP_ENDIF && !$stack    ? last
                        : $op == OP_IF                  ? $stack++
                        : $op == OP_ENDIF               ? $stack--
                        :                                 next;
                    }
                }
            }
            when (OP_ELSE) {
                # skip this OP_ELSE branch to nearest OP_ENDIF
                my $stack = 0;
                while (++$self->{pc} <= $#{ $self->{opcode} }) {
                    my $op = _op($self);
                      $op == OP_ENDIF && !$stack    ? last
                    : $op == OP_IF                  ? $stack++
                    : $op == OP_ENDIF               ? $stack--
                    :                                 next;
                }
            }
            when (OP_WHILE) {
                # We can "enter" OP_WHILE in two cases - for the first time,
                # OR because of continue() called inside this OP_WHILE.
                if (!@{$self->{iter}} || $self->{iter}[-1][1] != $self->{pc}) {
                    push @{ $self->{iter} }, [ 1, $self->{pc} ];
                }
                # We now already "inside" this OP_WHILE, so we can use break()
                # to exit _this_ OP_WHILE.
                if (!$param[0]->( $this )) {
                    return $this->break();
                }
            }
            when (OP_ENDWHILE) {
                # We now still "inside" current OP_WHILE, so we can use continue()
                # to repeat _this_ OP_WHILE.
                return $this->continue();
            }
        }
    }

    $self->{pc} = NOT_RUNNING;
    if ($self->{parent}) {
        return $self->{parent}->done(@result);
    }

    # If we're here, done() was called by last opcode, and this is
    # top-level Defer object, nothing more to do - STOP.
}

# Before executing continue/break logic we have to find and execute all
# OP_FINALLY for all already open OP_TRY blocks within this OP_WHILE.
# So, this helper skip opcodes inside this OP_WHILE until it found
# either OP_FINALLY or OP_ENDWHILE or last opcode.
sub _skip_while {
    my ($self) = @_;

    # 1. continue() can be called exactly on OP_ENDWHILE (by done())
    # 2. continue/break can be called by last opcode
    # In both cases we shouldn't do anything (including moving {pc} forward).
    if (_op($self) == OP_ENDWHILE || $self->{pc} == $#{ $self->{opcode} }) {
        return;
    }

    my $stack = 0;
    my $trystack = 0;
    while (++$self->{pc} < $#{ $self->{opcode} }) {
        my $op = _op($self);
          $op == OP_ENDWHILE && !$stack     ? last
        : $op == OP_WHILE                   ? $stack++
        : $op == OP_ENDWHILE                ? $stack--
        : $op == OP_TRY                     ? $trystack++
        : $op == OP_ENDTRY && $trystack     ? $trystack--
        : $op == OP_FINALLY && !$trystack   ? last
        :                                     next;
    }

    return;
}

sub continue {
    my ($this) = @_;
    my $self = $SELF{refaddr $this};

    # Any next call to continue/break/throw cancels current continue/break/throw (if any).
    $self->{findone} = undef;

    _skip_while($self);
    my ($op, @param) = _op($self);
    if ($op == OP_FINALLY) {
        # If OP_FINALLY ends with done() - call continue() again instead.
        $self->{findone} = ['continue'];
        return $param[0]->($this);
    }

    # We now at OP_ENDWHILE, rewind to corresponding OP_WHILE.
    my $stack = 0;
    while (--$self->{pc} > 0) {
        $op = _op($self);
          $op == OP_WHILE && !$stack    ? last
        : $op == OP_ENDWHILE            ? $stack++
        : $op == OP_WHILE               ? $stack--
        :                                 next;
    }

    # If continue was called outside OP_WHILE there is no iteration number.
    if (@{ $self->{iter} }) {
        $self->{iter}[-1][0]++;
    }

    # Step one opcode back because done() will move one opcode forward
    # and so process this OP_WHILE.
    --$self->{pc};
    return $this->done();
}

sub break {
    my ($this) = @_;
    my $self = $SELF{refaddr $this};

    # Any next call to continue/break/throw cancels current continue/break/throw (if any).
    $self->{findone} = undef;

    _skip_while($self);
    my ($op, @param) = _op($self);
    if ($op == OP_FINALLY) {
        # If OP_FINALLY ends with done() - call break() again instead.
        $self->{findone} = ['break'];
        return $param[0]->($this);
    }

    # We now at OP_ENDWHILE.
    pop @{ $self->{iter} };
    return $this->done();
}

sub throw {
    my ($this, $err) = @_;
    my $self = $SELF{refaddr $this};
    $err //= q{};

    # Any next call to continue/break/throw cancels current continue/break/throw (if any).
    $self->{findone} = undef;

    # If throw() was called by break opcode in this OP_TRY (either OP_FINALLY,
    # or OP_CATCH if there no OP_FINALLY in this OP_TRY), then we should look
    # for handler in outer OP_TRY, not in this one.
    # So we set $stack=1 to skip over current OP_TRY's OP_ENDTRY.
    my ($nextop) = @{ $self->{opcode}[ $self->{pc} + 1 ] || [] };
    my $stack = $nextop && $nextop == OP_ENDTRY ? 1 : 0;
    # Skip until OP_CATCH or OP_FINALLY in current OP_TRY block.
    # If while skipping we exit some OP_WHILE(s) - pop their iterators.
    while (++$self->{pc} <= $#{ $self->{opcode} }) {
        my $op = _op($self);
          $op == OP_CATCH   && !$stack      ? last
        : $op == OP_FINALLY && !$stack      ? last
        : $op == OP_TRY                     ? $stack++
        : $op == OP_ENDTRY                  ? $stack--
        : $op == OP_WHILE                   ? push @{ $self->{iter} }, [ 1, $self->{pc} ]
        : $op == OP_ENDWHILE                ? pop  @{ $self->{iter} }
        :                                     next;
    }

    if ($self->{pc} > $#{ $self->{opcode} }) {
        if ($self->{parent}) {
            return $self->{parent}->throw($err);
        } else {
            croak 'uncatched exception in Defer: '.$err;
        }
    }

    my ($op, @param) = _op($self);
    if ($op == OP_CATCH) {
        while (my ($cond, $code) = splice @param, 0, 2) {
            if ($err =~ /$cond/ms) {
                return $code->($this, $err);
            }
        }
        # Re-throw exception if no one regex in this OP_CATCH match it.
        return $this->throw($err);
    }
    else { # OP_FINALLY
        # If OP_FINALLY ends with done() - call throw($err) again instead.
        $self->{findone} = ['throw', $err];
        return $param[0]->($this, $err);
    }
}


1; # Magic true value required at end of module
__END__