File Coverage

blib/lib/RPC/EPC/Service.pm

Criterion	Covered	Total	%
statement	22	24	91.6
branch			n/a
condition			n/a
subroutine	8	8	100.0
pod			n/a
total	30	32	93.7

package RPC::EPC::Service;

use warnings;
use strict;
use utf8;
use Carp;

use version; our $VERSION = qv('0.0.11');

use base 'Exporter';

our @EXPORT = qw(
   to_sexp
);

use Encode;
use AnyEvent;
use AnyEvent::Socket;
use AnyEvent::Handle;
use Data::SExpression;
use Data::SExpression::Symbol;
use B;
use Data::Dumper; # for debug


##################################################
# sexp encoding
# (This code is based on Mojolicious's JSON library.)

# Translate an argument object into S-expression text.
sub to_sexp {
  my $arg = shift;
  if (ref $arg eq 'HASH') {
    return _to_sexp_hash($arg);
  } elsif (ref $arg eq 'ARRAY') {
    return _to_sexp_list($arg);
  }
  my $flags = B::svref_2object(\$arg)->FLAGS;
  if ($flags & (B::SVp_IOK | B::SVp_NOK) && !($flags & B::SVp_POK)) {
    return _to_sexp_num($arg);
  } else {
    return _to_sexp_string($arg);
  }
}


# for elisp's prin1 escape
my %ESCAPE = (
  '"'     => '"',
  '\\'    => '\\',
);

my %REVERSE;
for my $key (keys %ESCAPE) { $REVERSE{$ESCAPE{$key}} = "\\$key" }

