ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.42 by root, Tue May 27 06:23:15 2008 UTC vs.
Revision 1.45 by root, Thu May 29 00:20:39 2008 UTC

… …
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw(EAGAIN EINTR);
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
… …
91 91
92The object will not be in a usable state when this callback has been 92The object will not be in a usable state when this callback has been
93called. 93called.
94 94
95On callback entrance, the value of C<$!> contains the operating system 95On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>). 96error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 97
98The callback should throw an exception. If it returns, then 98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you. 99AnyEvent::Handle will C<croak> for you.
100 100
101While not mandatory, it is I<highly> recommended to set this callback, as 101While not mandatory, it is I<highly> recommended to set this callback, as
… …
119 119
120This sets the callback that is called when the write buffer becomes empty 120This sets the callback that is called when the write buffer becomes empty
121(or when the callback is set and the buffer is empty already). 121(or when the callback is set and the buffer is empty already).
122 122
123To append to the write buffer, use the C<< ->push_write >> method. 123To append to the write buffer, use the C<< ->push_write >> method.
124
125=item timeout => $fractional_seconds
126
127If non-zero, then this enables an "inactivity" timeout: whenever this many
128seconds pass without a successful read or write on the underlying file
129handle, the C<on_timeout> callback will be invoked (and if that one is
130missing, an C<ETIMEDOUT> error will be raised).
131
132Note that timeout processing is also active when you currently do not have
133any outstanding read or write requests: If you plan to keep the connection
134idle then you should disable the timout temporarily or ignore the timeout
135in the C<on_timeout> callback.
136
137Zero (the default) disables this timeout.
138
139=item on_timeout => $cb->($handle)
140
141Called whenever the inactivity timeout passes. If you return from this
142callback, then the timeout will be reset as if some activity had happened,
143so this condition is not fatal in any way.
124 144
125=item rbuf_max => <bytes> 145=item rbuf_max => <bytes>
126 146
127If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 147If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
128when the read buffer ever (strictly) exceeds this size. This is useful to 148when the read buffer ever (strictly) exceeds this size. This is useful to
… …
202 if ($self->{tls}) { 222 if ($self->{tls}) {
203 require Net::SSLeay; 223 require Net::SSLeay;
204 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
205 } 225 }
206 226
207 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
208 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 228# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
229# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
209 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
210 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 231
232 $self->{_activity} = AnyEvent->now;
233 $self->_timeout;
211 234
212 $self->start_read; 235 $self->start_read;
213 236
214 $self 237 $self
215} 238}
… …
262 285
263sub on_eof { 286sub on_eof {
264 $_[0]{on_eof} = $_[1]; 287 $_[0]{on_eof} = $_[1];
265} 288}
266 289
290=item $handle->on_timeout ($cb)
291
292Replace the current C<on_timeout> callback, or disables the callback
293(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
294argument.
295
296=cut
297
298sub on_timeout {
299 $_[0]{on_timeout} = $_[1];
300}
301
302#############################################################################
303
304=item $handle->timeout ($seconds)
305
306Configures (or disables) the inactivity timeout.
307
308=cut
309
310sub timeout {
311 my ($self, $timeout) = @_;
312
313 $self->{timeout} = $timeout;
314 $self->_timeout;
315}
316
317# reset the timeout watcher, as neccessary
318# also check for time-outs
319sub _timeout {
320 my ($self) = @_;
321
322 if ($self->{timeout}) {
323 my $NOW = AnyEvent->now;
324
325 # when would the timeout trigger?
326 my $after = $self->{_activity} + $self->{timeout} - $NOW;
327
328 # now or in the past already?
329 if ($after <= 0) {
330 $self->{_activity} = $NOW;
331
332 if ($self->{on_timeout}) {
333 $self->{on_timeout}->($self);
334 } else {
335 $! = Errno::ETIMEDOUT;
336 $self->error;
337 }
338
339 # callbakx could have changed timeout value, optimise
340 return unless $self->{timeout};
341
342 # calculate new after
343 $after = $self->{timeout};
344 }
345
346 Scalar::Util::weaken $self;
347
348 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
349 delete $self->{_tw};
350 $self->_timeout;
351 });
352 } else {
353 delete $self->{_tw};
354 }
355}
356
267############################################################################# 357#############################################################################
268 358
269=back 359=back
270 360
271=head2 WRITE QUEUE 361=head2 WRITE QUEUE
… …
315 my $cb = sub { 405 my $cb = sub {
316 my $len = syswrite $self->{fh}, $self->{wbuf}; 406 my $len = syswrite $self->{fh}, $self->{wbuf};
317 407
318 if ($len >= 0) { 408 if ($len >= 0) {
319 substr $self->{wbuf}, 0, $len, ""; 409 substr $self->{wbuf}, 0, $len, "";
410
411 $self->{_activity} = AnyEvent->now;
320 412
321 $self->{on_drain}($self) 413 $self->{on_drain}($self)
322 if $self->{low_water_mark} >= length $self->{wbuf} 414 if $self->{low_water_mark} >= length $self->{wbuf}
323 && $self->{on_drain}; 415 && $self->{on_drain};
324 416
… …
960 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1052 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
961 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1053 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
962 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1054 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
963 1055
964 if ($len > 0) { 1056 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now;
1058
965 $self->{filter_r} 1059 $self->{filter_r}
966 ? $self->{filter_r}->($self, $rbuf) 1060 ? $self->{filter_r}->($self, $rbuf)
967 : $self->_drain_rbuf; 1061 : $self->_drain_rbuf;
968 1062
969 } elsif (defined $len) { 1063 } elsif (defined $len) {
970 delete $self->{_rw}; 1064 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
971 $self->{_eof} = 1; 1067 $self->{_eof} = 1;
972 $self->_drain_rbuf; 1068 $self->_drain_rbuf;
973 1069
974 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
975 return $self->error; 1071 return $self->error;

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines