lib/DX/ResultStream.pm

   1 package DX::ResultStream;
   2
   3 use DX::Result;
   4 use Moo;
   5
   6 has for_state => (is => 'ro', required => 1);
   7
   8 has _current_state => (is => 'rw');
   9
  10 has is_exhausted => (is => 'rwp');
  11
  12 has observation_policy => (is => 'ro', default => sub { sub { 0 } });
  13
  14 sub next {
  15   my ($self) = @_;
  16   return if $self->is_exhausted;
  17   my $state = do {
  18     if (my $cur = $self->_current_state) {
  19       $cur->push_backtrack;
  20     } else {
  21       $self->for_state
  22     }
  23   };
  24   STATE: while ($state = $self->_current_state($state->run)) {
  25     last if $state->isa('DX::State');
  26     if ($state->isa('DX::ObservationRequired')) {
  27       if ($self->observation_policy->($state->observer)) {
  28         my @ob = $state->observer->run;
  29         $state = $state->resume;
  30         while (my ($type, $value) = splice(@ob, 0, 2)) {
  31           $state->facts->{$type}->set_value($value);
  32         }
  33       } else {
  34         die "Observation refused";
  35       }
  36     } else {
  37       die "WTF: ".$state;
  38     }
  39   }
  40   unless ($state) {
  41     $self->_set_is_exhausted(1);
  42     return;
  43   }
  44   return DX::Result->new(state => $state->copy_vars);
  45 }
  46
  47 sub results {
  48   my ($self) = @_;
  49   my @all;
  50   while (my $next = $self->next) {
  51     push @all, $next;
  52   }
  53   return @all;
  54 }
  55
  56 1;