File Coverage

blib/lib/Test/Stream/Hub.pm

Criterion	Covered	Total	%
statement	229	232	98.7
branch	82	94	87.2
condition	60	93	64.5
subroutine	33	33	100.0
pod	17	25	68.0
total	421	477	88.2

package Test::Stream::Hub;
use strict;
use warnings;

use Carp qw/carp croak/;
use Test::Stream::State();
use Test::Stream::Util qw/get_tid/;

use Scalar::Util qw/weaken/;

use Test::Stream::HashBase(
    accessors => [qw{
        pid tid hid ipc
        state
        no_ending
        _todo _meta parent_todo
        _mungers
        _filters
        _listeners
        _follow_ups
        _formatter
        _context_init
        _context_release
    }],
);

my $ID_POSTFIX = 1;
sub init {
    my $self = shift;

    $self->{+PID} = $$;
    $self->{+TID} = get_tid();
    $self->{+HID} = join '-', $self->{+PID}, $self->{+TID}, $ID_POSTFIX++;

    $self->{+_TODO} = [];
    $self->{+_META} = {};

    $self->{+STATE} ||= Test::Stream::State->new;

    if (my $formatter = delete $self->{formatter}) {
        $self->format($formatter);
    }

    if (my $ipc = $self->{+IPC}) {
        $ipc->add_hub($self->{+HID});
    }
}

sub inherit {
    my $self = shift;
    my ($from, %params) = @_;

    $self->{+_FORMATTER} = $from->{+_FORMATTER}
        unless $self->{+_FORMATTER} || exists($params{formatter});

    if ($from->{+IPC} && !$self->{+IPC} && !exists($params{ipc})) {
        my $ipc = $from->{+IPC};
        $self->{+IPC} = $ipc;
        $ipc->add_hub($self->{+HID});
    }

    if (my $ls = $from->{+_LISTENERS}) {
        push @{$self->{+_LISTENERS}} => grep { $_->{inherit} } @$ls;
    }

    if (my $ms = $from->{+_MUNGERS}) {
        push @{$self->{+_MUNGERS}} => grep { $_->{inherit} } @$ms;
    }

    if (my $fs = $from->{+_FILTERS}) {
        push @{$self->{+_FILTERS}} => grep { $_->{inherit} } @$fs;
    }
}

sub debug_todo {
    my ($self) = @_;
    my $array = $self->{+_TODO};
    pop @$array while @$array && !defined $array->[-1];
    return (
        parent_todo => $self->{+PARENT_TODO},
        todo        => @$array ? ${$array->[-1]} : undef,
    )
}

sub meta {
    my $self = shift;
    my ($key, $default) = @_;

    croak "Invalid key '" . (defined($key) ? $key : '(UNDEF)') . "'"
        unless $key;

    my $exists = $self->{+_META}->{$key};
    return undef unless $default || $exists;

    $self->{+_META}->{$key} = $default unless $exists;

    return $self->{+_META}->{$key};
}

sub delete_meta {
    my $self = shift;
    my ($key) = @_;

    croak "Invalid key '" . (defined($key) ? $key : '(UNDEF)') . "'"
        unless $key;

    delete $self->{+_META}->{$key};
}

sub set_todo {
    my $self = shift;
    my ($reason) = @_;

    unless (defined wantarray) {
        carp "set_todo(...) called in void context, todo not set!";
        return;
    }

    unless(defined $reason) {
        carp "set_todo() called with undefined argument, todo not set!";
        return;
    }

    my $ref = \$reason;
    push @{$self->{+_TODO}} => $ref;
    weaken($self->{+_TODO}->[-1]);
    return $ref;
}

sub get_todo {
    my $self = shift;
    my $array = $self->{+_TODO};
    pop @$array while @$array && !defined($array->[-1]);
    return undef unless @$array;
    return ${$array->[-1]};
}

sub format {
    my $self = shift;

    my $old = $self->{+_FORMATTER};
    ($self->{+_FORMATTER}) = @_ if @_;

    return $old;
}

sub is_local {
    my $self = shift;
    return $$ == $self->{+PID}
        && get_tid() == $self->{+TID};
}

sub listen {
    my $self = shift;
    my ($sub, %params) = @_;

    carp "Useless addition of a listener in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};

    croak "listen only takes coderefs for arguments, got '$sub'"
        unless ref $sub && ref $sub eq 'CODE';

    push @{$self->{+_LISTENERS}} => { %params, code => $sub };

    $sub; # Intentional return.
}

sub unlisten {
    my $self = shift;

    carp "Useless removal of a listener in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};

    my %subs = map {$_ => $_} @_;

    @{$self->{+_LISTENERS}} = grep { !$subs{$_->{code}} } @{$self->{+_LISTENERS}};
}

sub munge {
    my $self = shift;
    my ($sub, %params) = @_;

    carp "use of mungers is deprecated, look at filters instead. mungers will be removed in the near future.";

    carp "Useless addition of a munger in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};

    croak "munge only takes coderefs for arguments, got '$sub'"
        unless ref $sub && ref $sub eq 'CODE';

    push @{$self->{+_MUNGERS}} => { %params, code => $sub };

    $sub; # Intentional Return
}

