my %shells = (
bash => { map +($_ => 1),
- qw(joe.example.com kitty.scsys.co.uk pryde.scsys.co.uk) },
+ qw(joe.example.com kitty.scsys.co.uk) },
csh => { map +($_ => 1),
qw(jim.example.com joe.example.com bob.example.com) },
);
my ($var, $array) = @_;
sub {
my ($self, $state) = @_;
- $state->bind_stream_then(
+ $state->then($self->next)->bind_stream(
$state->scope_var($var),
ArrayStream->from_array(@$array),
- $self->next
)
}
}
}
}
-my $op = FromCode->new(
- code => bind_array(S => \@servers),
- next => FromCode->new(
- code => test_values([ 'S' ], sub { $_[0] =~ /\.example\.com$/ })
- )
-);
+sub make_op {
+ my ($inner) = @_;
+ FromCode->new(
+ code => bind_array(S => \@servers),
+ next => FromCode->new(
+ code => test_values([ 'S' ], sub { $_[0] =~ /\.example\.com$/ }),
+ next => $inner,
+ )
+ );
+}
+
+my $op = make_op;
sub make_state {
my ($vars, $op) = @_;
my $stream = DX::ResultStream->new(for_state => make_state([ 'S' ], $op));
-is($stream->next->{'S'}, $_)
+is($stream->next->value_for('S'), $_)
for qw(jim.example.com joe.example.com bob.example.com);
is($stream->next, undef, 'No more');
)
);
+sub bound_values {
+ map {
+ my $v = $_;
+ +{
+ map +($_ => $v->value_for($_)), $v->var_names,
+ }
+ } @_
+}
+
my $cstream = DX::ResultStream->new(
for_state => make_state([ qw(S P) ], $complex_op)
);
is_deeply(
- [ $cstream->results ],
+ [ bound_values $cstream->results ],
[
{ P => 'csh', S => 'jim.example.com' },
{ P => 'csh', S => 'joe.example.com' },
'Complex stream'
);
+my $pop_stack = FromCode->new(
+ code => sub { $_[1]->pop_return_stack }
+);
+
+my $inner_op = make_op($pop_stack);
+
+my $call_op = FromCode->new(
+ code => sub {
+ my ($self, $state) = @_;
+ my $save_scope = $state->scope;
+ my %scope = (S => $save_scope->{S});
+ my $ret_op = FromCode->new(
+ code => sub { $_[1]->but(scope => $save_scope, next_op => $_[0]->next) },
+ next => $self->next,
+ );
+ $state->but(scope => \%scope)->push_return_then($ret_op, $inner_op);
+ },
+ next => FromCode->new(
+ code => bind_array(P => \@shells),
+ next => FromCode->new(
+ code => test_values([ qw(S P) ], sub { $shells{$_[1]}{$_[0]} }),
+ )
+ )
+);
+
+my $callstream = DX::ResultStream->new(
+ for_state => make_state([ qw(S P) ], $call_op)
+);
+
+is_deeply(
+ [ bound_values $callstream->results ],
+ [
+ { P => 'csh', S => 'jim.example.com' },
+ { P => 'csh', S => 'joe.example.com' },
+ { P => 'bash', S => 'joe.example.com' },
+ { P => 'csh', S => 'bob.example.com' },
+ ],
+ 'Call stream'
+);
+
+my $has_csh = FromCode->new(
+ code => test_values([ 'S' ], sub { $shells{csh}{$_[0]} }),
+ next => $pop_stack
+);
+my $has_bash = FromCode->new(
+ code => test_values([ 'S' ], sub { $shells{bash}{$_[0]} }),
+ next => $pop_stack
+);
+
+my $or_code = sub {
+ my ($self, $state) = @_;
+ my $var = DX::Var->new(id => 'OR')->with_stream(
+ my $stream = ArrayStream->from_array($has_csh, $has_bash)
+ );
+ my $inner_or = FromCode->new(
+ code => sub { $_[1]->then($var->bound_value) }
+ );
+ $state->push_return_then($self->next, $inner_or)
+ ->mark_choice($var);
+};
+
+my $top_or = FromCode->new(
+ code => bind_array(S => \@servers),
+ next => FromCode->new(code => $or_code),
+);
+
+my $orstream = DX::ResultStream->new(
+ for_state => make_state([ qw(S) ], $top_or)
+);
+
+is_deeply(
+ [ bound_values $orstream->results ],
+ [
+ {
+ S => "kitty.scsys.co.uk"
+ },
+ {
+ S => "jim.example.com"
+ },
+ {
+ S => "joe.example.com"
+ },
+ {
+ S => "joe.example.com"
+ },
+ {
+ S => "bob.example.com"
+ }
+ ],
+ 'Or stream'
+);
+
+my $top_or_2 = FromCode->new(
+ code => bind_array(S => \@servers),
+ next => FromCode->new(
+ code => $or_code,
+ next => FromCode->new(
+ code => test_values([ 'S' ], sub { $_[0] =~ /\.example\.com$/ }),
+ ),
+ ),
+);
+
+my $orstream_2 = DX::ResultStream->new(
+ for_state => make_state([ qw(S) ], $top_or_2)
+);
+
+is_deeply(
+ [ bound_values $orstream_2->results ],
+ [
+ {
+ S => "jim.example.com"
+ },
+ {
+ S => "joe.example.com"
+ },
+ {
+ S => "joe.example.com"
+ },
+ {
+ S => "bob.example.com"
+ }
+ ],
+ 'Or stream'
+);
+
done_testing;