object proxying system

[scpubgit/Tak.git] / lib / Tak / Router.pm

package Tak::Router;

use Tak::Request;
use Tak::ServiceManager;
use Tak::MetaService;
use Moo;

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

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

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

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

sub next_serial { ++($_[0]->{last_serial}) }

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

sub BUILD {
  my ($self) = @_;
  $self->register(meta => Tak::MetaService->new(router => $self));
}

sub run { shift->run_until }

sub run_until {
  my ($self, $done) = @_;
  while (!$_[1] and my $message = $self->channel->receive) {
    $self->receive(@$message);
  }
}

sub receive {
  my ($self, $type, @payload) = @_;
  unless ($type) {
    $self->channel->send(MISTAKE => message_format => "No message type");
    return;
  }
  unless (@payload) {
    $self->channel->send(MISTAKE => message_format => "Tag missing");
    return;
  }
  unless (@payload > 1) {
    $self->channel->send(MISTAKE => message_format => "No payload");
  }
  if ($type eq 'REQUEST') {
    $self->receive_request(@payload);
    return;
  }
  if ($type eq 'RESPONSE') {
    $self->receive_response(@payload);
    return;
  }
}

sub receive_request {
  my ($self, $tag, $handler_name, @payload) = @_;
  if ($self->requests_received->{$tag}) {
    $self->channel->send(
      MISTAKE => request_tag => "Request for ${tag} in process"
    );
    return;
  }
  my $handler = $self->local_request_handlers->{$handler_name};
  unless ($handler) {
    $self->send_response(
      $tag => MISTAKE => handler_name => "No such handler ${handler_name}"
    );
    return;
  }
  my $request
    = $self->requests_received->{$tag}
    = Tak::Request->new(
        tag => $tag, respond_to => $self, respond_with => 'send_response',
      );
  $handler->start_request($request => @payload);
}

sub send_response {
  my ($self, $tag, @result) = @_;
  delete $self->requests_received->{$tag};
  $self->channel->send(RESPONSE => $tag => @result);
}

sub send_request {
  my ($self, $respond_to, $respond_with, @payload) = @_;
  my $tag = $self->next_serial;
  my $request
    = $self->requests_sent->{$tag}
    = Tak::Request->new(
        tag => $tag,
        respond_to => $respond_to,
        respond_with => $respond_with,
      );
  $self->channel->send(REQUEST => $tag => @payload);
  return $request;
}

sub receive_response {
  my ($self, $tag, @result) = @_;
  my $request = delete $self->requests_sent->{$tag};
  $request->respond(@result);
}

sub register {
  my ($self, $name, $service) = @_;
  $self->local_request_handlers->{$name} = Tak::ServiceManager->new(
    service => $service
  );
}

1;