head

`Proc::Feed` provides a couple wrappers for `Proc::Async` that are more
convenient to use than the built-in `run` and `shell` subs. Specifially, these
wrapper subs let you easily feed data to them, and also feed from them to other
callables using the feed operators, `==>` and `<==`.


code

=begin code

use Proc::Feed;

# Example
my $src = './mydir';
my $dest = "$src.backup";
proc «cp -r "$src" "$dest"»;


# Example
'hello perl6' \
==> capture('md5sum')
==> split(' ')
==> *.[0]()
==> put('Checksum is: ')
;

# Example
my $text = run {
    pipe «curl http://somewhere.com/text.gz», :bin \
    ==> pipe «gunzip -c», :bin<IN>, :!chomp
    ==> join('')
}
=end code



use v6d;

unit module Proc::Feed;

class ProcHandle {
    has Proc::Async $.proc;
    has Promise $.proc-end;
    has Promise $.stdin-writes;
    has Iterable $.stdout-iterable;
    has IO::Handle $.stderr-handle;
    has Bool $.close-stderr-handle;
}

class BrokenPipeline is Exception {

    has $.error of Exception;
    has @.write-errors of Exception;
    has @.procs where Proc|Exception;

    method message { "Proc pipeline failed with errors!" }

    method gist {
        my @errors;

        if $!error {
            @errors.push: "\n--- Block exception: --------";
            @errors.push: $!error.gist;
        }

        for @!procs.kv -> $i, $proc {
            my $error;
            given $proc {
                when Exception {
                    $error = .gist;
                }
                when .?exitcode ≠ 0 {
                    $error = "Process (pipe $i) exited unsuccessfully!\n";
                    $error ~= "Command: {.command.perl}\n";
                    $error ~= "Exit code: {.exitcode}\n";
                }
            }
            if $error {
                @errors.push: "\n--- Proc failed: -----------";
                @errors.push: $error;
            }
            with @!write-errors[$i] {
                @errors.push: "\n------ Failed writing to proc:";
                @errors.push: .gist;
            }
        }
        return @errors.join("\n");
    }
}

multi _proc_impl(Str:D $cmd, |c) {
    _proc_impl(@$cmd, |c);
}

multi _proc_impl(
    @command is copy [$cmd, *@args],
    $input? is raw,

    #| Return an Iterable of data from stdout of the proc.
    Bool :$iter = False,

    #| Capture and return the stdout of the proc as a string.
    Bool :$str = False,

    #| Remove the trailing newline(s) from proc's output.
    #| Only applies when output is not binary.
    Bool :$chomp = True,

    #| Throws an exception if proc or capture failed.
    Bool :$check = True,

    #| Redirects stderr to stdout. Default is False.
    Bool :$merge is copy,

    #| Redirects stderr to a callable, a path in the file system,
    #| or an open file. This option takes precedence over :merge.
    :$stderr where Any|Callable:D|Str:D|IO::Handle:D,

    #| :bin<IN> specifies only the input will be binary blobs;
    #| :bin<OUT> specifies only the output will be binary blobs;
    #| a boolean applies to both input and output of the proc.
    :$bin where Bool:D|'IN'|'OUT' = False,

    Str:D :$enc = 'UTF-8',
    Bool :$translate-nl = True,

    :$cwd = $*CWD,
    Hash() :$env = %*ENV,

    #| If so, run the command through the shell specified;
    #| if True is specified, 'bash' will be used as the shell.
    :$shell is copy where Bool:D|Str:D = False,

    :$scheduler = $*SCHEDULER

) {
    my Bool ($bin-in, $bin-out);
    given $bin {
        when Bool  { $bin-in = $bin-out = $bin }
        when 'IN'  { $bin-in = True }
        when 'OUT' { $bin-out = True }
    }

    if $shell {
        # default the shell to 'bash' when $shell is true.
        $shell = 'bash' when Bool given $shell;
        @command := «"$shell" -c "$cmd" "$shell"», |@args;
    }

    my $proc = Proc::Async.new(
        @command,
        :w($input.DEFINITE),
        :$enc,
        :$translate-nl
    );

    my $chomp-per-line = $str ?? False !! $chomp;

    my (IO::Handle $stderr-handle, Bool $close-stderr-handle);
    with $stderr {
        $merge = False;
        when IO::Handle:D {
            $stderr-handle = $_;
        }
        when Str:D {
            $stderr-handle = open($_, :w);
            $close-stderr-handle = True;
        }
        when Callable:D {
            $proc.stderr.lines(:chomp($chomp-per-line)).tap: $_;
        }
    }
    $proc.bind-stderr($_) with $stderr-handle;

    my $stdout-iterable := do {
        my $d := $chomp-per-line;
        if $merge {
            $bin-out ?? $proc.Supply(:bin).list
                     !! $proc.Supply(:$enc, :$translate-nl).lines(:chomp($d)).list;
        } else {
            $bin-out ?? $proc.stdout(:bin).list
                     !! $proc.stdout.lines(:chomp($d)).list
        }
    } if $iter || $str;


    my $proc-end = $proc.start(:ENV($env), :$cwd);

    my Promise $stdin-writes;
    with $input {
        $stdin-writes = start {
            if $input ~~ Blob {
                await $proc.write($input);
            } else {
                my $method = $bin-in ?? "write" !! "print";
                for $input -> $data {
                    await $proc."$method"($data);
                }
            }
            LEAVE { $proc.close-stdin }
            True;
        }
    }

    my $proc-handle = ProcHandle.new(
        :proc($proc), :$proc-end,
        :$stdin-writes, :$stdout-iterable,
        :$stderr-handle, :$close-stderr-handle
    );


    if $iter || $str {
        if $str {  # it's a capture()
            my $result = $stdout-iterable.join('');

            my $*PIPED-PROCS = Channel.new;
            $*PIPED-PROCS.send: $proc-handle;
            my $proc; {
                $proc = await-procs;
                CATCH {
                    when BrokenPipeline { $proc = .procs[*-1] }
                }
            }
            if $check {
                given $proc {
                    when Proc { .sink if .exitcode ≠ 0 }
                    when Exception { .rethrow }
                }
            }
            return $chomp ?? $result.chomp !! $result;

        } else {
            # it's a pipe() from a run block; let run() do the await-procs.
            $*PIPED-PROCS.send: $proc-handle;
            return $stdout-iterable;
        }
    } else {
        # this case is for the proc() calls
        my $*PIPED-PROCS = Channel.new;
        $*PIPED-PROCS.send: $proc-handle;
        my $proc; {
            $proc = await-procs;
            CATCH {
                when BrokenPipeline { $proc = .procs[*-1] }
            }
        }
        given $proc { when Exception { .rethrow } }
        $proc.sink if $check;
        return $proc;
    }
}



