<commit_before>=begin pod

=TITLE class Supply

=SUBTITLE Asynchronous data stream with multiple subscribers

    class Supply { ... }

A supply is a thread-safe, asynchronous data stream like a
L<Channel|/type/Channel>, but it can have multiple subscribers
(I<taps>) that all get the same values flowing through the supply.

It is a thread-safe implementation of the
L<Observer Pattern|https://en.wikipedia.org/wiki/Observer_pattern>,
and central to supporting reactive programming in Perl 6.


There are two types of Supplies: C<live> and C<on demand>. When tapping into a
C<live> supply, the tap will only see values that are flowing through the
supply B<after> the tap has been created. Such supplies are normally infinite
in nature, such as mouse movements. Closing such a tap does not stop mouse
events from occurring, it just means that the values will go by unseen. All
tappers see the same flow of values.

A tap on an C<on demand> supply will initiate the production of values, and
tapping the supply again may result in a new set of values. For example,
C<Supply.interval> produces a fresh timer with the appropriate interval each
time it is tapped. If the tap is closed, the timer simply stops emitting values
to that tap.

A C<live> C<Supply> is obtained from the L<Supplier|/type/Supplier>
factory method C<Supply>.  New values are emitted by calling C<emit> on
the C<Supplier> object.

	my $supplier = Supplier.new;
	my $supply = $supplier.Supply;
	$supply.tap(-> $v { say "$v" });
	$supplier.emit(42); # Will cause the tap to output "42"

The L<live method|#method live> returns C<True> on live supplies. Factory
methods such as L<interval|#method interval>, L<from-list|#method from-list>
will return I<on demand> supplies.

=head1 Methods that return Taps

=head2 method tap

    method tap(Supply:D: &emit = -> $ { },
            :&done,
            :&quit={ die $_ },
            :&closing,
        ) returns Tap:D

Creates a new tap (a kind of subscription if you will), in addition to all
existing taps. The first positional argument is a piece of code that will be
called when a new value becomes available through the C<emit> call.

The C<&done> callback is called when the C<done> method on the supply is
called, indicating the end of life of the channel. For a C<live> supply the
C<done> method will be called on the parent C<Supplier>.

The C<&quit> callback is called when the C<quit> method on the supply is
called, indicating an erroneous termination of the supply. For a C<live>
supply the C<done> method will be called on the parent C<Supplier>

Method C<tap> returns an object of type L<Tap|/type/Tap>, on which you can
call the C<close> method to cancel the subscription.

    my $s = Supply.from-list(0 .. 5);
    my $t = $s.tap(-> $v { say $v }, done => { say "no more ticks" });

Produces:

    0
    1
    2
    3
    4
    5
    no more ticks

=head2 method act

    method act(Supply:D: &act) returns Tap:D

Creates a tap on the given supply with the given code.  Differently from
C<tap>, the given code is guaranteed to be only executed by one thread at
a time.

=head1 Utility methods

=head2 method taps

    method taps(Supply:D:) returns Int:D

Returns the number of taps.

    my $s = Supply.new;
    say $s.taps;    # 0
    $s.tap(&sqrt);
    say $s.taps;    # 1

=head2 method Channel

    method Channel(Supply:D:) returns Channel:D

Returns a L<Channel|/type/Channel> object that will receive all future values
from the supply, and will be C<close>d when the Supply is done, and quit (shut
down with error) when the supply is quit.

=head2 method Promise

    method Promise(Supply:D:) returns Promise:D

Returns a L<Promise|/type/Promise> object for the next value. That means the
promise is kept when there is a next value, and broken when the supply is done
or quit before a value arrived.

    my $s = Supply.new;
    my $p = $s.Promise;
    $p.then(-> $v { say "got $v.result()" });
    $s.emit('cha');         # got cha

=head2 method close

    method close(Supply:D: Tap:D $t)

Closes the given tap, this is, calls its C<close> callback (if any), and
prevents it from getting any more values flowing through the supply.

=head2 method tappers

    method tappers(Supply:D:) returns List:D

Returns a list of taps as L<Tap|/type/Tap> objects.

=head2 method live

    method live(Supply:D:) returns Bool:D

