Fallback

What if a tube throws an exception? It might be time to setup a fallback!

DIY

The do it yourself approach is always a valid one, especially if you need to do something fancy. If you start your program via tubergen you will already have Try::Tiny available out of the box, so why not use it straight away?

use Try::Tiny;
my $tube_that_might_throw = sub {
   die {message => 'whatever!'} if rand(1) < 0.5;
   return shift;
};
my $wrap_or_passthrough = sub {
   my $record = shift;
   my @retval;
   try   { @retval = $tube_that_might_throw->($record); }
   catch { @retval = $record; };
   return @retval;
};

Well... your business logic might be a bit more complicated than simply passing through, but you get the idea: with a simple wrapper tube you can avoid interrupting the whole business just because one tube choked. Yay!

Fallback

One interesting thing to do when a tube fails so spectacularly that it throws an exception is to possibly send the input to some fallback pipeline. For example, you might setup a pipeline for the good records, and another one for the bad ones (e.g. to send them to a file for later inspection).

Let's start from the happy path, that we will call $tube_for_good. This will start from the already-read data in the raw field, and do all following processing (parsing, rendering and writing):

my $tube_for_good = pipeline(
   ['Parser::by_format' => 'id,artist,title'],
   ['Renderer::with_template_perlish' => <<'END'],
[% artist %] (#[% id %]): [% title %]
END
   'Writer::to_files',
);

Unfortunately, there's been a little misunderstanding with the production, in that they believe that they can use commas and semicolons interchangeably, so they provided us the following input:

1,Pearl Jam,Evenflow
2,Prince,Purple Rain
3;Take That;It Only Takes a Minute
4,Soundgarden,Black Hole Sun

It's evident: Take That are misplaced and will cause trouble in the following example:

#!/usr/bin/env perl
# vim: sts=3 ts=3 sw=3 et ai :
use strict;
use warnings;
use Data::Dumper;
$Data::Dumper::Indent = 1;
use Data::Tubes qw< pipeline >;
use Try::Tiny;

my $input = <<'END';
1,Pearl Jam,Evenflow
2,Prince,Purple Rain
3;Take That;It Only Takes a Minute
4,Soundgarden,Black Hole Sun
END

my $tube_for_good = pipeline(
   ['Parser::by_format'               => 'id,artist,title'],
   ['Renderer::with_template_perlish' => <<'END'],
[#[% id %]] [% artist %] - [% title %]
END
   'Writer::to_files',
   {tap => 'sink'},
);

try {
   pipeline(
      'Source::open_file', 'Reader::by_line',
      $tube_for_good, {tap => 'sink'}
   )->(\$input);
} ## end try
catch {
   die "exception: $_->{message}\n   for '$_->{record}{raw}'\n";
};

Here is what happens:

shell$ ./fallback-00
[#1] Pearl Jam - Evenflow
[#2] Prince - Purple Rain
exception: 'parse by format': invalid record, expected 3 items, got 1
   for '3;Take That;It Only Takes a Minute'

What we want is to setup a special tube where we can send all these troublesome input records, while still processing the good ones. Setting up the $tube_for_bad is pretty easy:

my $tube_for_bad = pipeline(
   sub {
      my $record = shift;
      my $name = $record->{source}{name};
      my $raw  = $record->{raw};
      $record->{rendered} = "$name $raw\n";
   },
   ['Writer::to_files', 'exceptions.txt'],
   {tap => 'sink'},
);

Now, we can setup a fallback from the good one to the bad one like in the following example:

#!/usr/bin/env perl
# vim: sts=3 ts=3 sw=3 et ai :
use strict;
use warnings;
use Data::Dumper;
$Data::Dumper::Indent = 1;
use Data::Tubes qw< pipeline >;
use Try::Tiny;

my $input = <<'END';
1,Pearl Jam,Evenflow
2,Prince,Purple Rain
3;Take That;It Only Takes a Minute
4,Soundgarden,Black Hole Sun
END

$|++;
my $tube_for_good = pipeline(
   ['Parser::by_format'               => 'id,artist,title'],
   ['Renderer::with_template_perlish' => <<'END'],
[#[% id %]] [% artist %] - [% title %]
END
   'Writer::to_files',
   {tap => 'sink'},
);

my $tube_for_bad = pipeline(
   sub {
      my $record = shift;
      my $name   = $record->{source}{name};
      my $raw    = $record->{raw};
      $record->{rendered} = "$name $raw\n";
      return $record;
   },
   ['Writer::to_files', 'exceptions.txt'],
   {tap => 'sink'},
);

pipeline(
   'Source::open_file',    # source, as before
   'Reader::by_line',      # reader, as before
   ['Plumbing::fallback', $tube_for_good, $tube_for_bad,],
   {tap => 'sink'},
)->(\$input);

Now, the result is the following:

shell$ ./fallback-01
[#1] Pearl Jam - Evenflow
[#2] Prince - Purple Rain
[#4] Soundgarden - Black Hole Sun
shell$ cat exceptions.txt
scalar:SCALAR(0x9a40a90) 3;Take That;It Only Takes a Minute

where the SCALAR(...) is the reference to the string used as input.

Beware Of Iterators

You surely noticed that both $tube_for_good and $tube_for_bad are built using pipeline where we set the tap argument to sink, i.e. throw away results. Why?

Remember that, by default, pipeline generates a tube that returns an iterator. This gets in the way of using fallback properly, because an iterator means deferred execution, i.e. execution that might happen outside of the fallback mechanism. For this reason, you should ensure that the tubes you wrap in fallback do not return iterators, e.g. by explicitly exhausting them.

As of release 0.731 it is possible to set the tap argument to both sink (for just throwing away output records) or bucket, for collecting them all as a new option. In this case, any iterator is automatically turned into one of the other possible return values, i.e. the empty list, a single record or a pair with the string records followed by an array reference holding multiple output records.

As Many As You Want

The fallback mechanism is mostly useful to guard one single pipeline from interrupting the whole business when one element is not compliant. What if the fallback tube throws an exception too? Just add more: each of them will be wrapped and tried when the previous one throws an exception, and the whole tube will be safe anyway. It might even be used as a poor's man retry scheme:

my $tube_that_might_throw = sub {
   die {message => 'whatever!'} if rand(1) < 0.5;
   return shift;
};
pipeline(
   [
      'Plumbing::fallback',
      $tube_that_might_throw,
      $tube_that_might_throw,
      $tube_that_might_throw
   ],
   {tap => 'sink'},
)->({});

Catch That Exception!

One last interesting aspect of fallback is that you get the chance to play with the exception when it is catched. You can pass a sub reference with argument catch to do this; when an exception is trapped, the sub reference will be called with two parameters, namely the exception itself and the offending input record.

One possible use for this might be for setting a delay in the poor man's retry scheme, like this:

pipeline(
   [
      'Plumbing::fallback',
      $tube_that_might_throw,
      $tube_that_might_throw,
      $tube_that_might_throw,
      {
         catch => sub {
            my ($exception, $record) = @_;
            sleep(3 + rand(5)) if $record->{retry}++ < 2;
         },
      }
   ],
   {tap => 'sink'},
)->({});

You can modify the record inside the catch sub, because it will be passed over to the following invocations. This allows you to keep track of what's going on, like in the example above where we use retry to avoid sleeping if/when the last try fails.

This allows you to also augment the record with the exception itself, so that following tubes can use it. This is the same example as before, but inside the exceptions.txt file we now also record the error message:

#!/usr/bin/env perl
# vim: sts=3 ts=3 sw=3 et ai :
use strict;
use warnings;
use Data::Dumper;
$Data::Dumper::Indent = 1;
use Data::Tubes qw< pipeline >;
use Try::Tiny;

my $input = <<'END';
1,Pearl Jam,Evenflow
2,Prince,Purple Rain
3;Take That;It Only Takes a Minute
4,Soundgarden,Black Hole Sun
END

$|++;
my $tube_for_good = pipeline(
   ['Parser::by_format'               => 'id,artist,title'],
   ['Renderer::with_template_perlish' => <<'END'],
[#[% id %]] [% artist %] - [% title %]
END
   'Writer::to_files',
   {tap => 'sink'},
);

my $tube_for_bad = pipeline(
   sub {
      my $record  = shift;
      my $name    = $record->{source}{name};
      my $raw     = $record->{raw};
      my $message = $record->{exception}{message};
      $record->{rendered} = "$name $raw\n -> $message\n";
      return $record;
   },
   ['Writer::to_files', 'exceptions.txt'],
   {tap => 'sink'},
);

pipeline(
   'Source::open_file',    # source, as before
   'Reader::by_line',      # reader, as before
   [
      'Plumbing::fallback',
      $tube_for_good,
      $tube_for_bad,
      {
         catch => sub { $_[1]{exception} = $_[0]; }
      }
   ],
   {tap => 'sink'},
)->(\$input);

Output:

shell$ ./fallback-02
[#1] Pearl Jam - Evenflow
[#2] Prince - Purple Rain
[#4] Soundgarden - Black Hole Sun
shell$ cat exceptions.txt 
scalar:SCALAR(0x9618948) 3;Take That;It Only Takes a Minute
 -> 'parse by format': invalid record, expected 3 items, got 1