lib/Tak/Router.pm

   1 package Tak::Router;
   2
   3 use Tak::Request;
   4 use Tak::ServiceManager;
   5 use Moo;
   6
   7 has channel => (is => 'ro', required => 1);
   8
   9 has local_request_handlers => (is => 'ro', default => sub { {} });
  10
  11 has requests_received => (is => 'ro', default => sub { {} });
  12
  13 has last_serial => (is => 'ro',default => sub { 'A0000' });
  14
  15 sub next_serial { ++($_[0]->{last_serial}) }
  16
  17 has requests_sent => (is => 'ro', default => sub { {} });
  18
  19 sub run { shift->run_until }
  20
  21 sub run_until {
  22   my ($self, $done) = @_;
  23   while (!$_[1] and my $message = $self->channel->receive) {
  24     $self->receive(@$message);
  25   }
  26 }
  27
  28 sub receive {
  29   my ($self, $type, @payload) = @_;
  30   unless ($type) {
  31     $self->channel->send(MISTAKE => message_format => "No message type");
  32     return;
  33   }
  34   unless (@payload) {
  35     $self->channel->send(MISTAKE => message_format => "Tag missing");
  36     return;
  37   }
  38   unless (@payload > 1) {
  39     $self->channel->send(MISTAKE => message_format => "No payload");
  40   }
  41   if ($type eq 'REQUEST') {
  42     $self->receive_request(@payload);
  43     return;
  44   }
  45   if ($type eq 'RESPONSE') {
  46     $self->receive_response(@payload);
  47     return;
  48   }
  49 }
  50
  51 sub receive_request {
  52   my ($self, $tag, $handler_name, @payload) = @_;
  53   if ($self->requests_received->{$tag}) {
  54     $self->channel->send(
  55       MISTAKE => request_tag => "Request for ${tag} in process"
  56     );
  57     return;
  58   }
  59   my $handler = $self->local_request_handlers->{$handler_name};
  60   unless ($handler) {
  61     $self->send_response(
  62       $tag => MISTAKE => handler_name => "No such handler ${handler_name}"
  63     );
  64     return;
  65   }
  66   my $request
  67     = $self->requests_received->{$tag}
  68     = Tak::Request->new(
  69         tag => $tag, respond_to => $self, respond_with => 'send_response',
  70       );
  71   $handler->start_request($request => @payload);
  72 }
  73
  74 sub send_response {
  75   my ($self, $tag, @result) = @_;
  76   delete $self->requests_received->{$tag};
  77   $self->channel->send(RESPONSE => $tag => @result);
  78 }
  79
  80 sub send_request {
  81   my ($self, $respond_to, $respond_with, @payload) = @_;
  82   my $tag = $self->next_serial;
  83   my $request
  84     = $self->requests_sent->{$tag}
  85     = Tak::Request->new(
  86         tag => $tag,
  87         respond_to => $respond_to,
  88         respond_with => $respond_with,
  89       );
  90   $self->channel->send(REQUEST => $tag => @payload);
  91   return $request;
  92 }
  93
  94 sub receive_response {
  95   my ($self, $tag, @result) = @_;
  96   my $request = delete $self->requests_sent->{$tag};
  97   $request->respond(@result);
  98 }
  99
 100 sub register {
 101   my ($self, $name, $service) = @_;
 102   $self->local_request_handlers->{$name} = Tak::ServiceManager->new(
 103     service => $service
 104   );
 105 }
 106
 107 1;