File Coverage

blib/lib/Ryu/Node.pm

Criterion	Covered	Total	%
statement	72	73	98.6
branch	43	54	79.6
condition	27	47	57.4
subroutine	19	19	100.0
pod	7	11	63.6
total	168	204	82.3

package Ryu::Node;

use strict;
use warnings;

our $VERSION = '4.001'; # VERSION
our $AUTHORITY = 'cpan:TEAM'; # AUTHORITY

=head1 NAME

Ryu::Node - generic node

=head1 DESCRIPTION

This is a common base class for all sources, sinks and other related things.
It does very little.

=cut

use Future;
use Scalar::Util qw(refaddr);

=head1 METHODS

Not really. There's a constructor, but that's not particularly exciting.

=cut

sub new {
    bless {
        pause_propagation => 1,
        @_[1..$#_]
    }, $_[0]
}

=head2 describe

Returns a string describing this node and any parents - typically this will result in a chain
like C<< from->combine_latest->count >>.

=cut

# It'd be nice if L already provided a method for this, maybe I should suggest it
sub describe {
    my ($self) = @_;
    my $completed = $self->_completed;
    ($self->parent ? $self->parent->describe . '=>' : '') . '[' . ($self->label // 'unknown') . '](' . ($completed ? $completed->state : 'inactive') . ')';
}

=head2 completed

Returns a L indicating completion (or failure) of this stream.

=cut

sub completed {
    my ($self) = @_;
    my $completion = $self->_completed
        or return undef;
    return $completion->without_cancel->on_ready(sub {
        my $f = shift;
        my ($expected) = $f->state =~ /^(\S+)/;
        my ($actual) = $completion->state =~ /^(\S+)/;
        if($expected ne $actual) {
            warn "Completed state $actual does not match internal state $expected - if you are calling ->completed->$expected, this will not work: use ->finish or ->fail instead";
        }
    });
}

# Internal use only, since it's cancellable
sub _completed {
    my ($self) = @_;
    return $self->{completed} if $self->{completed};
    $self->{completed} = my $f = $self->new_future(
        'completion'
    );
    $f->on_ready(
        $self->curry::weak::cleanup
    ) if $self->can('cleanup');
    $f
}

=head2 pause

Does nothing useful.

=cut

sub pause {
    my ($self, $src) = @_;
    my $k = (defined $src and ref $src)
        ? refaddr($src // $self) // 0
        : $src // 0;

    my $was_paused = $self->{is_paused} && keys %{$self->{is_paused}};
    if($self->{unblocked} and $self->{unblocked}->is_ready) {
        delete $self->{unblocked};
    }
    ++$self->{is_paused}{$k};
    if(my $parent = $self->parent) {
        $parent->pause($self) if $self->{pause_propagation};
    }
    if(my $flow_control = $self->{flow_control}) {
        $flow_control->emit(0) unless $was_paused;
    }
    $self
}

=head2 resume

Is about as much use as L.

=cut

sub resume {
    my ($self, $src) = @_;
    my $k = (defined $src and ref $src)
        ? refaddr($src // $self) // 0
        : $src // 0;
    delete $self->{is_paused}{$k} unless --$self->{is_paused}{$k} > 0;
    return $self if $self->{is_paused} and keys %{$self->{is_paused}};

    if(my $f = $self->{unblocked}) {
        $f->done unless $f->is_ready;
    }
    if(my $parent = $self->parent) {
        $parent->resume($self) if $self->{pause_propagation};
    }
    if(my $flow_control = $self->{flow_control}) {
        $flow_control->emit(1);
    }
    $self
}

=head2 unblocked

Returns a L representing the current flow control state of this node.

It will be L if this node is currently paused,
otherwise L.

=cut

sub unblocked {
    # Since we don't want stray callers to affect our internal state, we always return
    # a non-cancellable version of our internal Future.
    my $self = shift;
    return $self->_unblocked->without_cancel
}

sub _unblocked {
    my ($self) = @_;
    # Since we don't want stray callers to affect our internal state, we always return
    # a non-cancellable version of our internal Future.
    $self->{unblocked} //= do {
        $self->is_paused
        ? $self->new_future
        : Future->done
    };
}

=head2 is_paused

Might return 1 or 0, but is generally meaningless.

=cut

sub is_paused {
    my ($self, $src) = @_;
    return keys(%{ $self->{is_paused} }) ? 1 : 0 unless defined $src;
    my $k = (defined $src and ref $src)
        ? refaddr($src // $self) // 0
        : $src // 0;
    return exists $self->{is_paused}{$k}
    ? 0 + $self->{is_paused}{$k}
    : 0;
}

sub flow_control {
    my ($self) = @_;
    return $self->{flow_control} if $self->{flow_control};
    $self->{flow_control} = my $fc = Ryu::Source->new(
        new_future => $self->{new_future}
    );
    $self->_completed->on_ready(sub {
        my $fc = delete $self->{flow_control}
            or return;
        $fc->finish;
    });
    $fc
}

sub label { shift->{label} }

sub parent { shift->{parent} }

=head2 new_future

Used internally to get a L.

=cut

sub new_future {
    my $self = shift;
    (
        $self->{new_future} //= $Ryu::Source::FUTURE_FACTORY
    )->($self, @_)
}


1;

__END__