sub unmunge {
    my $self = shift;
    carp "Useless removal of a munger in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};
    my %subs = map {$_ => $_} @_;
    @{$self->{+_MUNGERS}} = grep { !$subs{$_->{code}} } @{$self->{+_MUNGERS}};
}

sub filter {
    my $self = shift;
    my ($sub, %params) = @_;

    carp "Useless addition of a filter in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};

    croak "filter only takes coderefs for arguments, got '$sub'"
        unless ref $sub && ref $sub eq 'CODE';

    push @{$self->{+_FILTERS}} => { %params, code => $sub };

    $sub; # Intentional Return
}

sub unfilter {
    my $self = shift;
    carp "Useless removal of a filter in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};
    my %subs = map {$_ => $_} @_;
    @{$self->{+_FILTERS}} = grep { !$subs{$_->{code}} } @{$self->{+_FILTERS}};
}

sub follow_up {
    my $self = shift;
    my ($sub) = @_;

    carp "Useless addition of a follow-up in a child process or thread!"
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};

    croak "follow_up only takes coderefs for arguments, got '$sub'"
        unless ref $sub && ref $sub eq 'CODE';

    push @{$self->{+_FOLLOW_UPS}} => $sub;
}

sub add_context_init {
    my $self = shift;
    my ($sub) = @_;

    croak "add_context_init only takes coderefs for arguments, got '$sub'"
        unless ref $sub && ref $sub eq 'CODE';

    push @{$self->{+_CONTEXT_INIT}} => $sub;

    $sub; # Intentional return.
}

sub remove_context_init {
    my $self = shift;
    my %subs = map {$_ => $_} @_;
    @{$self->{+_CONTEXT_INIT}} = grep { !$subs{$_} == $_ } @{$self->{+_CONTEXT_INIT}};
}

sub add_context_release {
    my $self = shift;
    my ($sub) = @_;

    croak "add_context_release only takes coderefs for arguments, got '$sub'"
        unless ref $sub && ref $sub eq 'CODE';

    push @{$self->{+_CONTEXT_RELEASE}} => $sub;

    $sub; # Intentional return.
}

sub remove_context_release {
    my $self = shift;
    my %subs = map {$_ => $_} @_;
    @{$self->{+_CONTEXT_RELEASE}} = grep { !$subs{$_} == $_ } @{$self->{+_CONTEXT_RELEASE}};
}

sub send {
    my $self = shift;
    my ($e) = @_;

    my $ipc = $self->{+IPC} || return $self->process($e);

    if($e->global) {
        $ipc->send('GLOBAL', $e);
        return $self->process($e);
    }

    return $ipc->send($self->{+HID}, $e)
        if $$ != $self->{+PID} || get_tid() != $self->{+TID};

    $self->process($e);
}

sub process {
    my $self = shift;
    my ($e) = @_;

    if ($self->{+_MUNGERS}) {
        for (@{$self->{+_MUNGERS}}) {
            $_->{code}->($self, $e);
            return unless $e;
        }
    }

    if ($self->{+_FILTERS}) {
        for (@{$self->{+_FILTERS}}) {
            $e = $_->{code}->($self, $e);
            return unless $e;
        }
    }

    my $state = $self->{+STATE};
    $e->update_state($state);
    my $count = $state->count;

    $self->{+_FORMATTER}->write($e, $count) if $self->{+_FORMATTER};

    if ($self->{+_LISTENERS}) {
        $_->{code}->($self, $e, $count) for @{$self->{+_LISTENERS}};
    }

    my $code = $e->terminate;
    $self->terminate($code, $e) if defined $code;

    return $e;
}

sub terminate {
    my $self = shift;
    my ($code) = @_;
    exit($code);
}

sub cull {
    my $self = shift;

    my $ipc = $self->{+IPC} || return;
    return if $self->{+PID} != $$ || $self->{+TID} != get_tid();

    # No need to do IPC checks on culled events
    $self->process($_) for $ipc->cull($self->{+HID});
}

sub finalize {
    my $self = shift;
    my ($dbg, $do_plan) = @_;

    $self->cull();
    my $state = $self->{+STATE};

    my $plan   = $state->plan;
    my $count  = $state->count;
    my $failed = $state->failed;

    # return if NOTHING was done.
    return unless $do_plan || defined($plan) || $count || $failed;

    unless ($state->ended) {
        if ($self->{+_FOLLOW_UPS}) {
            $_->($dbg, $self) for reverse @{$self->{+_FOLLOW_UPS}};
        }

        # These need to be refreshed now
        $plan   = $state->plan;
        $count  = $state->count;
        $failed = $state->failed;

        if (($plan && $plan eq 'NO PLAN') || ($do_plan && !$plan)) {
            $self->send(
                Test::Stream::Event::Plan->new(
                    debug => $dbg,
                    max => $count,
                )
            );
            $plan = $state->plan;
        }
    }

    $state->finish($dbg->frame);
}

sub DESTROY {
    my $self = shift;
    my $ipc = $self->{+IPC} || return;
    return unless $$ == $self->{+PID};
    return unless get_tid() == $self->{+TID};

    local $?;
    $ipc->drop_hub($self->{+HID});
}

1;

__END__