Commit | Line | Data |
12fb4a80 |
1 | package Object::Remote::ReadChannel; |
2 | |
3 | use CPS::Future; |
4 | use Scalar::Util qw(weaken); |
5 | use Moo; |
6 | |
7 | has fh => ( |
8 | is => 'ro', required => 1, |
9 | trigger => sub { |
10 | my ($self, $fh) = @_; |
11 | weaken($self); |
12 | Object::Remote->current_loop |
13 | ->watch_io( |
14 | handle => $fh, |
15 | on_read_ready => sub { $self->_receive_data_from($fh) } |
16 | ); |
17 | }, |
18 | ); |
19 | |
20 | has on_close_call => ( |
21 | is => 'rw', default => sub { sub {} }, |
22 | ); |
23 | |
24 | has on_line_call => (is => 'rw'); |
25 | |
26 | has _receive_data_buffer => (is => 'ro', default => sub { my $x = ''; \$x }); |
27 | |
28 | sub _receive_data_from { |
29 | my ($self, $fh) = @_; |
30 | my $rb = $self->_receive_data_buffer; |
31 | my $len = sysread($fh, $$rb, 1024, length($$rb)); |
32 | my $err = defined($len) ? '' : ": $!"; |
33 | if (defined($len) and $len > 0) { |
34 | while (my $cb = $self->on_line_call and $$rb =~ s/^(.*)\n//) { |
35 | $cb->(my $line = $1); |
36 | } |
37 | } else { |
38 | Object::Remote->current_loop |
39 | ->unwatch_io( |
40 | handle => $self->fh, |
41 | on_read_ready => 1 |
42 | ); |
43 | $self->on_close_call->($err); |
44 | } |
45 | } |
46 | |
47 | sub DEMOLISH { |
48 | my ($self, $gd) = @_; |
49 | return if $gd; |
50 | Object::Remote->current_loop |
51 | ->unwatch_io( |
52 | handle => $self->fh, |
53 | on_read_ready => 1 |
54 | ); |
55 | } |
56 | |
57 | 1; |