1 package Tak::Daemon::ListenerService;
3 use Scalar::Util qw(weaken);
6 with 'Tak::Role::Service';
8 has socket_location => (is => 'ro', required => 1);
9 has router => (is => 'ro', required => 1);
11 has state => (is => 'rw', default => sub { 'down' }, init_arg => undef);
13 has _start_in_progress => (is => 'lazy', clearer => '_clear_start_in_progress');
15 has listener => (is => 'rw', clearer => 'clear_listener');
17 has connections => (is => 'ro', default => sub { {} });
19 sub start_start_request {
20 my ($self, $req) = @_;
21 $req->result('already_started') if $self->state eq 'running';
22 push(@{$self->_start_in_progress->{requests}}, $req);
23 $self->_start_in_progress->{start}();
26 sub _build__start_in_progress {
29 my %start = (requests => (my $requests = []));
31 $self->state('starting');
37 path => $self->socket_location,
40 $self->listener($_[0]);
41 $_->success('started') for @$requests;
42 $self->_clear_start_in_progress;
43 $self->state('running');
45 on_resolve_error => sub { # no-op until we add non-unix
46 $_->failure(resolve => @_) for @$requests;
47 $self->_clear_start_in_progress;
48 $self->state('stopped');
50 on_listen_error => sub {
51 $_->failure(listen => @_) for @$requests;
52 $self->_clear_start_in_progress;
53 $self->state('stopped');
56 $self->setup_connection($_[0]);
58 on_accept_error => sub {
62 $start{start} = sub {}; # delete yourself
69 return 'already_stopped' if $self->state eq 'down';
70 # there's probably something more intelligent to do here, but meh
71 die failure => 'starting' if $self->state eq 'starting';
72 Tak->loop->remove($self->clear_listener);
73 unlink($self->socket_location);
81 if (my $l = $self->listener) {
82 $l->get_loop->remove($l);
83 unlink($self->socket_location);
87 sub setup_connection {
88 my ($self, $socket) = @_;
89 my $conn_set = $self->connections;
91 my $connection = Tak::ConnectionService->new(
92 read_fh => $socket, write_fh => $socket,
93 listening_service => $self->router->clone_or_self,
94 on_close => sub { delete $conn_set->{$conn_str} }
96 $conn_str = "$connection";
97 $connection->receiver->service->service->register_weak(remote => $connection);
98 $conn_set->{$conn_str} = $connection;