1 package Object::Remote::Connection;
6 use Module::Runtime qw(use_module);
7 use Scalar::Util qw(weaken blessed refaddr);
8 use JSON::PP qw(encode_json);
12 is => 'ro', required => 1,
13 trigger => sub { $_[1]->autoflush(1) },
16 has receive_from_fh => (
17 is => 'ro', required => 1,
21 Object::Remote->current_loop
24 on_read_ready => sub { $self->_receive_data_from($fh) }
29 has _receive_data_buffer => (is => 'ro', default => sub { my $x = ''; \$x });
31 has local_objects_by_id => (is => 'ro', default => sub { {} });
33 has remote_objects_by_id => (is => 'ro', default => sub { {} });
38 _deserialize => 'decode',
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 {
51 or Object::Remote->new(connection => $self, id => $id)
58 my ($self, $remote) = @_;
59 weaken($self->remote_objects_by_id->{$remote->id} = $remote);
65 delete $self->remote_objects_by_id->{$id};
66 $self->_send([ free => $id ]);
70 my ($self, $type, @call) = @_;
72 unshift @call, $type => my $future = CPS::Future->new;
80 my ($self, $to_send) = @_;
82 print { $self->send_to_fh } $self->_serialize($to_send);
86 my ($self, $data) = @_;
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";
96 my ($self, $data) = @_;
98 my $id = refaddr($data);
99 $self->local_objects_by_id->{$id} ||= do {
100 push our(@New_Ids), $id;
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 ];
110 die "Can't collapse reftype $ref";
113 return $data; # plain scalar
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//) {
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 };
136 my ($self, $id) = @_;
137 delete $self->local_objects_by_id->{$id};
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);
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);
156 my ($self, $future, $local, $method, @args) = @_;
157 eval { $future->done($local->$method(@args)); 1 }
158 or do { $future->fail($@); return; };
163 my ($self, $gd) = @_;
165 Object::Remote->current_loop
167 handle => $self->receive_from_fh,