[![Actions Status](https://github.com/nukopian/Ekorn-Pipeline/actions/workflows/test.yml/badge.svg?branch=main)](https://github.com/nukopian/Ekorn-Pipeline/actions?workflow=test) [![MetaCPAN Release](https://badge.fury.io/pl/Ekorn-Pipeline.svg)](https://metacpan.org/release/Ekorn-Pipeline)
[![Kwalitee](https://cpants.cpanauthors.org/dist/Ekorn-Pipeline.svg)](https://cpants.cpanauthors.org/dist/Ekorn-Pipeline)

# Ekorn-Pipeline

Stages for any function: let the caller finish the job.

A function that hands out values can't know everything its callers
will want done with them. With Ekorn::Pipeline, it takes a list of
_stages_ from its caller: small functions that each see one value at
a time. It runs what it would return through them. The caller says
what they want, beside the call, and the function does only what it
alone can.

    use Ekorn::Pipeline;
    use Ekorn::Pipeline::Stages::Flow;    # take and drop

    # The author's function: it reads log lines, and nothing else.
    sub log_lines ($file, @stages) {
        open my $fh, '<', $file or die "Can't read $file: $!\n";
        return Ekorn::Pipeline->new(@stages)
            ->iterator(sub { scalar(<$fh>) // () });
    }

    # A caller's use of it: the first ten errors, as their messages.
    my $errors = log_lines(
        'app.log',
        sub { m( \A ERROR \  (.*) )x ? $1 : () },    # only errors, as messages
        take(10),                                    # ten, then stop reading
    );
    while (my($message) = $errors->()) { say $message }

## The problem it solves

Every function that returns a list meets the same pressure. One caller
wants only some of the values. Another wants them in a different
shape, a third wants the first ten, and a fourth wants them in batches
of 500. The author has two ways to answer, and both cost something.

- **Grow options.** `active_only`, `fields`, `limit`,
`offset`, `as_hash`, `batch_size`, `callback`. Each is code the
author writes, tests, documents and maintains. Each interacts with the
others: does `limit` count before or after `active_only`? The list
is never finished, because the next caller wants something nobody
thought of. And every caller has to read about options they'll never
use.
- **Leave it to the caller.** The function returns everything, and
each caller writes a loop around the call: a temporary array, a
counter, a flag, a `last`. The calling code fills with noise, and the
same few loops are written over and over. Worse, the function can't
stop early: by the time the caller has the values, all the work of
making them is done.

Stages are a third way. The function takes a list of them as its last
arguments, and runs each value it makes through them. The caller's
processing happens _inside_ the function, one value at a time, as
the values are made:

    # Options
    sub users (%opt) {
        my @users = load_users();
        @users = grep { $_->{active} } @users if $opt{active_only};
        @users = map  { $_->{name} } @users   if $opt{names_only};
        @users = splice @users, 0, $opt{limit} if $opt{limit};
        return @users;
    }

    # Stages
    sub users (@stages) {
        return Ekorn::Pipeline->new(@stages)->(load_users());
    }

    my @names = users(sub { $_->{active} ? $_ : () }, sub { $_->{name} }, take(10));

The second `users` does everything the first does, and everything it
doesn't, without a line more. Its author is done with it. Whatever a
caller wants next, they can write as a stage, without asking.

## What everyone gets

- **The caller says it once, beside the call.** No loop, no
temporary array, no counter, no flag. What the caller wants done is
written where they ask for the values, in the order it happens, and
reads as a list of steps.
- **The work stops when the caller has enough.** A stage that
stops ends the function's work too. A caller who wants ten results
reads ten lines, or fetches one page, not all of them. Nothing is held
between stages, so a function over a million values needs no more
memory than one over ten.
- **The author keeps an interface that doesn't grow.** One list of
stages replaces every option a caller might ask for next. The
function's signature, documentation and tests stay the size of what
only it can do. A new need is met in the caller's code, and breaks
nobody else's.
- **The author keeps control of what only it should touch.** The
function still opens and closes the file, fetches the pages, and
handles the connection or transaction. The caller shapes the values
without ever seeing the handle they came from.
- **Order answers the questions options raise.** Does `limit`
count before or after `active_only`? With stages, they run in the
order the caller wrote them.
- **A stage is written once, for every function.** Stages are
plain code references, or pipelines of them. A stage written for one
function works with
every function that takes stages, whoever wrote it. The canned stages
here work with all of them.

## Stages

A stage is either a code reference or another pipeline.

A code reference is called with one value, as its argument and in
`$_`, and returns a list:

- **nothing,** to drop the value: a _filter_;
- **one value,** to pass it on, changed or not: a _map_;
- **several,** each of which goes on through the rest of the
stages by itself: a _flatMap_.

One shape covers all three. A pipeline is just a list of stages, and a
stage needs no wrapper to say which kind it is:

    sub { $_ % 2 ? $_ : () }                # filter: the odd numbers
    sub { $_ * 10 }                         # map: ten times each
    sub ($user) { @{ $user->{roles} } }     # flatMap: each of a user's roles

Returning a list, rather than a value or `undef`, is what makes this
work. An empty list plainly means "nothing", so `undef`, `0` and the
empty string are ordinary values, and can be passed on.

A pipeline used as a stage stands for its own stages, spliced in where
it's given. So stages you've put together once can be reused, in
another pipeline or with any function that takes stages:

    my $active_names = Ekorn::Pipeline->new(sub { $_->{active} ? $_ : () }, sub { $_->{name} });
    my @names = users($active_names, take(10));

If a function accepts only code references, give it the pipeline's
stages instead: `users($active_names->stages, take(10))`.

## One value at a time

Each value goes through every stage before the next one starts, as it
would in a `for` loop. The values come out in order, and each stage
sees them in order. Nothing is held between stages, so a pipeline over
a million values needs no more memory than one over ten.

A pipeline runs two ways:

- **Pushed:** called with values, it returns the list that comes
out of the last stage.

        my @results = $pipeline->(@values);

- **Pulled:** given a _source_, a function that returns the next
value each time it's called, `iterator` returns a function that hands
out the results one per call. The source is called only when a result
is wanted. A function reading a file, a socket or a database reads
only as much as its caller asks for.

        my $next = $pipeline->iterator(sub { @queue ? shift @queue : () });
        while (my($result) = $next->()) { ... }

## Stopping early

A stage can end the pipeline with [Ekorn::Pipeline::Stop](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AStop). The values
already out are kept, and any values the stop carries go on through
the rest of the stages. Nothing else is made: once a stage has stopped
it, a pulled pipeline never calls its source again.

    use Ekorn::Pipeline::Stop;

    sub ($user) { $user->{id} == $wanted ? Stop($user) : () }    # find one, then stop

This is how a caller's stages save the author's work. With
`take(10)`, the function in the first example reads up to the tenth
error line, and leaves the rest of the file unread. Nobody wrote an
option for it, and the caller wrote no `last`.

It matters more where each value costs more. A function over a paged
web API fetches a page only when its caller wants a result from it:

    sub search ($query, @stages) {
        my $page = 0;
        return Ekorn::Pipeline->new(@stages)->iterator(sub {
            return @{ fetch_page($query, ++$page) };    # a request; empty after the last page
        });
    }

    my $hits = search('squirrels', take(10));
    while (my($hit) = $hits->()) { say $hit->{title} }    # one request, if a page holds ten

Without stages, that function either fetches every page, or grows
`max_results`, `max_pages` and `on_result` options, for its author
to write and maintain.

## Stages with state, and with an end

A stage is a closure, so it can keep state, such as a count. But a
plain closure keeps its state from one run to the next, and can't tell
when the last value has gone by. [Ekorn::Pipeline::Gather](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AGather) makes a
stage afresh at the start of each run. The stage can also have an
_end function_, called once the values run out, whose values go on
through the rest of the stages:

    use Ekorn::Pipeline::Gather;

    sub batches_of ($size) {
        return gather {
            my @batch;    # this run's
            return (
                sub ($x) { push @batch, $x; @batch == $size ? [splice @batch] : () },
                sub ()   { @batch ? [@batch] : () },    # the last, short batch
            );
        };
    }

That's how `batch`, `collect`, `take` and `drop` are made.

## Stages are values

Stages are code references or pipelines, so they can be kept in
variables, built by functions and passed around. A function that
returns a stage is a word in a vocabulary, and the vocabulary grows in
the caller's code, not the author's:

    sub only_active ()    { sub { $_->{active} ? $_ : () } }
    sub value_of ($name)  { sub { $_->{$name} } }

    my @page  = (drop(20), take(10));
    my @names = users(only_active, value_of('name'), @page);

A stage written for `users` works for `log_lines` too, or for a
function on CPAN whose author has never heard of it.

## Writing a function that takes stages

- **Take the stages last.** The function's own arguments come
first, then the stages: `sub log_lines ($file, @stages)`. A caller
who wants nothing done passes no stages. A pipeline with no stages
returns its values unchanged, so there's nothing to special-case.
- **Push what you have; pull what you make.** If the values are
already in memory, call the pipeline with them. If they're made one at
a time, from a file, a socket, a database or a computation, give
`iterator` a source and return the iterator. Then a stage that stops
saves the work of making the rest.
- **Let the stages do the optional work.** If you're about to add
an option that filters, reshapes or limits what's returned, a stage
does it already, and it's one less thing for you to maintain. Keep the
arguments only the function can act on, such as which file to read, or
a filter a database can apply with an index.
- **Report your callers' mistakes at their lines.** Put
`@Ekorn::Pipeline::Errors::INTERNAL` in your module's `@CARP_NOT`.
Then a stage that dies is reported at the caller's line, where the
stage was written, not at the line in your module that ran the
pipeline.
- **Find the stages with `is_stage`.** If the function takes a
list of values of its own before the stages, take the stages off the
end with `is_stage`, from Ekorn::Pipeline. It's the test `new` uses,
so your callers can pass a pipeline as a stage, too. The values
mustn't be code references or pipelines themselves.

A class of your own can have `pipe` and `tap` methods, from
[Ekorn::Pipeline::Pipes](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3APipes), which run an object through stages. `pipe`
returns what comes out. `tap` returns the object, so a chain of
method calls can carry on after it.

## Errors

A stage that dies is named by its position in the pipeline, and
reported at the line of the caller's code that ran it, with its
message below:

    Pipeline stage #2 failed at app.pl line 12:
        No name for user 7 at app.pl line 9.

An exception object is passed on unchanged. Either way the run ends,
pushed or pulled, and an iterator returns nothing more. A mistake in
making a pipeline or a canned stage, such as a stage that isn't a code
reference or `take(-1)`, is also reported at the caller's line.

## Canned stages

- [Ekorn::Pipeline::Stages::Flow](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AStages%3A%3AFlow): `take`, `drop`,
`take_while` and `drop_while`, which pass values on by their
position or by a test.
- [Ekorn::Pipeline::Stages::Batch](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AStages%3A%3ABatch): `batch` and `collect`,
which gather values and pass them on together.
- [Ekorn::Pipeline::Stages::Taps](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AStages%3A%3ATaps): `peek`, `count_into` and
`dump_to`, which look at each value and pass it on, for logging,
counting and debugging.

## Stage modules of your own

Stages are plain code references, or pipelines of them, so anyone can
make them, in any namespace. Where a module of stages belongs depends on what its stages
need to know:

- **Stages for any values** go under `Ekorn::Pipeline::Stages`,
like the canned stages. They work whatever the values are and wherever
they come from.
- **Stages for one kind of value, or one source,** go with the
module that knows about it, under its namespace. DBIx::Squirrel's
stages for rows, `pluck` and `as_hash`, need a statement's column
names, so they're in `DBIx::Squirrel::Stages::Rows`. Stages for a log
format of your own might be in `My::Log::Stages`.

Wherever it goes, a module of stages is built from the same parts as
the canned stages:

- [Ekorn::Pipeline::Gather](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AGather), for state and end functions;
- [Ekorn::Pipeline::Stop](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AStop), to end early;
- `croakf`, from [Ekorn::Pipeline::Errors](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AErrors), to report a mistake
in making a stage at the line of the caller who made it.

Here's a small one:

    package My::Stages::Words;

    use v5.36;
    use Exporter qw(import);
    use Ekorn::Pipeline::Errors qw(croakf);

    our @EXPORT_OK = qw(longer_than);

    sub longer_than ($length) {
        croakf 'Expected a length of 0 or more, not %s', $length // 'undef'
            unless defined $length && $length =~ m( \A \d+ \z )xa;
        return sub ($word) { length $word > $length ? $word : () };
    }

## When to use something else

- **Code with no other caller.** Where the values are made and
used in one place, by one author, a plain `for`, `grep` or `map` is
clearer. Stages pay off where a function and its callers are written
separately.
- **Tight loops over cheap values.** Each stage costs a
subroutine call for each value. That's nothing beside reading a file
or a socket, but it shows in a loop doing arithmetic on millions of
numbers in memory.
- **Work that needs every value at once,** such as sorting. A
stage sees one value at a time, so a sort has to wait for them all, as
`collect { sort @_ }` does, and saves nothing. Where the source can
sort, as a database can with `ORDER BY`, let it.

## Where it came from

The pipeline began in [DBIx::Squirrel](https://metacpan.org/dist/DBIx-Squirrel),
where stages written beside a query shape the rows it returns. None of
it needs a database, so it's here on its own, for any function to use,
and to grow beyond what rows need. DBIx::Squirrel now uses it, in
place of its own copy, from v2.0.5.

## Documentation

[The how-to](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AManual%3A%3AHowTo) shows how to do common
tasks, one recipe at a time, for callers, function authors and stage
authors. [The reference](https://metacpan.org/pod/Ekorn%3A%3APipeline%3A%3AManual%3A%3AReference) describes
every module, function and method in full, and how a pipeline
behaves, with notes for stage authors. Each module's POD, starting
with [Ekorn::Pipeline](https://metacpan.org/pod/Ekorn%3A%3APipeline), says the same more briefly, and is the
authority where they differ.

## Installing

From CPAN:

    cpanm Ekorn::Pipeline

Or the latest from GitHub:

    cpanm https://github.com/nukopian/Ekorn-Pipeline.git

It needs Perl 5.36 or later.

## Licence

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. See the file `LICENSE`.

## Acknowledgements

Written by Iain Campbell, with Claude, Anthropic's AI model. Claude
reviews the code, and writes parts of it where Iain asks, often from
his code in DBIx::Squirrel. It also helps with the Git Ops, the
repository's setup, and the tests and documentation.