NAME
Ekorn::Pipeline - run values through a list of stages
SYNOPSIS
use Ekorn::Pipeline;
my $p = Ekorn::Pipeline->new(
sub { $_->{active} ? $_ : () }, # filter
sub ($user) { @{ $user->{roles} } }, # flatMap
sub { uc }, # map
);
# Push: values in, a list out
my @roles = $p->(@users);
# Pull: one value per call, from a source
my $next = $p->iterator(sub { shift @queue // () });
while (my($role) = $next->()) { ... }
# A function of your own, taking stages from its caller
sub lines_of ($file, @stages) {
open my $fh, '<', $file or die "Can't read $file: $!\n";
return Ekorn::Pipeline->new(@stages)->iterator(sub { scalar(<$fh>) // () });
}DESCRIPTION
A pipeline is a list of stages, each a code reference. A stage is called with one value, as its argument and in $_, so it can name the value with a signature, sub ($row) { $row->[1] }, or leave the signature out and use $_, sub { $_->[1] }. $_ is localised for each call, so the caller's is left as it was. A stage returns a list:
nothing, to drop the value (a filter);
one value, to pass it on (a map);
several, each of which goes through the rest of the stages on its own (a flatMap).
Each value goes through every stage before the next one starts, so the values come out in order, and a stage sees them in the order a for loop would.
A stage can end the pipeline early with Ekorn::Pipeline::Stop. The values already out are kept, any values the stop carries go on through the rest of the stages, and nothing else is processed:
sub ($row) { $row->{id} == $wanted ? Stop($row) : () }A stage made with Ekorn::Pipeline::Gather is made afresh for each run of the pipeline, and can have a function called once the values run out, whose values go through the rest of the stages. Each push of values, and each iterator, is a run.
Stages from your caller
A pipeline is for the author of any function that hands out values. The function takes stages from its caller, as its last arguments, and runs what it would return through them, so the caller says what they want done with each value, beside the call, and gets back only what they want:
sub bands (@stages) {
return Ekorn::Pipeline->new(@stages)->(load_bands());
}
my @names = bands(sub { $_->{name} }, take(5));With "iterator", the values are made one at a time, as the caller asks for them, so a function reading a file or a socket reads no more than it has to. Once a stage has stopped the pipeline, the source isn't called again.
The canned stages, in Ekorn::Pipeline::Stages::Flow, Ekorn::Pipeline::Stages::Batch and Ekorn::Pipeline::Stages::Taps, work in any pipeline, and Ekorn::Pipeline::Pipes gives a class of your own pipe and tap methods, which run the object through stages.
METHODS
new
my $p = Ekorn::Pipeline->new(@stages);Returns a pipeline running values through @stages, in order. It croaks if a stage isn't a code reference. With no stages, a pipeline returns its values as they are.
A pipeline is an object that can be called as a code reference. Called with values, it returns the list that comes out of the last stage:
my @results = $p->(@values);In scalar context, it returns how many results came out, as an array would:
my $count = $p->(@values);iterator
my $next = $p->iterator($source);Returns a closure handing out, one per call, the values that come out of the stages. $source is a code reference returning the next value or values on each call, and an empty list when there are no more; it's called only when the values from its last call have all been handed out.
The closure returns an empty list when the values run out, or once a stage has stopped the pipeline and the values its stop carried are handed out, and on every call after that. After a stop, the source isn't called again. Read it in list context, so an undef value isn't mistaken as the end:
while (my($x) = $next->()) { ... }The end functions of stages made with Ekorn::Pipeline::Gather are called once the source runs out, or after a stop, before the closure returns an empty list.
If a stage or an end function dies, the call dies with it, and the iterator has ended: every call after returns an empty list, nothing waiting goes on through the stages, no end function is called, and the source isn't called again, as when a pushed run dies. To skip a bad value and carry on, catch the error inside the stage.
stages
my @stages = $p->stages;Returns the pipeline's stages, in order, as given to "new". In scalar context, it returns how many there are.
A pipeline isn't a stage, but its stages can be spliced into another pipeline, or given to any function that takes stages, so a pipeline made once can be used wherever stages go:
my $clean = Ekorn::Pipeline->new(sub { s( \A \s+ | \s+ \z )()xgr }, sub { length ? $_ : () });
my @titles = Ekorn::Pipeline->new($clean->stages, take(10))->(@raw);The stages run in the other pipeline's run, as if listed there: a stop in one ends that run, and a stage made with Ekorn::Pipeline::Gather is made afresh for each run, so its state isn't shared between the pipelines it's spliced into. A plain closure is the same code reference wherever it goes, so whatever it keeps is shared; keep state in gather. A stage that dies is named by its place in the pipeline it was spliced into.
FUNCTIONS
ending
Ekorn::Pipeline::ending()True while the values going through the stages come from an end function, rather than from the source. It's for code that tells its stages about the value they're given, such as the value it came from, which a value an end function gives doesn't have. It isn't exported.
DIAGNOSTICS
- Expected stage #%d to be a code reference
-
A stage given to "new" wasn't a code reference.
- Expected a code reference
-
The source given to "iterator" wasn't a code reference.
- Expected stage #%d, made with gather, to return one or two code references
-
The block given to "gather" in Ekorn::Pipeline::Gather for that stage returned something other than a stage, or a stage and an end function.
- Pipeline stage #%d failed at %s line %d:
-
A stage, or its end function, died. The message it died with follows, indented. An exception object is passed on as it is.
SEE ALSO
Ekorn::Pipeline::Gather, Ekorn::Pipeline::Stop, Ekorn::Pipeline::Pipes, and DBIx::Squirrel, where the pipeline began
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.