lib/Object/Remote/Future.pm

   1 package Object::Remote::Future;
   2
   3 use strict;
   4 use warnings;
   5 use base qw(Exporter);
   6
   7 use Object::Remote::Logging qw( :log Dlog_trace );
   8
   9 use CPS::Future;
  10
  11 our @EXPORT = qw(future await_future await_all);
  12
  13 sub future (&;$) {
  14   my $f = $_[0]->(CPS::Future->new);
  15   return $f if ((caller(1+($_[1]||0))||'') eq 'start');
  16   await_future($f);
  17 }
  18
  19 our @await;
  20
  21 sub await_future {
  22   my $f = shift;
  23   log_trace { my $ir = $f->is_ready; "await_future() invoked; is_ready: $ir" };
  24   return $f if $f->is_ready;
  25   require Object::Remote;
  26   my $loop = Object::Remote->current_loop;
  27   {
  28     local @await = (@await, $f);
  29     $f->on_ready(sub {
  30       log_trace { my $l = @await; "future has become ready, length of \@await: '$l'" };
  31       if ($f == $await[-1]) {
  32         log_trace { "This future is not waiting on anything so calling stop on the run loop" };
  33         $loop->stop;
  34       }
  35     });
  36     log_trace { "Starting run loop for newly created future" };
  37     $loop->run;
  38   }
  39   if (@await and $await[-1]->is_ready) {
  40     log_trace { "Last future in await list was ready, stopping run loop" };
  41     $loop->stop;
  42   }
  43   log_trace { "await_future() returning" };
  44   return wantarray ? $f->get : ($f->get)[0];
  45 }
  46
  47 sub await_all {
  48   log_trace { my $l = @_; "await_all() invoked with '$l' futures to wait on" };
  49   await_future(CPS::Future->wait_all(@_));
  50   map $_->get, @_;
  51 }
  52
  53 package start;
  54
  55 our $start = sub { my ($obj, $call) = (shift, shift); $obj->$call(@_); };
  56
  57 sub AUTOLOAD {
  58   my $invocant = shift;
  59   my ($method) = our $AUTOLOAD =~ /^start::(.+)$/;
  60   my $res;
  61   unless (eval { $res = $invocant->$method(@_); 1 }) {
  62     my $f = CPS::Future->new;
  63     $f->fail($@);
  64     return $f;
  65   }
  66   unless (Scalar::Util::blessed($res) and $res->isa('CPS::Future')) {
  67     my $f = CPS::Future->new;
  68     $f->done($res);
  69     return $f;
  70   }
  71   return $res;
  72 }
  73
  74 package maybe;
  75
  76 sub start {
  77   my ($obj, $call) = (shift, shift);
  78   if ((caller(1)||'') eq 'start') {
  79     $obj->$start::start($call => @_);
  80   } else {
  81     $obj->$call(@_);
  82   }
  83 }
  84
  85 package maybe::start;
  86
  87 sub AUTOLOAD {
  88   my $invocant = shift;
  89   my ($method) = our $AUTOLOAD =~ /^maybe::start::(.+)$/;
  90   $method = "start::${method}" if ((caller(1)||'') eq 'start');
  91   $invocant->$method(@_);
  92 }
  93
  94 package then;
  95
  96 sub AUTOLOAD {
  97   my $invocant = shift;
  98   my ($method) = our $AUTOLOAD =~ /^then::(.+)$/;
  99   my @args = @_;
 100   # Need two copies since if we're called on an already complete future
 101   # $f will be freed immediately
 102   my $ret = my $f = CPS::Future->new;
 103   $invocant->on_fail(sub { $f->fail(@_); undef($f); });
 104   $invocant->on_done(sub {
 105     my ($obj) = @_;
 106     my $next = $obj->${\"start::${method}"}(@args);
 107     $next->on_done(sub { $f->done(@_); undef($f); });
 108     $next->on_fail(sub { $f->fail(@_); undef($f); });
 109   });
 110   return $ret;
 111 }
 112
 113 1;
 114
 115 =head1 NAME
 116
 117 Object::Remote::Future - Asynchronous calling for L<Object::Remote>
 118
 119 =head1 LAME
 120
 121 Shipping prioritised over writing this part up. Blame mst.
 122
 123 =cut