sub _to_sexp_string {
  my $string = shift;
  return "nil" unless $string;

  # Escape
  $string =~ s/([\\\"])/$REVERSE{$1}/gs;

  $string = Encode::encode_utf8($string) if Encode::is_utf8($string);

  # Stringify
  return "\"$string\"";
}

sub _to_sexp_num {
  return shift;
}

sub _to_sexp_list {
  my $list = shift;
  my @out = map {to_sexp $_} @$list;
  return "(" . join(" ", @out) . ")";
}

sub _to_sexp_hash {
  my $hash = shift;
  my $out = [];
  foreach my $key ( keys %$hash ) {
    push @$out, "(".to_sexp($key)." . ".to_sexp($hash->{$key}).")";
  }
  return "(" . join(" ", @$out) . ")";
}

##################################################
# nil to undef

our $NIL = Data::SExpression::Symbol->new("nil");

sub _nil_to_undef {
  my $arg = shift;
  if (ref $arg eq 'HASH') {
    return _nil_to_undef_hash($arg);
  } elsif (ref $arg eq 'ARRAY') {
    return _nil_to_undef_list($arg);
  }
  return if ($NIL eq $arg);
  return $arg;
}

sub _nil_to_undef_list {
  my $list = shift;
  for (my $i = 0; $i < @$list; $i++) {
    $list->[$i] = _nil_to_undef($list->[$i]);
  }
  return $list;
}

sub _nil_to_undef_hash {
  my $hash = shift;
  foreach my $key ( keys %$hash ) {
    $hash->{$key} = _nil_to_undef($hash->{$key});
  }
  return $hash;
}

sub _correct_method_spec {
  my $a = shift;
  if (ref($a) eq "CODE") {
    return {'sub' => $a, 'arg_spec' => undef, 'docstring' => undef};
  } elsif (ref($a) eq "ARRAY") {
    my ($sub,$arg,$doc) = @$a;
    return {'sub' => $sub, 'arg_spec' => $arg, 'docstring' => $doc};
  } elsif (ref($a) eq "HASH") {
    return $a;
  } else {
    return $a;
  }
}

sub _correct_method_specs {
  my $methods = shift;
  my $ret = {};
  foreach my $key ( keys %$methods ) {
    $ret->{$key} = _correct_method_spec( $methods->{$key} );
  }
  return $ret;
}


##################################################
# Protocol Stacks

sub new {
  my ($class, $port, $methods) = @_;
  my $cv = AnyEvent->condvar;
  return bless { 'port' => $port,
                 'count' => 0,
                 'methods' => _correct_method_specs($methods),
                 'sessions' => {},
                 'wait' => $cv }, $class;
}

sub _register_event_loop {
  my ($self,$fh) = @_;

  my $ds = Data::SExpression->new({use_symbol_class=>1,fold_alists=>1});

  my $hdl; $hdl = new AnyEvent::Handle
    (fh => $fh,
     on_error => sub {
       my ($hdl, $fatal, $msg) = @_;
       AE::log warn => "got error $msg\n";
       $hdl->destroy;
       $self->{wait}->send;
     });
  $self->{handle} = $hdl;

  my $reading_buffer = "";
  my @reader; @reader =
    (line => sub {
       my ($hdl, $incoming) = @_;
       # print STDERR "INCOMING:$incoming";
       $reading_buffer .= $incoming . "\n";
       my $len = substr($reading_buffer, 0, 6);
       if (!$len =~ /[0-9a-f]{6}/i) {
         # print STDERR "Wrong length code: $reading_buffer\nAbort\n";
         AE::log warn => "Wrong length code: $reading_buffer\n";
         $hdl->destroy;
         $self->{wait}->send;
         return;
       }
       $len = hex($len);
       my $body = substr($reading_buffer, 6);
       if ($len > length $body) {
         # try next lines
         # print STDERR "lencode:$len / current:" . (length $body) . "\n";
         $hdl->push_read(@reader);
         return;
       }
       # print STDERR "Here len:$len / current:" . (length $body) . "\n";
       if ($len < length $body) {
         $reading_buffer = substr($body, $len+1);
         $body = substr($body, 0, $len);
       } else {
         $reading_buffer = "";
       }

       my ($text, $sexp);
       eval {
         ($sexp, $text) = $ds->read(Encode::decode_utf8($body));
       };
       # print STDERR "SEXP:".Dumper $sexp;
       if ($sexp->[0]) {
         if ($sexp->[0]->name eq "quit") {
           $hdl->destroy;
           $self->{wait}->send;
           return;
         }

         my $handler = $self->{handlers}->{shift(@$sexp)->name};
         if ($handler) {
           $handler->($sexp);
         }
       } else {
         # print STDERR 'NULL:'.Dumper $sexp;
       }
       $hdl->push_read(@reader);
     });

  $hdl->push_read(@reader);
}

sub _handle_connection {
  my ($self,$fh,$host,$port) = @_;

  my $handlers = {
      'call' => sub { $self->_call(@_); },
      'return' => sub { $self->_return(@_); },
      'return-error' => sub { $self->_return_error(@_); },
      'epc-error' => sub { $self->_epc_error(@_); },
      'methods' => sub { $self->_query_methods(@_); },
    };
  $self->{handlers} = $handlers;
  $self->_register_event_loop($fh);
}

sub _uid {
  my $self = shift;
  return $self->{count}++;
}

sub _send_message {
  my ($self, $message) = @_;
  my $hdl = $self->{handle};
  my $len = length($message) + 1;
  # print STDERR ">>> [$message]\n";
  $hdl->push_write(sprintf("%06x%s\n", $len, $message));
}

sub _call {
  my ($self, $sexp) = @_;
  # print STDERR "CALL:" . Dumper $sexp;
  my $id = shift(@$sexp);
  my $name = shift(@$sexp);
  my $task = $self->{methods}->{$name}->{sub};
  if ($task) {
    my $args = _nil_to_undef($sexp->[0]);
    eval {
      my $ret = $task->($args);
      if ((ref $ret) eq "AnyEvent::CondVar") {
        $ret = $ret->recv;
      }
      $self->_send_message("(return $id ".to_sexp($ret).")");
    };
    if ($@) {
      $self->_send_message("(return-error $id ".to_sexp($@).")");
    }
  } else {
    $self->_send_message("(epc-error $id \"Not found the method: $name\")");
  }
}

sub _return {
  my ($self, $sexp) = @_;
  # print STDERR "RET:" . Dumper $sexp;
  my $id = shift(@$sexp);
  my $result = _nil_to_undef($sexp);
  my $cv = $self->{sessions}->{$id};
  if ($cv) {
    delete $self->{sessions}->{$id};
    $cv->send($result->[0]);
  } else {
    print STDERR "Not found ID : $id\n";
  }
}

sub _return_error {
  my ($self, $sexp) = @_;
  # print STDERR "ERR-RET:" . Dumper $sexp;
  my $id = shift(@$sexp);
  my $result = $sexp->[0];
  my $cv = $self->{sessions}->{$id};
  if ($cv) {
    delete $self->{sessions}->{$id};
    $cv->croak(['ERROR',$result]);
  } else {
    print STDERR "Not found ID : $id\n";
  }
}

sub _epc_error {
  my ($self, $sexp) = @_;
  # print STDERR "EPCERR-RET:" . Dumper $sexp;
  my $id = shift(@$sexp);
  my $result = $sexp->[0];
  my $cv = $self->{sessions}->{$id};
  if ($cv) {
    delete $self->{sessions}->{$id};
    $cv->croak(['EPC_ERROR',$result]);
  } else {
    print STDERR "Not found ID : $id\n";
  }
}

sub _query_methods {
  my ($self, $sexp) = @_;
  # print STDERR "METHODS:" . Dumper $sexp;
  my $id = shift(@$sexp);
  eval {
    my @ret = ();
    my $hash = $self->{methods};
    # print STDERR "HASH:" . Dumper $hash;
    foreach my $key ( keys %$hash ) {
      my $method = $hash->{$key};
      push @ret, [$key, $method->{'arg_spec'}, $method->{'docstring'}];
    }

    $self->_send_message("(return $id ".to_sexp(\@ret).")");
  };
  if ($@) {
    $self->_send_message("(return-error $id ".to_sexp($@).")");
  }
}

sub call_method {
  my $self = shift;
  my $name = shift;
  my $args = shift;
  my $cv = AnyEvent->condvar;
  my $id = $self->_uid;
  $self->{sessions}->{$id} = $cv;
  $self->_send_message("(call $id $name ".(to_sexp $args).")");
  return $cv;
}

sub define_method {
  my $self = shift;
  my $name = shift;
  $self->{methods}->{$name} = _correct_method_spec(\@_);
}

sub query_methods {
  my $self = shift;
  my $cv = AnyEvent->condvar;
  my $id = $self->_uid;
  $self->{sessions}->{$id} = $cv;
  $self->_send_message("(methods $id)");
  return $cv;
}

sub start {
  my $self = shift;
  my $server = tcp_server undef, $self->{port}, sub {
    $self->_handle_connection(@_);
  }, sub {
    my ($fh, $thishost, $thisport) = @_;
	binmode( STDOUT, ":unix" ); # immediate flush
    print "$thisport\n"; # epc protocol
  };
  $self->{server} = $server;
  $self->{wait}->recv;
}

sub client_start {
  my $self = shift;
  my $host = "127.0.0.1";
  my $cv = AnyEvent->condvar;
  my $client = tcp_connect $host, $self->{port}, sub {
    my ($fh) = @_;
    if ($fh) {
      $self->_handle_connection($fh, $host, $self->{port});
      $cv->send;
    } else {
      $cv->croak(['Could not connect server.']);
    }
  };
  $self->{client} = $client;
  $cv->recv;
}

sub stop {
  my $self = shift;
  $self->{handle}->push_shutdown;
}



1; # Magic true value required at end of module
__END__