NAME

Ekorn::Pipeline::Gather - a stage made afresh for each run, told when the values run out

SYNOPSIS

use Ekorn::Pipeline::Gather;

# Values in batches of $size, the last batch given at the end
sub batches_of ($size) {
    return gather {
        my @batch;    # this run's
        return (
            sub ($x) { push @batch, $x; @batch == $size ? [splice @batch] : () },
            sub ()   { @batch ? [@batch] : () },
        );
    };
}

my $next = Ekorn::Pipeline->new(batches_of(500))->iterator($source);
while (my($batch) = $next->()) { ... }

DESCRIPTION

A stage is a code reference, called with each value in turn. It can't tell when the last value has gone by, and whatever it keeps in a closure stays there from one run of its pipeline to the next. A stage that gathers values, to give them in batches or all together, needs both: state of its own on each run, and a call at the end.

gather makes such a stage. Its block is called at the start of each run of a pipeline, so whatever the block declares is fresh on each run: each call of a pipeline with values, and each iterator made from it. The block returns:

• the stage, called with each value as any stage is;

• optionally, a function called with no arguments, and $_ undefined, once the values have run out.

What the end function returns goes through the rest of the stages, as what a stage returns does: nothing, one value, or several. It's called even when there were no values, so an end can give an empty result, such as []. The ends are called in order, first stage first, and what one returns goes through the stages after it before the next end is called, so an end later in the pipeline gets everything sent to it.

A stage that keeps state of any kind, such as a count, should be made with gather, even without an end function, so that it starts afresh on each run.

Stopping

After a Ekorn::Pipeline::Stop, the ends of the stages after the one that stopped are called, so they get the values before the stop, and any it carries; those of the stage that stopped, and the stages before it, are not, as nothing else is processed. An end can stop the pipeline too: the values its stop carries go on, and the ends after it are still called.

Context

Inside an end function, and the stages it sends values to, "ending" in Ekorn::Pipeline is true, so code that tells its stages about the value they're given can say that the value comes from many, not one.

FUNCTIONS

gather

my $stage = gather { ...; return ($stage, $end) };

Returns a stage that a pipeline makes for each run by calling the block. It's a code reference, blessed into Ekorn::Pipeline::Gather, so it can be given wherever a stage can. gather is exported by default; to keep it out of your namespace:

use Ekorn::Pipeline::Gather ();

DIAGNOSTICS

Expected a code reference

gather was given something other than a block or code reference.

Expected stage #%d, made with gather, to return one or two code references

A block given to gather returned something other than a stage, or a stage and an end function. Reported when a run starts: when the pipeline is called with values, or an iterator is made from it.

SEE ALSO

Ekorn::Pipeline, Ekorn::Pipeline::Stop, and the canned stages, many made with it: Ekorn::Pipeline::Stages::Flow, Ekorn::Pipeline::Stages::Batch and Ekorn::Pipeline::Stages::Taps

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.