sub await-procs($error?, :$SIGPIPE

False


    $*PIPED-PROCS.close;
    for $*PIPED-PROCS.list -> (
        :$proc, :$proc-end,
        :$stdin-writes, :$stdout-iterable,
        :$stderr-handle, :$close-stderr-handle
    ) {
        # Signal the child process that we are done with the pipe;
        # Do it in 1s to give the process a chance to exit.
        $proc.ready.then: {
            Promise.in(1).then: { $proc.kill(Signal::SIGPIPE) }
        } if $SIGPIPE;

        with $stdin-writes {
            @write-errors[$i++] = do {
                try await $stdin-writes;
                $! // Nil;
            }
        }

        sink $stdout-iterable.Seq if $stdout-iterable.defined;
        #= needed to ensure $proc-end will be kept.

        @procs.push: (try await $proc-end) // $!;

        $stderr-handle.close if $stderr-handle && $close-stderr-handle;
    }

    if $error ||
       @procs.any ~~ Exception:D|{.?exitcode} ||
       @write-errors.any ~~ Exception:D
    {
        BrokenPipeline.new(
            :error($error // Nil),
            :@procs, :@write-errors
        ).throw;
    }
    return @procs[*-1] if @procs;
}

multi sub run(&block, :$check = True) is export(:DEFAULT, :run) {
    my $*PIPED-PROCS = Channel.new;

    try my $result := block();
    my $error = $!;

    try await-procs($error, :SIGPIPE);
    if $check {
        .rethrow with $!;
        return $result;
    } else {
        return $result, $!;
    }
}

sub proc(
    \command where Str:D|List:D,
    $input? is raw,

    Bool :$check = False,
    :$bin where 'IN' | {! .so} = False,

    :$stderr where Any|Callable:D|Str:D|IO::Handle:D,

    :$cwd = $*CWD,
    Hash() :$env = %*ENV,
    :$shell where Bool:D|Str:D = False,
    :$scheduler = $*SCHEDULER

    --> Proc

) is export(:DEFAULT, :proc) {
    _proc_impl(command, $input, :!str, :!iter,
               :$check,
               :$bin,
               :$stderr,
               :$cwd, :$env,
               :$shell, :$scheduler
    );
}

sub capture(
    \command where Str:D|List:D,
    $input? is raw,

    Bool :$check = True,
    Bool :$chomp = True,
    :$bin where 'IN' | {! .so} = False,

    :$stderr where Any|Callable:D|Str:D|IO::Handle:D,
    Bool :$merge,

    Str:D :$enc = 'UTF-8',
    Bool :$translate-nl = True,
    :$cwd = $*CWD,
    Hash() :$env = %*ENV,
    :$shell where Bool:D|Str:D = False,
    :$scheduler = $*SCHEDULER

    --> Str:D

) is export(:DEFAULT, :capture) {

    _proc_impl(command, $input, :str, :!iter,
               :$check, :$chomp, :$stderr, :$merge,
               :$bin,
               :$enc, :$translate-nl,
               :$cwd, :$env,
               :$shell,
               :$scheduler
    );
}

sub pipe(
    \command where Str:D|List:D,
    $input? is raw,

    Bool :$chomp is copy,

    :$bin where {
        $_ ~~ Bool:D|'IN'|'OUT' and (
            ! ($_ eq 'OUT' || ($_ ~~ Bool && $_))
            || !$chomp
        )
    } = False,

    :$stderr where Any|Callable:D|Str:D|IO::Handle:D,
    Bool :$merge,

    Str:D :$enc = 'UTF-8',
    Bool :$translate-nl = True,
    :$cwd = $*CWD,
    Hash() :$env = %*ENV,
    :$shell where Bool:D|Str:D = False,
    :$scheduler = $*SCHEDULER

    --> List:D

) is export(:DEFAULT, :pipe) {

    $chomp //= do given $bin {
        when 'IN'  { True  }
        when 'OUT' { False }
        default    { !$bin }
    }
    _proc_impl(command, $input, :iter, :!str,
               :$chomp, :$stderr, :$merge, :$bin,
               :$enc, :$translate-nl,
               :$cwd, :$env,
               :$shell,
               :$scheduler
    );
}


