4 use warnings FATAL => 'all';
5 use Scalar::Util qw(blessed);
9 our @ISA = qw(Exporter);
11 our @EXPORT = qw(pmap pgrep psink);
14 warnings->unimport('void');
15 shift->export_to_level(1, @_);
18 sub pmap (&) { IO::Pipeline->from_code_map($_[0]) }
19 sub pgrep (&) { IO::Pipeline->from_code_grep($_[0]) }
20 sub psink (&) { IO::Pipeline->from_code_sink($_[0]) }
23 '|' => '_pipe_operator',
26 sub IO::Pipeline::CodeSink::print {
27 my $code = (shift)->{code};
28 foreach my $line (@_) {
35 bless({ map => [ $_[1] ] }, $_[0]);
39 my ($class, $grep) = @_;
40 $class->from_code_map(sub { $grep->($_) ? ($_) : () });
44 bless({ code => $_[1] }, 'IO::Pipeline::CodeSink');
48 my ($self, $other, $reversed) = @_;
49 if (blessed($other) && $other->isa('IO::Pipeline')) {
50 my ($left, $right) = $reversed ? ($other, $self) : ($self, $other);
51 my %new = (map => [ @{$left->{map}}, @{$right->{map}} ]);
52 die "Right hand side has a source, makes no sense"
54 $new{source} = $left->{source} if $left->{source};
55 die "Left hand side has a sink, makes no sense"
57 $new{sink} = $right->{sink} if $right->{sink};
58 return bless(\%new, ref($self));
60 my ($is, $isnt) = $reversed ? qw(source sink) : qw(sink source);
61 if (my $fail = $self->{$is}) {
62 die "Tried to add ${is} ${other} but we already had ${fail}";
64 my $new = bless({ $is => $other, %$self }, ref($self));
76 my $source = $self->{source};
77 my $sink = $self->{sink};
78 LINE: while (defined(my $line = $source->getline)) {
80 foreach my $map (@{$self->{map}}) {
81 @lines = map $map->($_), @lines;
82 next LINE unless @lines;