p6.p6

use v6.d.PREVIEW;

module Proc::Q {
sub proc-q (
    +@commands where .so && .all ~~ List & .so,

           :@tags where .elems == @commands   && .all ~~ Cool = @commands,
           :@in   where .elems == @commands|0 && .all ~~ Cool,
    UInt   :$timeout,
    Int:D  :$batch = 8,
    Bool:D :$out   = True,
    Bool:D :$err   = True,
    Bool:D :$merge where .not | .so & $out & $err = False,

    --> Supply:D
) is export {
    supply (@commands Z @tags Z @in).batch($batch).map: -> $pack {
        my @results = $pack.map: -> ($command, $tag, $in) {
            # dd [$tag, $in, $command];
            start do with Proc::Async.new: |$command, :w($in.so) -> $proc {
                my $out-res = ''; my $err-res = ''; my $mer-res = '';
                $out and $proc.stdout.tap: $out-res ~= *;
                $err and $proc.stderr.tap: $err-res ~= *;
                if $merge {
                    $proc.stdout.tap: $mer-res ~= *;
                    $proc.stdout.tap: $mer-res ~= *;
                }
                my $prom = $proc.start;
                if $in {
                    $in ~~ Blob ?? $proc.write: $in !! $proc.print: $in;
                    $proc.close-stdin;
                }

                my $killed = False;
                $timeout.DEFINITE and Promise.in($timeout).then: {
                    $prom or try {
                        $killed = True;
                        $proc.kill: SIGHUP;
                        Promise.in(½).then: $proc.kill: SIGTERM;
                        Promise.in(1).then: $proc.kill: SIGSEGV;
                    }
                }
                $ = await $prom;
                await Promise.in(1);

                class Proc::Q::Res {
                    has Str:D  $.err    is required;
                    has Str:D  $.out    is required;
                    has Str:D  $.tag    is required;
                    has Bool:D $.killed is required;
                    has Str:D  $.merged is required;
                }.new: :err($err-res), :out($out-res), :merged($mer-res),
                  :$tag, :$killed
            }
        }

        while @results {
            await Promise.anyof: @results;
            my @ready = @results.grep: *.so;
            @results .= grep: none @ready;
            emit .status ~~ Kept ?? .result !! .cause for @ready;
        }
    }
}
}

import Proc::Q;

my @commands = <foo bar ber meow moo>;
my $sup = proc-q @commands.map({(
    $*EXECUTABLE, '-e', 'say "\qq[$_]" ~ $*IN.slurp')
}), :tags[@commands.map: 'tag-' ~ *], :in[@commands».uc], :1timeout;

my @res; react whenever $sup { @res.push: $_ }
# is @res.all, Proc::Q::Res;
dd .out for @res;