lib/Object/Remote/Connection.pm

   1 package Object::Remote::Connection;
   2
   3 use CPS::Future;
   4 use Object::Remote;
   5 use IO::Handle;
   6 use Module::Runtime qw(use_module);
   7 use Scalar::Util qw(weaken blessed refaddr);
   8 use JSON::PP qw(encode_json);
   9 use Moo;
  10
  11 has send_to_fh => (
  12   is => 'ro', required => 1,
  13   trigger => sub { $_[1]->autoflush(1) },
  14 );
  15
  16 has receive_from_fh => (
  17   is => 'ro', required => 1,
  18   trigger => sub {
  19     my ($self, $fh) = @_;
  20     weaken($self);
  21     Object::Remote->current_loop
  22                   ->watch_io(
  23                       handle => $fh,
  24                       on_read_ready => sub { $self->_receive_data_from($fh) }
  25                     );
  26   },
  27 );
  28
  29 has _receive_data_buffer => (is => 'ro', default => sub { my $x = ''; \$x });
  30
  31 has local_objects_by_id => (is => 'ro', default => sub { {} });
  32
  33 has remote_objects_by_id => (is => 'ro', default => sub { {} });
  34
  35 has _json => (
  36   is => 'lazy',
  37   handles => {
  38     _deserialize => 'decode',
  39     _encode => 'encode',
  40   },
  41 );
  42
  43 sub _build__json {
  44   weaken(my $self = shift);
  45   my $remotes = $self->remote_objects_by_id;
  46   JSON::PP->new->filter_json_single_key_object(
  47     __remote_object__ => sub {
  48       my $id = shift;
  49       (
  50         $remotes->{$id}
  51         or Object::Remote->new(connection => $self, id => $id)
  52       )->proxy;
  53     }
  54   );
  55 }
  56
  57 sub register_remote {
  58   my ($self, $remote) = @_;
  59   weaken($self->remote_objects_by_id->{$remote->id} = $remote);
  60   return $remote;
  61 }
  62
  63 sub send_free {
  64   my ($self, $id) = @_;
  65   delete $self->remote_objects_by_id->{$id};
  66   $self->_send([ free => $id ]);
  67 }
  68
  69 sub send {
  70   my ($self, $type, @call) = @_;
  71
  72   unshift @call, $type => my $future = CPS::Future->new;
  73
  74   $self->_send(\@call);
  75
  76   return $future;
  77 }
  78
  79 sub _send {
  80   my ($self, $to_send) = @_;
  81
  82   print { $self->send_to_fh } $self->_serialize($to_send);
  83 }
  84
  85 sub _serialize {
  86   my ($self, $data) = @_;
  87   local our @New_Ids;
  88   eval { return $self->_encode($self->_deobjectify($data)) };
  89   my $err = $@; # won't get here if the eval doesn't die
  90   # don't keep refs to new things
  91   delete @{$self->local_objects_by_id}{@New_Ids};
  92   die "Error serializing: $err";
  93 }
  94
  95 sub _deobjectify {
  96   my ($self, $data) = @_;
  97   if (blessed($data)) {
  98     my $id = refaddr($data);
  99     $self->local_objects_by_id->{$id} ||= do {
 100       push our(@New_Ids), $id;
 101       $data;
 102     };
 103     return +{ __remote_object__ => $id };
 104   } elsif (my $ref = ref($data)) {
 105     if ($ref eq 'HASH') {
 106       return +{ map +($_ => $self->_deobjectify($data->{$_})), keys %$data };
 107     } elsif ($ref eq 'ARRAY') {
 108       return [ map $self->_deobjectify($_), @$data ];
 109     } else {
 110       die "Can't collapse reftype $ref";
 111     }
 112   }
 113   return $data; # plain scalar
 114 }
 115
 116 sub _receive_data_from {
 117   my ($self, $fh) = @_;
 118   my $rb = $self->_receive_data_buffer;
 119   if (sysread($self->read_fh, $$rb, 1024, length($$rb)) > 0) {
 120     while ($$rb =~ s/^(.*)\n//) {
 121       $self->_receive($1);
 122     }
 123   }
 124 }
 125
 126 sub _receive {
 127   my ($self, $data) = @_;
 128   my ($type, @rest) = eval { @{$self->_deserialize($data)} }
 129     or do { warn "Deserialize failed for ${data}: $@"; return };
 130   eval { $self->${\"receive_${type}"}(@rest); 1 }
 131     or do { warn "Receive failed for ${data}: $@"; return };
 132   return;
 133 }
 134
 135 sub receive_free {
 136   my ($self, $id) = @_;
 137   delete $self->local_objects_by_id->{$id};
 138   return;
 139 }
 140
 141 sub receive_call {
 142   my ($self, $future, $id, @rest) = @_;
 143   my $local = $self->local_objects_by_id->{$id}
 144     or do { $future->fail("No such object $id"); return };
 145   $self->_invoke($future, $local, @rest);
 146 }
 147
 148 sub receive_class_call {
 149   my ($self, $future, $class, @rest) = @_;
 150   eval { use_module($class) }
 151     or do { $future->fail("Error loading ${class}: $@"); return };
 152   $self->_invoke($future, $class, @rest);
 153 }
 154
 155 sub _invoke {
 156   my ($self, $future, $local, $method, @args) = @_;
 157   eval { $future->done($local->$method(@args)); 1 }
 158     or do { $future->fail($@); return; };
 159   return;
 160 }
 161
 162 sub DEMOLISH {
 163   my ($self, $gd) = @_;
 164   return if $gd;
 165   Object::Remote->current_loop
 166                 ->unwatch_io(
 167                     handle => $self->receive_from_fh,
 168                     on_read_ready => 1
 169                   );
 170 }
 171
 172 1;