File Coverage

blib/lib/RxPerl/Extras.pm

Criterion	Covered	Total	%
statement	85	88	96.5
branch	16	26	61.5
condition	5	9	55.5
subroutine	13	13	100.0
pod	4	4	100.0
total	123	140	87.8

package RxPerl::Extras;

use strict;
use warnings;

use RxPerl::Operators::Creation 'rx_observable', 'rx_timer', 'rx_defer';
use RxPerl::Operators::Pipeable 'op_map', 'op_take';

use Exporter 'import';
our @EXPORT_OK = qw/
    op_exhaust_all_with_latest op_exhaust_map_with_latest
    op_throttle_time_with_both_leading_and_trailing
    op_throttle_with_both_leading_and_trailing
/;
our %EXPORT_TAGS = (all => \@EXPORT_OK);

our $VERSION = "v0.0.5";

sub op_exhaust_all_with_latest {
    return sub {
        my ($source) = @_;

        return rx_observable->new(sub {
            my ($subscriber) = @_;

            my $active_subscription;
            my $big_completed;
            my $own_subscription = RxPerl::Subscription->new;

            my ($owed_val, $val_is_owed);
            my $helper_sub;

            $subscriber->subscription->add(
                \$active_subscription,
                $own_subscription,
                sub { undef $helper_sub }, # break ref cycle
            );

            $helper_sub = sub {
                my ($new_obs) = @_;

                !$active_subscription or do {
                    $owed_val = $new_obs;
                    $val_is_owed = 1;
                    return;
                };

                $active_subscription = RxPerl::Subscription->new;
                my $small_subscriber = {
                    new_subscription => $active_subscription,
                    next             => sub {
                        $subscriber->{next}->(@_) if defined $subscriber->{next};
                    },
                    error            => sub {
                        $subscriber->{error}->(@_) if defined $subscriber->{error};
                    },
                    complete         => sub {
                        undef $active_subscription;
                        if ($val_is_owed) {
                            $val_is_owed = 0;
                            $helper_sub->($owed_val);
                        } else {
                            $subscriber->{complete}->() if $big_completed and defined $subscriber->{complete};
                        }
                    },
                };
                $new_obs->subscribe($small_subscriber);
            };

            my $own_subscriber = {
                new_subscription => $own_subscription,
                next             => $helper_sub,
                error            => sub {
                    $subscriber->{error}->(@_) if defined $subscriber->{error};
                },
                complete         => sub {
                    $big_completed = 1;
                    $subscriber->{complete}->() if !$active_subscription and defined $subscriber->{complete};
                },
            };

            $source->subscribe($own_subscriber);

            return;
        });
    }
}

sub op_exhaust_map_with_latest {
    my ($observable_factory) = @_;

    return sub {
        my ($source) = @_;

        return $source->pipe(
            op_map(sub {
                my @args = @_;
                rx_defer(sub {
                    local $_ = $args[0];
                    $observable_factory->(@args);
                });
            }),
            op_exhaust_all_with_latest(),
        );
    };
}

sub op_throttle_time_with_both_leading_and_trailing {
    my ($duration) = @_;

    return op_throttle_with_both_leading_and_trailing(sub { rx_timer($duration) });
}

sub op_throttle_with_both_leading_and_trailing {
    my ($duration_selector) = @_;

    return sub {
        my ($source) = @_;

        return rx_observable->new(sub {
            my ($subscriber) = @_;

            my ($owed_val, $val_is_owed);
            my ($helper_sub, $mini_subscriber, $mini_subscription);

            my $own_subscription = RxPerl::Subscription->new;

            $subscriber->subscription->add(
                $own_subscription,
                \$mini_subscription,
                sub { undef $helper_sub }, # break ref cycle
            );

            $mini_subscriber = {
                error    => sub {
                    $subscriber->{error}->(@_) if defined $subscriber->{error};
                },
                complete => sub {
                    undef $mini_subscription;
                    if ($val_is_owed) {
                        $val_is_owed = 0;
                        $helper_sub->($owed_val);
                    }
                },
            };

            $helper_sub = sub {
                my ($v) = @_;

                if ($mini_subscription) {
                    $owed_val = $v;
                    $val_is_owed = 1;
                } else {
                    $mini_subscription = do { local $_ = $v; $duration_selector->($v) }->pipe(
                        op_take(1),
                    )->subscribe($mini_subscriber);
                    $subscriber->{next}->(@_) if defined $subscriber->{next};
                }
            };

            my $own_subscriber = {
                new_subscription => $own_subscription,
                %$subscriber,
                next => $helper_sub,
                complete => sub {
                    $subscriber->{next}->($owed_val), $val_is_owed = 0 if $val_is_owed and defined $subscriber->{next};
                    $subscriber->{complete}->() if defined $subscriber->{complete};
                },
            };

            $source->subscribe($own_subscriber);

            return;
        });
    };
}


1;
__END__