Skip to content

Commit

Permalink
Give Proc::Async support for merged output.
Browse files Browse the repository at this point in the history
Available through the `.Supply` method, which feels fairly natural.
In IO::Socket::Async, .Supply is how you get the data received over
the socket; `.Supply` on a process can be seen as "get me the output"
without caring which output.

Added as part of work towards re-working Proc in terms of Proc::Async,
which means giving Proc::Async the features that Proc has. However, as
it turns out :merge was busted and disabled for Proc, this counts as a
bonus, I guess.
  • Loading branch information
jnthn committed Jun 5, 2017
1 parent 6c6bf36 commit fd3ef6f
Showing 1 changed file with 30 additions and 0 deletions.
30 changes: 30 additions & 0 deletions src/core/Proc/Async.pm
Expand Up @@ -11,6 +11,13 @@ my class X::Proc::Async::TapBeforeSpawn does X::Proc::Async {
}
}

my class X::Proc::Async::SupplyOrStd does X::Proc::Async {
method message() {
"Using .Supply on a Proc::Async implies merging stdout and stderr; .stdout " ~
"and .stderr cannot therefore be used in combination with it"
}
}

my class X::Proc::Async::CharsOrBytes does X::Proc::Async {
has $.handle;
method message() {
Expand Down Expand Up @@ -54,6 +61,8 @@ my class Proc::Async {
has CharsOrBytes $!stdout_type;
has $!stderr_supply;
has CharsOrBytes $!stderr_type;
has $!merge_supply;
has CharsOrBytes $!merge_type;
has $!process_handle;
has $!exit_promise;
has @!promises;
Expand All @@ -75,28 +84,46 @@ my class Proc::Async {

proto method stdout(|) { * }
multi method stdout(Proc::Async:D: :$bin!) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
$bin
?? self!supply('stdout', $!stdout_supply, $!stdout_type, Bytes).Supply
!! self.stdout(|%_)
}
multi method stdout(Proc::Async:D: :$enc, :$translate-nl) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
self!wrap-decoder:
self!supply('stdout', $!stdout_supply, $!stdout_type, Chars).Supply,
$enc, :$translate-nl
}

proto method stderr(|) { * }
multi method stderr(Proc::Async:D: :$bin!) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
$bin
?? self!supply('stderr', $!stderr_supply, $!stderr_type, Bytes).Supply
!! self.stderr(|%_)
}
multi method stderr(Proc::Async:D: :$enc, :$translate-nl) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
self!wrap-decoder:
self!supply('stderr', $!stderr_supply, $!stderr_type, Chars).Supply,
$enc, :$translate-nl
}

proto method Supply(|) { * }
multi method Supply(Proc::Async:D: :$bin!) {
die X::Proc::Async::SupplyOrStd.new if $!stdout_supply || $!stderr_supply;
$bin
?? self!supply('merge', $!merge_supply, $!merge_type, Bytes).Supply
!! self.Supply(|%_)
}
multi method Supply(Proc::Async:D: :$enc, :$translate-nl) {
die X::Proc::Async::SupplyOrStd.new if $!stdout_supply || $!stderr_supply;
self!wrap-decoder:
self!supply('merge', $!merge_supply, $!merge_type, Chars).Supply,
$enc, :$translate-nl
}

method ready(--> Promise) {
$!ready_promise;
}
Expand Down Expand Up @@ -151,6 +178,9 @@ my class Proc::Async {
@!promises.push(
self!capture($callbacks,'stderr',$!stderr_supply)
) if $!stderr_supply;
@!promises.push(
self!capture($callbacks,'merge',$!merge_supply)
) if $!merge_supply;

nqp::bindkey($callbacks, 'buf_type', buf8.new);
nqp::bindkey($callbacks, 'write', True) if $.w;
Expand Down

0 comments on commit fd3ef6f

Please sign in to comment.