sub each(&code, $input? is raw) is export(:each) { code($_) for $input }

multi sub map(Range \range, &code, $input? is raw) is export(:map) {
    gather for $input[range] { take code($_) }
}

sub quote(Str $s --> Str) is export(:quote) { "'{$s.subst("'", Q/'\''/)}'" }


root

README

React

=begin React :component<HeaderCol> :id<menu> 
=para L<Documentation|/doc/introduction>
=para L<Modules| /mods>
=para L<Examples| file:../examples/index.podlite>
=para L<Download |file:../download/index.podlite>
=para L<About|file:../about/index.podlite>
=para 🔍 K<⌘K>/K<ctrl-K>
=end React


=begin React :component<HeaderCol> :id<footer>  
=begin nested :!nested
=item1 B<Language>
    =item2 L<Get started|file:../getting-started/getting-started.podlite>
    =item2 L<Why Raku?|/doc/language/faq#Why-should-I-learn-Raku-What's-so-great-about-it>
    =item2 L<Try Raku|https://glot.io/new/raku>
    =item2 L<Raku cheat sheet|https://github.com/Raku/mu/blob/master/docs/Perl6/Cheatsheet/cheatsheet.txt>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<Raku on Exercism|https://exercism.org/tracks/raku>
    =item2 L<Wikipedia|https://en.wikipedia.org/wiki/Raku_(programming_language)>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=end nested
=begin nested :!nested

=item1 B<L<Documentation|/doc>>
    =item2 L<Getting started, Migration guides from other languages, & Tutorials | file:../doc/introduction.podlite>
    =item2 L<Language References|file:../doc/reference.podlite>
    =item2 L<Type Reference| file:../doc/types.podlite>
    =item2 L<Miscellaneous| file:../doc/miscellaneous.podlite>
    =item2 L<FAQs (Frequently Asked Questions)|/doc/language/faq>
    =item2 L<Community|/doc/language/community>
    =item2 L<The list of all documents|file:../doc/index.podlite>

=item1 B<Resources>
    =item2 L<The Raku Guide|https://raku.guide/>
    =item2 L<Books |https://perl6book.com/>
    =item2 L<Rosetta Code | https://www.raku.org/community/rosettacode>

=end nested
=begin nested :!nested

=item1 B<Resoures>
    =item2 L<Download | file:../download/index.podlite>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=item1 B<Learning>
    =item2 L<The Raku Guide|https://raku.guide/>
    =item2 L<Wikibook |https://en.wikibooks.org/wiki/Raku_Programming>
    =item2 L<Books |https://perl6book.com/>
    =item2 L<Rosetta Code | https://www.raku.org/community/rosettacode>
    =item2 L<Learn Raku in Y minutes|https://learnxinyminutes.com/docs/raku/>
    =item2 L<Golfing |https://github.com/AlexDaniel/raku-golf-cheatsheet>
=end nested
=begin nested :!nested

=item1 B<Explore>
    =item2 L<Raku Blog Aggregator|https://planet.raku.org/>
    =item2 L<Rakudo Weekly|https://rakudoweekly.blog/>
    =item2 L<The Weekly Challenge |https://perlweeklychallenge.org/>
    =item2 L<Raku Advent Calendar|https://raku-advent.blog/>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=end nested
=end React


=begin React :component<CookieConsent> :id<CookieConsent> :buttonCaption("Got it!")
=para
This website uses cookies for analytics.
=end React


Feed

NAME

SYNOPSIS