7e1d4b193b505b124d3b3d4955743de608688f43
[scpubgit/Tak.git] / lib / Tak / ConnectorService.pm
1 package Tak::ConnectorService;
2
3 use IPC::Open2;
4 use IO::All;
5 use Tak::Router;
6 use Tak::Client;
7 use Tak::ConnectionService;
8 use Net::OpenSSH;
9 use Moo;
10
11 with 'Tak::Role::Service';
12
13 has connections => (is => 'ro', default => sub { Tak::Router->new });
14
15 has ssh => (is => 'ro', default => sub { {} });
16
17 sub handle_create {
18   my ($self, $on, %args) = @_;
19   my $log_level = $args{log_level}||'info';
20   my ($kid_in, $kid_out, $kid_pid) = $self->_open($on, $log_level);
21   $kid_in->print(io('maint/mk-fat |')->all, "__END__\n");
22   # Need to get a handshake to indicate STDIOSetup has finished
23   # messing around with file descriptors, otherwise we can severely
24   # confuse things by sending before the dup.
25   my $up = <$kid_out>;
26   die [ failure => "Garbled response from child: $up" ]
27     unless $up eq "Ssyshere\n";
28   my $connection = Tak::ConnectionService->new(
29     read_fh => $kid_out, write_fh => $kid_in,
30     listening_service => Tak::Router->new
31   );
32   my $client = Tak::Client->new(service => $connection);
33   # actually, we should register with a monotonic id and
34   # stash the pid elsewhere. but meh for now.
35   my $pid = $client->do(meta => 'pid');
36   my $name = ($on||'|').':'.$pid;
37   my $conn_router = Tak::Router->new;
38   $conn_router->register(local => $connection->receiver->service);
39   $conn_router->register(remote => $connection);
40   $self->connections->register($name, $conn_router);
41   return ($name);
42 }
43
44 sub _open {
45   my ($self, $on, @args) = @_;
46   unless ($on) {
47     my $kid_pid = IPC::Open2::open2(my $kid_out, my $kid_in, $^X, '-', '-', @args)
48       or die "Couldn't open2 child: $!";
49     return ($kid_in, $kid_out, $kid_pid);
50   }
51   my $ssh = $self->ssh->{$on} ||= Net::OpenSSH->new($on);
52   $ssh->error and
53     die "Couldn't establish ssh connection: ".$ssh->error;
54   return $ssh->open2('perl','-', $on, @args);
55 }
56
57 sub start_connection_request {
58   my ($self, $req, @payload) = @_;;
59   $self->connections->start_request($req, @payload);
60 }
61
62 sub receive_connection {
63   my ($self, @payload) = @_;
64   $self->connections->receive(@payload);
65 }
66
67 1;