pipeline.pl

#!/usr/bin/perl
use strict;
use warnings;
use IO::Async;
use IO::Async::Loop;
use IO::Async::Process;

my $cmd1 = [qw(ls -1)];
my $cmd2 = [qw(rev)];
my $transform = sub {
  return uc $_[0];
};

my $loop = IO::Async::Loop->new;
my ($process1, $process2);

$process1 = IO::Async::Process->new(
  command => $cmd1,
  stdout => {
    on_read => sub {
      my ($self, $buffref) = @_;
      while ($$buffref =~ s/^(.*?\n)//) {
        $process2->stdin->write(
          $transform->($1)
        );
      }
    },
  },
  on_finish => sub {
    my ($self, $exit) = @_;
    warn "process1 exited with code $exit\n";
    $process2->stdin->close;
  }
);

$process2 = IO::Async::Process->new(
  command => $cmd2,
  stdin => {
    via => 'pipe_write',
  },
  stdout => {
    on_read => sub {
      my ($self, $buffref) = @_;
      while ($$buffref =~ s/^(.*?\n)//) {
        print "Output: $1";
      }
    },
  },
  on_finish => sub {
    my ($self, $exit) = @_;
    warn "process2 exited with code $exit\n";
    $loop->stop;
  }
);

$loop->add($process1);
$loop->add($process2);
$loop->run;