Piping made easy

Pied Piper of Hamelin is one of the most powerful superheros ever to walk the earth. He could lead large groups of children away by playing a tune on his pipe. Most parents struggle to lead a single child away from the telly. I’m quite sure that’s why |is called a pipe on *nix. This also means that super hero movies are just fairy tales.

It’s a very powerful tool indeed, allowing to compose programs that can handle data in the form of lines of text. If your program lacks the ability to filter its output you can just pipe to grep. You might even reuse a program you have written for a different purpose to do so.

A pipe is actually a very simple construct. We start two programs and connect STDOUT of the first with STDIN of the second. From the stand point of the programs they are writing to filehandles that where opened without a filename. Raku allows us to do so by using Proc::Async.

my $find = Proc::Async.new('/usr/bin/find', '/usr');
my $grep = Proc::Async.new('/bin/grep', 'lib');

$grep.bind-stdin: $find.stdout;

await $find.start, $grep.start;

That’s a lot more wordy than how Bash is doing it.

find /usr | grep lib

We wont get as dense as Bash. Raku is not a shell scripting language. However, in a operator oriented language we should be able to define an operator that does the binding of STDOUT and STDIN. Preferably, with starting the threads and waiting for them to finish. We want to be able to chain that operator too.

role Shell::Pipe {
    method sink {
        say [self[0].command, self[1].command, "sinking"];
        await self[1].start, self[0].start;
    }
}
my multi infix:«|>»(Proc::Async:D $out, Proc::Async:D $in) {
    $in.bind-stdin: $out.stdout;
    [$out, $in] does Shell::Pipe
}

$find |> $grep
# OUTPUT: [(/usr/bin/find /usr) (/bin/grep lib) sinking]
#         <lots of lines found by find containing 'lib'>

The chaining is the tricky part. We want to handle two cases here. Sink context and list context. Raku allows us to handle a list that is not assigned to anything or has a method called. The runtime will call the method .sink on a bare list. We can use that to tell that we have to start processing the pipe. By defining another operator we can capture the output of the whole pipe as lines of text in an Array.

my multi infix:«|>»(@pipe where @pipe ~~ Shell::Pipe, @array) {
    @pipe[1].stdout.lines.tap: -> $line is raw { @array.push: $line };
    @pipe.sink;
}
my @a;
$find |> $grep |> @a;

In this case we call .sink by hand. Chaining pipes needs a little more work. To use sink context a single pipe returns an Array with a role mixed in. We can use that to write a multi candidate that expects just that as its left operand and a Proc::Async on the right. The tricky part is that I want to be able to handle any odd Proc::Async objects. Both for connecting two of them and one of them to an Array to feed data from. I tried a custom IO::Handle but that failed because Proc::Async.bind-stdin wants to call .native-descriptor. If I feed data from an Array I don’t got that. I believe that’s a Rakudobug because if we call Proc::Async.write it just works. So it clearly don’t really need that native descriptor. I got help finding a workaround. As long as Proc::Async.w is changed befor .start-internal is called, the callback is setup properly. Sadly, that attribute does not have a public write accessor. With the help of the MOP we can write a workaround.

my multi infix:«|>»(@array where @array !~~ Shell::Pipe, Proc::Async:D $in) {
    my $h = Shell::ArrayHandle.new(:@array);
    # HERE BE DRAGONS!
    $in.^attributes.grep(*.name eq '$!w')[0].set_value($in, True);
    my $out = class {
        method command { @array.WHAT.gist ~ ' ↦ ' ~ $in.command }
        method start {
            my $p_out = start {
                LEAVE $in.close-stdin;
                $in.write: ($_ ~ "\n").encode for @array;
            }
            slip $p_out
        }
    }
    # $in.bind-stdin: $h;
    [$out, $in] does Shell::Pipe
}

my $sort = Proc::Async.new('/usr/bin/sort');
my @a;
$find |> $grep |> @a;
# fiddle-with(@a);
@a |> $sort;

If you follow my blog you likely spotted already that this is another quest item in my Bag of Holding. Looks like I got quite close to a module that makes Raku a nice replacement for Bash.

When I started to think of how to implement easy Unix pipes in Raku I expected it to be a daunting task. It wasn’t. I have come to the conclusion that “Raku” is actually a verb with the meaning “making things fall into place”.