# NAME

RxPerl::Extras - extra operators for RxPerl

# SYNOPSIS

    use RxPerl::Mojo qw/ ... /; # work also with RxPerl::IOAsync or RxPerl::AnyEvent
    use RxPerl::Extras 'op_exhaust_map_with_latest'; # or ':all'

    # (pause 5 seconds) 0, (pause 5 seconds) 2, complete
    rx_timer(0, 2)->pipe(
        op_take(3),
        op_exhaust_map_with_latest(sub ($val, @) {
            return rx_of($val)->pipe( op_delay(5) );
        }),
    )->subscribe($observer);

# DESCRIPTION

RxPerl::Extras is a collection of original [RxPerl](https://metacpan.org/pod/RxPerl) operators which the author thinks could be useful to many.

It currently contains only one pipeable operator.

# EXPORTABLE FUNCTION

The code samples in this section assume `$observer` has been set to:

    $observer = {
        next     => sub {say "next: ", $_[0]},
        error    => sub {say "error: ", $_[0]},
        complete => sub {say "complete"},
    };

## PIPEABLE OPERATORS

- op\_exhaust\_map\_with\_latest

    Works like [RxPerl](https://metacpan.org/pod/RxPerl)'s [op\_exhaust\_map](https://metacpan.org/pod/RxPerl%23op_exhaust_map), except if any new next events arrive before exhaustion,
    the latest of those events will also be processed after exhaustion.

        # (pause 5 seconds) 0, (pause 5 seconds) 2, complete
        rx_timer(0, 2)->pipe(
            op_take(3),
            op_exhaust_map_with_latest(sub ($val, @) {
                return rx_of($val)->pipe( op_delay(5) );
            }),
        )->subscribe($observer);

# NOTIFICATIONS FOR NEW RELEASES

You can start receiving emails for new releases of this module, at [https://perlmodules.net](https://perlmodules.net).

# COMMUNITY CODE OF CONDUCT

[RxPerl's Community Code of Conduct](https://metacpan.org/pod/RxPerl%3A%3ACodeOfConduct) applies to this module too.

# LICENSE

Copyright (C) 2024 Alexander Karelas.

This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself.

# AUTHOR

Alexander Karelas <karjala@cpan.org>