| 1 |
=head1 NAME |
| 2 |
|
| 3 |
AnyEvent::MCP::Helper - asynchronous JSON exchange with an external helper |
| 4 |
|
| 5 |
=head1 SYNOPSIS |
| 6 |
|
| 7 |
use AnyEvent::MCP::Helper; |
| 8 |
my $job = AnyEvent::MCP::Helper->new ( |
| 9 |
helper => ["my-helper", "--option"], timeout => 300, |
| 10 |
); |
| 11 |
$job->ask ($input, sub { |
| 12 |
my ($success, $result) = @_; |
| 13 |
... |
| 14 |
}); |
| 15 |
|
| 16 |
=head1 DESCRIPTION |
| 17 |
|
| 18 |
Runs an executable (or an argument array, never a shell command), writes one |
| 19 |
UTF-8 JSON value to its stdin and closes that input. A successful helper |
| 20 |
writes one JSON value to stdout and exits zero. Diagnostics use stderr. |
| 21 |
|
| 22 |
C<ask ($input, $done)> calls C<$done-E<gt>($success, $result)> once. On failure, |
| 23 |
C<$result> is an error string. One exchange may be active per object. |
| 24 |
C<timeout> defaults to 300 seconds and C<max_size> to 1 MiB of output. |
| 25 |
C<cancel ($reason)> terminates the helper and completes with failure. |
| 26 |
The independent child watcher reaps the helper after cancellation too. |
| 27 |
|
| 28 |
This module handles processes and bytes, not the meaning of the JSON. |
| 29 |
It is the transport for local form and permission helpers: an MCP server |
| 30 |
or multibroker can ask a human without blocking its AnyEvent loop. |
| 31 |
Ordinary server authors should normally use |
| 32 |
L<AnyEvent::MCP::Server::Permissions> or the form facilities in |
| 33 |
L<AnyEvent::MCP::Permissions::Ask>, not launch helpers directly. |
| 34 |
L<AnyEvent::MCP::Permissions::Ask> provides the permission interface; |
| 35 |
C<anyevent-mcp-form-elicitation> implements the standard form request/response contract. |
| 36 |
|
| 37 |
=cut |
| 38 |
|
| 39 |
package AnyEvent::MCP::Helper; |
| 40 |
|
| 41 |
use common::sense; |
| 42 |
|
| 43 |
use JSON::XS (); |
| 44 |
use Scalar::Util qw(weaken); |
| 45 |
|
| 46 |
use AnyEvent (); |
| 47 |
use AnyEvent::Util (); |
| 48 |
use AnyEvent::Handle (); |
| 49 |
use AnyEvent::MCP (); |
| 50 |
|
| 51 |
our @ISA = qw(AnyEvent::MCP::Base); |
| 52 |
|
| 53 |
sub ask { |
| 54 |
my ($self, $prepared, $done) = @_; |
| 55 |
|
| 56 |
die "Helper is already active\n" if $self->{on_done}; |
| 57 |
$self->{finished} = 0; |
| 58 |
$self->{on_done} = $done; |
| 59 |
delete @$self{qw(status eof)}; |
| 60 |
my $helper = $self->{helper} or die "No helper configured\n"; |
| 61 |
my $input = JSON::XS::encode_json $prepared; |
| 62 |
my $output = ""; |
| 63 |
my $ok = eval { |
| 64 |
my ($parent, $child) = AnyEvent::Util::portable_socketpair; |
| 65 |
die "Cannot create helper socket: $!\n" unless defined $parent && defined $child; |
| 66 |
$self->{cv} = AnyEvent::Util::run_cmd ( |
| 67 |
ref $helper eq "ARRAY" ? $helper : [$helper], |
| 68 |
"<" => $child, ">" => $child, '$$' => \$self->{pid}, close_all => 1, |
| 69 |
); |
| 70 |
close $child; |
| 71 |
weaken $self; |
| 72 |
$self->{handle} = AnyEvent::Handle->new ( |
| 73 |
fh => $parent, |
| 74 |
on_error => sub { $self->cancel ("Helper socket: $_[2]") if $self }, |
| 75 |
on_read => sub { |
| 76 |
return unless $self && !$self->{finished}; |
| 77 |
$output .= $_[0]{rbuf}; |
| 78 |
$_[0]{rbuf} = ""; |
| 79 |
$self->cancel ("Helper output exceeds size limit") |
| 80 |
if length $output > ($self->{max_size} || 1024 * 1024); |
| 81 |
}, |
| 82 |
on_eof => sub { |
| 83 |
return unless $self && !$self->{finished}; |
| 84 |
$self->{eof} = 1; |
| 85 |
$self->_complete ($output); |
| 86 |
}, |
| 87 |
); |
| 88 |
$self->{handle}->push_write ($input); |
| 89 |
$self->{handle}->push_shutdown; |
| 90 |
$self->{cv}->cb (sub { |
| 91 |
my $status = $_[0]->recv; |
| 92 |
return unless $self; |
| 93 |
delete $self->{pid}; # The child is reaped even if descendants keep stdout open. |
| 94 |
return if $self->{finished}; |
| 95 |
$self->{status} = $status; |
| 96 |
$self->_complete ($output); |
| 97 |
}); |
| 98 |
$self->{timer} = AnyEvent->timer (after => $self->{timeout} || 300, cb => sub { |
| 99 |
$self->cancel ("Helper timed out") if $self; |
| 100 |
}); |
| 101 |
1 |
| 102 |
}; |
| 103 |
$self->_finish (0, "$@") unless $ok; |
| 104 |
} |
| 105 |
|
| 106 |
sub _complete { |
| 107 |
my ($self, $output) = @_; |
| 108 |
|
| 109 |
return unless defined $self->{status} && $self->{eof}; |
| 110 |
return $self->_finish (0, "Helper exited with status $self->{status}") if $self->{status}; |
| 111 |
my $decision = eval { JSON::XS::decode_json $output }; |
| 112 |
return $self->_finish (0, "Invalid helper JSON: $@") if $@; |
| 113 |
$self->_finish (1, $decision); |
| 114 |
} |
| 115 |
|
| 116 |
sub _finish { |
| 117 |
my ($self, $success, $result) = @_; |
| 118 |
|
| 119 |
return if $self->{finished}; |
| 120 |
$self->{finished} = 1; |
| 121 |
delete @$self{qw(cv timer)}; |
| 122 |
(delete $self->{handle})->destroy if $self->{handle}; |
| 123 |
my $done = delete $self->{on_done}; |
| 124 |
$done->($success, $result) if $done; |
| 125 |
} |
| 126 |
|
| 127 |
sub cancel { |
| 128 |
my ($self, $reason) = @_; |
| 129 |
|
| 130 |
return if $self->{finished}; |
| 131 |
kill TERM => delete $self->{pid} if $self->{pid}; |
| 132 |
$self->_finish (0, defined $reason ? $reason : "Helper cancelled"); |
| 133 |
} |
| 134 |
|
| 135 |
sub DESTROY { |
| 136 |
$_[0]->cancel |
| 137 |
} |
| 138 |
|
| 139 |
1 |
