/ 
Ekorn-Pipeline-v0.3.1
/ Ekorn::Pipeline::Sink

NAME

Ekorn::Pipeline::Sink - a sink made afresh for each run

SYNOPSIS

use Ekorn::Pipeline::Sink;

# One value: the total
my $total = Ekorn::Pipeline::Sink->new_sink_one(sub {
    my $sum = 0;    # this run's
    return (sub ($x) { $sum += $x }, sub () { $sum });
});

my $sum = Ekorn::Pipeline->new(
    sub { $_ * 2 },
    $total,
)->(1 .. 4);    # 20
$total->(1 .. 4);    # 10

# A list: each value, written as it comes, given back
sub logged_to ($path) {
    return Ekorn::Pipeline::Sink->new_sink_many(sub {
        open my $fh, '>', $path or die "Can't write $path: $!\n";
        my @seen;
        return (
            sub ($x)   { print {$fh} "$x\n"; push @seen, $x },
            sub ()     { @seen },
            sub ($how) { close $fh },
        );
    });
}

DESCRIPTION

A sink consumes every value it's given and, when they run out, gives its result: one value, such as a count or a total, for a sink of one, or a list, such as every value, for a sink of many. It's usually a pipeline's last stage, so that its result is what a run gives. Anywhere else, its result goes on through the stages after it.

A sink is made afresh for each run, by its set-up, a code reference called at the start of the run. The set-up declares, with my, whatever the run needs, and returns closures sharing it:

  • the step, called with each value, in $_ and as its argument. What it returns is ignored;

  • the flush, called with nothing, and $_ undefined, once its input runs out, returning the result. It's required. A sink of one's flush gives one value: the first it returns, or undef if it returns none;

  • optionally, the end, called once, when the run has ended, however it ended, told how: done, stopped or died. What it returns is dropped.

It returns them as a list, in that order, or by name, as a hash reference, as in {step => $step, flush => $flush}: "Made for a run" in Ekorn::Pipeline has how.

Its input runs out when the values run out, or after a stop in a stage before it. So a sink after take(5) gives its result, and is told done. A sink's step can stop the run itself, with Ekorn::Pipeline::Stop, once it has all it wants, as keep_first does: it still gives its result, and is told done. "How a run ends" in Ekorn::Pipeline has the rest.

A pipeline ending in a sink gives the sink's result. In list context, that's the list, or, from a sink of one, its one value. In scalar context, a sink of one gives its value, and a sink of many its list as an array reference:

my @all = Ekorn::Pipeline->new(keep_all)->(1, 2);    # (1, 2)
my $all = Ekorn::Pipeline->new(keep_all)->(1, 2);    # [1, 2]

A sink can be called as a code reference with values, as a pipeline of the one sink would be, and gives its result. It makes a fresh sink of its set-up for each call, so it can be kept and called again.

The words sink_one {...} and sink_many {...}, in Ekorn::Pipeline::Factories, make these from a block, and Ekorn::Pipeline::Sinks has ready-made ones.

A class of your own needn't be one of these to be a sink. Any object with an EKORN_SINK_ONE or an EKORN_SINK_MANY method, returning the step, the flush, and optionally the end, as a list or by name, is one. Offering both, or either with EKORN_STAGE, croaks where it's given as a stage, as it couldn't say which it is.

METHODS

new_sink_one

my $sink = Ekorn::Pipeline::Sink->new_sink_one($setup);

Returns a sink of one, made afresh for each run by $setup, a code reference, as "DESCRIPTION" describes. Anything else croaks. It's an Ekorn::Pipeline::Sink::One, unless the class it's called on has an EKORN_SINK_ONE method of its own, when it's of that class.

new_sink_many

my $sink = Ekorn::Pipeline::Sink->new_sink_many($setup);

Returns a sink of many, as "new_sink_one" does: an Ekorn::Pipeline::Sink::Many, or of the class it's called on, if that has an EKORN_SINK_MANY method of its own.

EKORN_SINK_ONE

my @parts = $sink->EKORN_SINK_ONE;

A sink of one's: calls the set-up, for a new run, and returns what it returns: the parts, as a list or by name. A run calls it as it starts; you needn't.

EKORN_SINK_MANY

my @parts = $sink->EKORN_SINK_MANY;

A sink of many's, as "EKORN_SINK_ONE" is a sink of one's.

DIAGNOSTICS

Expected a code reference

"new_sink_one" or "new_sink_many" was given something other than a code reference.

Expected sink #%s to return a code reference, a flush, and optionally an end

The set-up of the sink at that place in the pipeline returned something else. It's reported as the run starts. The sink is named by its position, as "DIAGNOSTICS" in Ekorn::Pipeline describes.

SEE ALSO

Ekorn::Pipeline, Ekorn::Pipeline::Source, and the ready-made sinks in Ekorn::Pipeline::Sinks

AUTHOR

Iain Campbell <cpanic@cpan.org>

LICENSE

Copyright (c) 2024-2026 by Iain Campbell.

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