Returns C<True> if the supply is "live", that is, values are emitted to taps
as soon as they arrive. Always returns C<True> in the default C<Supply> (but
for example on the supply returned from C<Supply.from-list> it's C<False>).

    say Supply.new.live;    # True

=head2 method schedule-on

    method schedule-on(Supply:D: $scheduler)

Runs the emit, done and quit callbacks on the specified scheduler.

This is useful for GUI toolkits that require certain actions to be run from<commit_msg>Remove taps/tappers<commit_after>=begin pod

=TITLE class Supply

=SUBTITLE Asynchronous data stream with multiple subscribers

    class Supply { ... }

A supply is a thread-safe, asynchronous data stream like a
L<Channel|/type/Channel>, but it can have multiple subscribers
(I<taps>) that all get the same values flowing through the supply.

It is a thread-safe implementation of the
L<Observer Pattern|https://en.wikipedia.org/wiki/Observer_pattern>,
and central to supporting reactive programming in Perl 6.


There are two types of Supplies: C<live> and C<on demand>. When tapping into a
C<live> supply, the tap will only see values that are flowing through the
supply B<after> the tap has been created. Such supplies are normally infinite
in nature, such as mouse movements. Closing such a tap does not stop mouse
events from occurring, it just means that the values will go by unseen. All
tappers see the same flow of values.

A tap on an C<on demand> supply will initiate the production of values, and
tapping the supply again may result in a new set of values. For example,
C<Supply.interval> produces a fresh timer with the appropriate interval each
time it is tapped. If the tap is closed, the timer simply stops emitting values
to that tap.

A C<live> C<Supply> is obtained from the L<Supplier|/type/Supplier>
factory method C<Supply>.  New values are emitted by calling C<emit> on
the C<Supplier> object.

	my $supplier = Supplier.new;
	my $supply = $supplier.Supply;
	$supply.tap(-> $v { say "$v" });
	$supplier.emit(42); # Will cause the tap to output "42"

The L<live method|#method live> returns C<True> on live supplies. Factory
methods such as L<interval|#method interval>, L<from-list|#method from-list>
will return I<on demand> supplies.

=head1 Methods that return Taps

=head2 method tap

    method tap(Supply:D: &emit = -> $ { },
            :&done,
            :&quit={ die $_ },
            :&closing,
        ) returns Tap:D

Creates a new tap (a kind of subscription if you will), in addition to all
existing taps. The first positional argument is a piece of code that will be
called when a new value becomes available through the C<emit> call.

The C<&done> callback is called when the C<done> method on the supply is
called, indicating the end of life of the channel. For a C<live> supply the
C<done> method will be called on the parent C<Supplier>.

The C<&quit> callback is called when the C<quit> method on the supply is
called, indicating an erroneous termination of the supply. For a C<live>
supply the C<done> method will be called on the parent C<Supplier>

Method C<tap> returns an object of type L<Tap|/type/Tap>, on which you can
call the C<close> method to cancel the subscription.

    my $s = Supply.from-list(0 .. 5);
    my $t = $s.tap(-> $v { say $v }, done => { say "no more ticks" });

Produces:

    0
    1
    2
    3
    4
    5
    no more ticks

=head2 method act

    method act(Supply:D: &act) returns Tap:D

Creates a tap on the given supply with the given code.  Differently from
C<tap>, the given code is guaranteed to be only executed by one thread at
a time.

=head1 Utility methods

=head2 method Channel

    method Channel(Supply:D:) returns Channel:D

Returns a L<Channel|/type/Channel> object that will receive all future values
from the supply, and will be C<close>d when the Supply is done, and quit (shut
down with error) when the supply is quit.

=head2 method Promise

    method Promise(Supply:D:) returns Promise:D

Returns a L<Promise|/type/Promise> object for the next value. That means the
promise is kept when there is a next value, and broken when the supply is done
or quit before a value arrived.

    my $s = Supply.new;
    my $p = $s.Promise;
    $p.then(-> $v { say "got $v.result()" });
    $s.emit('cha');         # got cha

=head2 method close

    method close(Supply:D: Tap:D $t)

Closes the given tap, this is, calls its C<close> callback (if any), and
prevents it from getting any more values flowing through the supply.

=head2 method live

    method live(Supply:D:) returns Bool:D

Returns C<True> if the supply is "live", that is, values are emitted to taps
as soon as they arrive. Always returns C<True> in the default C<Supply> (but
for example on the supply returned from C<Supply.from-list> it's C<False>).

    say Supply.new.live;    # True

=head2 method schedule-on

    method schedule-on(Supply:D: $scheduler)

Runs the emit, done and quit callbacks on the specified scheduler.

This is useful for GUI toolkits that require certain actions to be run from
{'fraction_non_alphanumeric': 0.08259900201719927, 'fraction_numerical': 0.005096082386665251, 'mean_word_length': 3.6633663366336635, 'pattern_counts': {'":': 0, '<': 92, '<?xml version=': 0, '>': 102, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4767943', 'n_tokens_mistral': 2929, 'n_tokens_neox': 2768, 'n_words': 1373}