Commit | Line | Data |
12fb4a80 |
1 | package Object::Remote::ReadChannel; |
2 | |
3 | use CPS::Future; |
bd20b1bf |
4 | use Scalar::Util qw(weaken openhandle); |
21988035 |
5 | use Object::Remote::Logging qw(:log :dlog get_router ); |
12fb4a80 |
6 | use Moo; |
7 | |
21988035 |
8 | BEGIN { get_router()->exclude_forwarding } |
9 | |
12fb4a80 |
10 | has fh => ( |
11 | is => 'ro', required => 1, |
12 | trigger => sub { |
13 | my ($self, $fh) = @_; |
14 | weaken($self); |
0511910e |
15 | log_trace { "Watching filehandle via trigger on 'fh' attribute in Object::Remote::ReadChannel" }; |
12fb4a80 |
16 | Object::Remote->current_loop |
17 | ->watch_io( |
18 | handle => $fh, |
19 | on_read_ready => sub { $self->_receive_data_from($fh) } |
20 | ); |
21 | }, |
22 | ); |
23 | |
24 | has on_close_call => ( |
25 | is => 'rw', default => sub { sub {} }, |
26 | ); |
27 | |
28 | has on_line_call => (is => 'rw'); |
29 | |
30 | has _receive_data_buffer => (is => 'ro', default => sub { my $x = ''; \$x }); |
31 | |
32 | sub _receive_data_from { |
33 | my ($self, $fh) = @_; |
07105aca |
34 | Dlog_trace { "Preparing to read data from $_" } $fh; |
12fb4a80 |
35 | my $rb = $self->_receive_data_buffer; |
07105aca |
36 | my $len = sysread($fh, $$rb, 32768, length($$rb)); |
69aaad21 |
37 | my $err = defined($len) ? 'eof' : ": $!"; |
12fb4a80 |
38 | if (defined($len) and $len > 0) { |
0511910e |
39 | log_trace { "Read $len bytes of data" }; |
12fb4a80 |
40 | while (my $cb = $self->on_line_call and $$rb =~ s/^(.*)\n//) { |
41 | $cb->(my $line = $1); |
42 | } |
b1f39f94 |
43 | } else { |
0511910e |
44 | log_trace { "Got EOF or error, this read channel is done" }; |
12fb4a80 |
45 | Object::Remote->current_loop |
46 | ->unwatch_io( |
47 | handle => $self->fh, |
48 | on_read_ready => 1 |
49 | ); |
8ed52376 |
50 | log_trace { "Invoking on_close_call() for dead read channel" }; |
12fb4a80 |
51 | $self->on_close_call->($err); |
52 | } |
53 | } |
54 | |
55 | sub DEMOLISH { |
56 | my ($self, $gd) = @_; |
57 | return if $gd; |
0511910e |
58 | log_trace { "read channel is being demolished" }; |
bd20b1bf |
59 | |
12fb4a80 |
60 | Object::Remote->current_loop |
61 | ->unwatch_io( |
62 | handle => $self->fh, |
63 | on_read_ready => 1 |
64 | ); |
bd20b1bf |
65 | |
66 | |
12fb4a80 |
67 | } |
68 | |
69 | 1; |