X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?p=scpubgit%2FObject-Remote.git;a=blobdiff_plain;f=lib%2FObject%2FRemote%2FConnection.pm;h=c5bc9f673af6e22e7af162adac9798de1e782f8b;hp=a5ad3021563fac8e1f74aa09a08c15122cab48df;hb=11dbd4a02cce95ee689233d2b1d2392470518732;hpb=e144d525c19bc4488952e4cea0ea0fe47a8db133 diff --git a/lib/Object/Remote/Connection.pm b/lib/Object/Remote/Connection.pm index a5ad302..c5bc9f6 100644 --- a/lib/Object/Remote/Connection.pm +++ b/lib/Object/Remote/Connection.pm @@ -35,20 +35,21 @@ has on_close => (is => 'rw', default => sub { CPS::Future->new }); has child_pid => (is => 'ro'); -has is_ready => (is => 'rwp', trigger => sub { - my ($self, $value) = @_; - $self->ready_future->done if $value; -}); - has ready_future => (is => 'lazy'); sub _build_ready_future { CPS::Future->new } has _receive_data_buffer => (is => 'ro', default => sub { my $x = ''; \$x }); -has local_objects_by_id => (is => 'ro', default => sub { {} }); +has local_objects_by_id => ( + is => 'ro', default => sub { {} }, + coerce => sub { +{ %{$_[0]} } }, # shallow clone on the way in +); -has remote_objects_by_id => (is => 'ro', default => sub { {} }); +has remote_objects_by_id => ( + is => 'ro', default => sub { {} }, + coerce => sub { +{ %{$_[0]} } }, # shallow clone on the way in +); has outstanding_futures => (is => 'ro', default => sub { {} }); @@ -88,6 +89,7 @@ BEGIN { eval { require Object::Remote::Connector::Local }; eval { require Object::Remote::Connector::LocalSudo }; eval { require Object::Remote::Connector::SSH }; + eval { require Object::Remote::Connector::UNIX }; } sub new_from_spec { @@ -99,13 +101,26 @@ sub new_from_spec { die "Couldn't figure out what to do with ${spec}"; } -sub new_remote { +sub remote_object { my ($self, @args) = @_; Object::Remote::Handle->new( connection => $self, @args )->proxy; } +sub connect { + my ($self, $to) = @_; + return await_future($self->send( + class_call => 'Object::Remote', 0, connect => $to + )); +} + +sub remote_sub { + my ($self, $sub) = @_; + my ($pkg, $name) = $sub =~ m/^(.*)::([^:]+)$/; + return await_future($self->send(class_call => $pkg, 0, can => $name)); +} + sub register_remote { my ($self, $remote) = @_; weaken($self->remote_objects_by_id->{$remote->id} = $remote); @@ -114,7 +129,6 @@ sub register_remote { sub await_ready { my ($self) = @_; - return if $self->is_ready; await_future($self->ready_future); } @@ -149,7 +163,7 @@ sub send_discard { sub _send { my ($self, $to_send) = @_; - $self->await_ready unless $self->is_ready; + $self->await_ready; print { $self->send_to_fh } $self->_serialize($to_send)."\n"; } @@ -203,8 +217,10 @@ sub _deobjectify { sub _receive_data_from { my ($self, $fh) = @_; my $rb = $self->_receive_data_buffer; - my $ready = $self->is_ready; - if (sysread($fh, $$rb, 1024, length($$rb)) > 0) { + my $ready = $self->ready_future->is_ready; + my $len = sysread($fh, $$rb, 1024, length($$rb)); + my $err = defined($len) ? undef : ": $!"; + if (defined($len) and $len > 0) { while ($$rb =~ s/^(.*)\n//) { if ($ready) { $self->_receive($1); @@ -212,7 +228,7 @@ sub _receive_data_from { my $line = $1; die "New remote container did not send Shere - got ${line}" unless $line eq "Shere"; - $self->_set_is_ready($ready = 1); + $self->ready_future->done; } } } else { @@ -222,7 +238,7 @@ sub _receive_data_from { on_read_ready => 1 ); my $outstanding = $self->outstanding_futures; - $_->fail("Connection lost") for values %$outstanding; + $_->fail("Connection lost${err}") for values %$outstanding; %$outstanding = (); $self->on_close->done(); }