Supplies, react and whenever

36.1 Supplies (Asynchronous Streams)

# Live supply from a Supplier
my $supplier = Supplier.new;
my $supply = $supplier.Supply;
$supply.tap(-> $v { say "Tap 1: $v" });
$supply.grep(* %% 2).map(* * 10).tap(-> $v { say "Tap 2: $v" });
$supplier.emit($_) for 1..4;
$supplier.done;

# On-demand supply (each tap gets its own stream)
my $numbers = supply {
    for 1..3 { emit $_ }
    done;
};
$numbers.tap(&say);

# Timers
my $ticks = Supply.interval(0.2).head(3);
react {
    whenever $ticks -> $t { say "tick $t" }
}

36.2 react/whenever: Event-Driven Programs

react {
    whenever Supply.interval(1) -> $sec {
        say "Elapsed: $sec s";
        done if $sec >= 3;                       # exit the react block
    }
    whenever Promise.in(10) { say "timeout"; done }
    whenever signal(SIGINT) { say "Interrupted, shutting down"; done }
}
say "react block finished";

# Combining multiple sources
my $s1 = Supply.interval(0.3).map({ "A$_" }).head(3);
my $s2 = Supply.interval(0.5).map({ "B$_" }).head(2);
react {
    whenever $s1.merge($s2) -> $event { say $event }
}

36.3 Supply Operations

map, grep, head, tail, unique, squish merge, zip, zip-latest, combine...

batch(:elems,:seconds) throttle(n, seconds), delayed(s), stable(s) Rate control / debounce lines, words, comb Text stream processing reduce, produce, elems, minmax Promise / wait / list Bridge to Promises or blocking lists share, schedule-on, serialize Sharing and scheduling

my $file-changes = '.'.IO.watch;              # filesystem events as a Supply
my $lines = $*IN.Supply.lines;                # STDIN lines as a Supply