view release on metacpan or  search on metacpan

lib/Future/Workflow/Pipeline.pm  view on Meta::CPAN

   });

   # A synchronous (in-process) stage; e.g. some HTML parsing
   $pipeline->append_stage_sync( sub ($response) {
      my $dom = Mojo::DOM->new( $response->decoded_content );
      return $dom->at('div[id="main"]')->text;
   });

   # A detached (out-of-process/thread) stage; e.g. some silly CPU-intensive task
   $pipeline->append_stage_detached( sub ($text) {
      my $iter = Algorithm::Permute->new([ split m/\s+/, $text ]);

      my $best; my $bestscore;
      while(my @words = $iter->next) {
         my $str = join "\0", @words;
         my $score = md5sum( $str );
         next if defined $bestscore and $score ge $bestscore;

         $best      = $str;
         $bestscore = $score;
      }

lib/Future/Workflow/Pipeline.pm  view on Meta::CPAN

   # An asynchronous output
   my $dbh = Database::Async->new( ... );
   $pipeline->set_output_async( async sub ($best) {
      await $dbh->do('INSERT INTO Results VALUES (?)', $best);
   });

   # A synchronous output
   $pipeline->set_output_sync( sub ($best) {
      print "MD5 minimized sort order is:\n";
      print "  $_\n" for split m/\0/, $best;
   });


   # 4: Now start it running on some input values

   foreach my $url (slurp_lines("urls.txt")) {
      await $pipeline->push_input($url);
   }