File Coverage

blib/lib/Tak/ConnectionReceiver.pm

Criterion	Covered	Total	%
statement	12	35	34.2
branch	0	4	0.0
condition			n/a
subroutine	4	14	28.5
pod	0	7	0.0
total	16	60	26.6

package Tak::ConnectionReceiver;

use Tak::Request;
use Scalar::Util qw(weaken);
use Log::Contextual qw(:log);
use Moo;

with 'Tak::Role::Service';

has requests => (is => 'ro', default => sub { {} });

has channel => (is => 'ro', required => 1);

has service => (is => 'ro', required => 1);

has on_close => (is => 'ro', required => 1);

sub BUILD {
  weaken(my $self = shift);
  my $channel = $self->channel;
  Tak->loop->watch_io(
    handle => $channel->read_fh,
    on_read_ready => sub {
      $channel->read_messages(sub { $self->receive(@_) });
    }
  );
}

sub DEMOLISH {
  Tak->loop->unwatch_io(
    handle => $_[0]->channel->read_fh,
    on_read_ready => 1,
  );
}

sub receive_request {
  my ($self, $tag, $meta, @payload) = @_;
  my $channel = $self->channel;
  unless (ref($meta) eq 'HASH') {
    $channel->write_message(mistake => $tag => 'meta value not a hashref');
    return;
  }
  my $req = Tak::Request->new(
    ($meta->{progress}
        ? (on_progress => sub { $channel->write_message(progress => $tag => @_) })
        : ()),
    on_result => sub { $channel->write_message(result => $tag => $_[0]->flatten) }
  );
  $self->service->start_request($req => @payload);
}

sub receive_progress {
  my ($self, $tag, @payload) = @_;
  $self->requests->{$tag}->progress(@payload);
}

sub receive_result {
  my ($self, $tag, @payload) = @_;
  (delete $self->requests->{$tag})->result(@payload);
}

sub receive_message {
  my ($self, @payload) = @_;
  $self->service->receive(@payload);
}

sub receive_close {
  my ($self, @payload) = @_;
  $self->on_close->(@payload);
}

1;