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.40 by root, Tue May 27 05:36:27 2008 UTC vs.
Revision 1.44 by root, Thu May 29 00:00:07 2008 UTC

… …
2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAWOULDBLOCK); 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> errror 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
… …
172 192
173=item json => JSON or JSON::XS object 193=item json => JSON or JSON::XS object
174 194
175This is the json coder object used by the C<json> read and write types. 195This is the json coder object used by the C<json> read and write types.
176 196
177If you don't supply it, then AnyEvent::Handle will use C<encode_json> and 197If you don't supply it, then AnyEvent::Handle will create and use a
178C<decode_json>. 198suitable one, which will write and expect UTF-8 encoded JSON texts.
179 199
180Note that you are responsible to depend on the JSON module if you want to 200Note that you are responsible to depend on the JSON module if you want to
181use this functionality, as AnyEvent does not have a dependency itself. 201use this functionality, as AnyEvent does not have a dependency itself.
182 202
183=item filter_r => $cb 203=item filter_r => $cb
… …
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
… …
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, "";
320 410
411 $self->{_activity} = AnyEvent->now;
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
325 delete $self->{_ww} unless length $self->{wbuf}; 417 delete $self->{_ww} unless length $self->{wbuf};
326 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
327 $self->error; 419 $self->error;
328 } 420 }
329 }; 421 };
330 422
331 # try to write data immediately 423 # try to write data immediately
… …
395in UTF-8. 487in UTF-8.
396 488
397JSON objects (and arrays) are self-delimiting, so you can write JSON at 489JSON objects (and arrays) are self-delimiting, so you can write JSON at
398one end of a handle and read them at the other end without using any 490one end of a handle and read them at the other end without using any
399additional framing. 491additional framing.
492
493The generated JSON text is guaranteed not to contain any newlines: While
494this module doesn't need delimiters after or between JSON texts to be
495able to read them, many other languages depend on that.
496
497A simple RPC protocol that interoperates easily with others is to send
498JSON arrays (or objects, although arrays are usually the better choice as
499they mimic how function argument passing works) and a newline after each
500JSON text:
501
502 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
503 $handle->push_write ("\012");
504
505An AnyEvent::Handle receiver would simply use the C<json> read type and
506rely on the fact that the newline will be skipped as leading whitespace:
507
508 $handle->push_read (json => sub { my $array = $_[1]; ... });
509
510Other languages could read single lines terminated by a newline and pass
511this line into their JSON decoder of choice.
400 512
401=cut 513=cut
402 514
403register_write_type json => sub { 515register_write_type json => sub {
404 my ($self, $ref) = @_; 516 my ($self, $ref) = @_;
… …
8592.09 (and JSON::XS version 2.2) and above. You have to provide a 9712.09 (and JSON::XS version 2.2) and above. You have to provide a
860dependency on your own: this module will load the JSON module, but 972dependency on your own: this module will load the JSON module, but
861AnyEvent does not depend on it itself. 973AnyEvent does not depend on it itself.
862 974
863Since JSON texts are fully self-delimiting, the C<json> read and write 975Since JSON texts are fully self-delimiting, the C<json> read and write
864types are an ideal simple RPC protocol: just exchange JSON datagrams. 976types are an ideal simple RPC protocol: just exchange JSON datagrams. See
977the C<json> write type description, above, for an actual example.
865 978
866=cut 979=cut
867 980
868register_read_type json => sub { 981register_read_type json => sub {
869 my ($self, $cb, $accept, $reject, $skip) = @_; 982 my ($self, $cb, $accept, $reject, $skip) = @_;
… …
871 require JSON; 984 require JSON;
872 985
873 my $data; 986 my $data;
874 my $rbuf = \$self->{rbuf}; 987 my $rbuf = \$self->{rbuf};
875 988
876 my $json = $self->{json} ||= JSON::XS->new->utf8; 989 my $json = $self->{json} ||= JSON->new->utf8;
877 990
878 sub { 991 sub {
879 my $ref = $json->incr_parse ($self->{rbuf}); 992 my $ref = $json->incr_parse ($self->{rbuf});
880 993
881 if ($ref) { 994 if ($ref) {
… …
939 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1052 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
940 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1053 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
941 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;
942 1055
943 if ($len > 0) { 1056 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now;
1058
944 $self->{filter_r} 1059 $self->{filter_r}
945 ? $self->{filter_r}->($self, $rbuf) 1060 ? $self->{filter_r}->($self, $rbuf)
946 : $self->_drain_rbuf; 1061 : $self->_drain_rbuf;
947 1062
948 } elsif (defined $len) { 1063 } elsif (defined $len) {
949 delete $self->{_rw}; 1064 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
950 $self->{_eof} = 1; 1067 $self->{_eof} = 1;
951 $self->_drain_rbuf; 1068 $self->_drain_rbuf;
952 1069
953 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
954 return $self->error; 1071 return $self->error;
955 } 1072 }
956 }); 1073 });
957 } 1074 }
958} 1075}

Diff Legend

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