1 package Tak::ConnectionReceiver;
4 use Scalar::Util qw(weaken);
7 with 'Tak::Role::Service';
9 has requests => (is => 'ro', default => sub { {} });
11 has channel => (is => 'ro', required => 1);
13 has service => (is => 'ro', required => 1);
15 has on_close => (is => 'ro', required => 1);
18 weaken(my $self = shift);
19 my $channel = $self->channel;
21 handle => $channel->read_fh,
22 on_read_ready => sub {
23 if (my $message = $channel->read_message) {
24 $self->receive(@$message);
31 Tak->loop->unwatch_io(
32 handle => $_[0]->channel->read_fh,
38 my ($self, $tag, $meta, @payload) = @_;
39 my $channel = $self->channel;
40 my $req = Tak::Request->new(
42 ? (on_progress => sub { $channel->write_message(progress => $tag => @_) })
44 on_result => sub { $channel->write_message(result => $tag => $_[0]->flatten) }
46 $self->service->start_request($req => @payload);
49 sub receive_progress {
50 my ($self, $tag, @payload) = @_;
51 $self->requests->{$tag}->progress(@payload);
55 my ($self, $tag, @payload) = @_;
56 (delete $self->requests->{$tag})->result(@payload);
60 my ($self, @payload) = @_;
61 $self->service->receive(@payload);
65 my ($self, @payload) = @_;
66 $self->on_close->(@payload);