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.39 by root, Tue May 27 04:59:51 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
73The filehandle this L<AnyEvent::Handle> object will operate on. 73The filehandle this L<AnyEvent::Handle> object will operate on.
74 74
75NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
76AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
77 77
78=item on_eof => $cb->($self) 78=item on_eof => $cb->($handle)
79 79
80Set the callback to be called on EOF. 80Set the callback to be called on EOF.
81 81
82While not mandatory, it is highly recommended to set an eof callback, 82While not mandatory, it is highly recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still 83otherwise you might end up with a closed socket while you are still
84waiting for data. 84waiting for data.
85 85
86=item on_error => $cb->($self) 86=item on_error => $cb->($handle)
87 87
88This is the fatal error callback, that is called when, well, a fatal error 88This is the fatal error callback, that is called when, well, a fatal error
89occurs, such as not being able to resolve the hostname, failure to connect 89occurs, such as not being able to resolve the hostname, failure to connect
90or a read error. 90or a read error.
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
102you will not be notified of errors otherwise. The default simply calls 102you will not be notified of errors otherwise. The default simply calls
103die. 103die.
104 104
105=item on_read => $cb->($self) 105=item on_read => $cb->($handle)
106 106
107This sets the default read callback, which is called when data arrives 107This sets the default read callback, which is called when data arrives
108and no read request is in the queue. 108and no read request is in the queue.
109 109
110To access (and remove data from) the read buffer, use the C<< ->rbuf >> 110To access (and remove data from) the read buffer, use the C<< ->rbuf >>
111method or access the C<$self->{rbuf}> member directly. 111method or access the C<$handle->{rbuf}> member directly.
112 112
113When an EOF condition is detected then AnyEvent::Handle will first try to 113When an EOF condition is detected then AnyEvent::Handle will first try to
114feed all the remaining data to the queued callbacks and C<on_read> before 114feed all the remaining data to the queued callbacks and C<on_read> before
115calling the C<on_eof> callback. If no progress can be made, then a fatal 115calling the C<on_eof> callback. If no progress can be made, then a fatal
116error will be raised (with C<$!> set to C<EPIPE>). 116error will be raised (with C<$!> set to C<EPIPE>).
117 117
118=item on_drain => $cb->() 118=item on_drain => $cb->($handle)
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
168 188
169Use the given Net::SSLeay::CTX object to create the new TLS connection 189Use the given Net::SSLeay::CTX object to create the new TLS connection
170(unless a connection object was specified directly). If this parameter is 190(unless a connection object was specified directly). If this parameter is
171missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 191missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
172 192
193=item json => JSON or JSON::XS object
194
195This is the json coder object used by the C<json> read and write types.
196
197If you don't supply it, then AnyEvent::Handle will create and use a
198suitable one, which will write and expect UTF-8 encoded JSON texts.
199
200Note that you are responsible to depend on the JSON module if you want to
201use this functionality, as AnyEvent does not have a dependency itself.
202
173=item filter_r => $cb 203=item filter_r => $cb
174 204
175=item filter_w => $cb 205=item filter_w => $cb
176 206
177These exist, but are undocumented at this time. 207These exist, but are undocumented at this time.
192 if ($self->{tls}) { 222 if ($self->{tls}) {
193 require Net::SSLeay; 223 require Net::SSLeay;
194 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
195 } 225 }
196 226
197 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
198 $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
199 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
200 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 231
232 $self->{_activity} = AnyEvent->now;
233 $self->_timeout;
201 234
202 $self->start_read; 235 $self->start_read;
203 236
204 $self 237 $self
205} 238}
252 285
253sub on_eof { 286sub on_eof {
254 $_[0]{on_eof} = $_[1]; 287 $_[0]{on_eof} = $_[1];
255} 288}
256 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
257############################################################################# 357#############################################################################
258 358
259=back 359=back
260 360
261=head2 WRITE QUEUE 361=head2 WRITE QUEUE
306 my $len = syswrite $self->{fh}, $self->{wbuf}; 406 my $len = syswrite $self->{fh}, $self->{wbuf};
307 407
308 if ($len >= 0) { 408 if ($len >= 0) {
309 substr $self->{wbuf}, 0, $len, ""; 409 substr $self->{wbuf}, 0, $len, "";
310 410
411 $self->{_activity} = AnyEvent->now;
412
311 $self->{on_drain}($self) 413 $self->{on_drain}($self)
312 if $self->{low_water_mark} >= length $self->{wbuf} 414 if $self->{low_water_mark} >= length $self->{wbuf}
313 && $self->{on_drain}; 415 && $self->{on_drain};
314 416
315 delete $self->{_ww} unless length $self->{wbuf}; 417 delete $self->{_ww} unless length $self->{wbuf};
316 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
317 $self->error; 419 $self->error;
318 } 420 }
319 }; 421 };
320 422
321 # try to write data immediately 423 # try to write data immediately
378 sprintf "%d:%s,", (length $string), $string 480 sprintf "%d:%s,", (length $string), $string
379}; 481};
380 482
381=item json => $array_or_hashref 483=item json => $array_or_hashref
382 484
485Encodes the given hash or array reference into a JSON object. Unless you
486provide your own JSON object, this means it will be encoded to JSON text
487in UTF-8.
488
489JSON objects (and arrays) are self-delimiting, so you can write JSON at
490one end of a handle and read them at the other end without using any
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.
512
513=cut
514
515register_write_type json => sub {
516 my ($self, $ref) = @_;
517
518 require JSON;
519
520 $self->{json} ? $self->{json}->encode ($ref)
521 : JSON::encode_json ($ref)
522};
523
383=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args) 524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
384 525
385This function (not method) lets you add your own types to C<push_write>. 526This function (not method) lets you add your own types to C<push_write>.
386Whenever the given C<type> is used, C<push_write> will invoke the code 527Whenever the given C<type> is used, C<push_write> will invoke the code
387reference with the handle object and the remaining arguments. 528reference with the handle object and the remaining arguments.
388 529
628Predefined types are (if you have ideas for additional types, feel free to 769Predefined types are (if you have ideas for additional types, feel free to
629drop by and tell us): 770drop by and tell us):
630 771
631=over 4 772=over 4
632 773
633=item chunk => $octets, $cb->($self, $data) 774=item chunk => $octets, $cb->($handle, $data)
634 775
635Invoke the callback only once C<$octets> bytes have been read. Pass the 776Invoke the callback only once C<$octets> bytes have been read. Pass the
636data read to the callback. The callback will never be called with less 777data read to the callback. The callback will never be called with less
637data. 778data.
638 779
661 802
662sub unshift_read_chunk { 803sub unshift_read_chunk {
663 $_[0]->unshift_read (chunk => $_[1], $_[2]); 804 $_[0]->unshift_read (chunk => $_[1], $_[2]);
664} 805}
665 806
666=item line => [$eol, ]$cb->($self, $line, $eol) 807=item line => [$eol, ]$cb->($handle, $line, $eol)
667 808
668The callback will be called only once a full line (including the end of 809The callback will be called only once a full line (including the end of
669line marker, C<$eol>) has been read. This line (excluding the end of line 810line marker, C<$eol>) has been read. This line (excluding the end of line
670marker) will be passed to the callback as second argument (C<$line>), and 811marker) will be passed to the callback as second argument (C<$line>), and
671the end of line marker as the third argument (C<$eol>). 812the end of line marker as the third argument (C<$eol>).
708sub unshift_read_line { 849sub unshift_read_line {
709 my $self = shift; 850 my $self = shift;
710 $self->unshift_read (line => @_); 851 $self->unshift_read (line => @_);
711} 852}
712 853
713=item netstring => $cb->($string) 854=item netstring => $cb->($handle, $string)
714 855
715A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement). 856A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
716 857
717Throws an error with C<$!> set to EBADMSG on format violations. 858Throws an error with C<$!> set to EBADMSG on format violations.
718 859
746 887
747 1 888 1
748 } 889 }
749}; 890};
750 891
751=item regex => $accept[, $reject[, $skip], $cb->($data) 892=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
752 893
753Makes a regex match against the regex object C<$accept> and returns 894Makes a regex match against the regex object C<$accept> and returns
754everything up to and including the match. 895everything up to and including the match.
755 896
756Example: read a single line terminated by '\n'. 897Example: read a single line terminated by '\n'.
817 958
818 () 959 ()
819 } 960 }
820}; 961};
821 962
963=item json => $cb->($handle, $hash_or_arrayref)
964
965Reads a JSON object or array, decodes it and passes it to the callback.
966
967If a C<json> object was passed to the constructor, then that will be used
968for the final decode, otherwise it will create a JSON coder expecting UTF-8.
969
970This read type uses the incremental parser available with JSON version
9712.09 (and JSON::XS version 2.2) and above. You have to provide a
972dependency on your own: this module will load the JSON module, but
973AnyEvent does not depend on it itself.
974
975Since JSON texts are fully self-delimiting, the C<json> read and write
976types are an ideal simple RPC protocol: just exchange JSON datagrams. See
977the C<json> write type description, above, for an actual example.
978
979=cut
980
981register_read_type json => sub {
982 my ($self, $cb, $accept, $reject, $skip) = @_;
983
984 require JSON;
985
986 my $data;
987 my $rbuf = \$self->{rbuf};
988
989 my $json = $self->{json} ||= JSON->new->utf8;
990
991 sub {
992 my $ref = $json->incr_parse ($self->{rbuf});
993
994 if ($ref) {
995 $self->{rbuf} = $json->incr_text;
996 $json->incr_text = "";
997 $cb->($self, $ref);
998
999 1
1000 } else {
1001 $self->{rbuf} = "";
1002 ()
1003 }
1004 }
1005};
1006
822=back 1007=back
823 1008
824=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args) 1009=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
825 1010
826This function (not method) lets you add your own types to C<push_read>. 1011This function (not method) lets you add your own types to C<push_read>.
827 1012
828Whenever the given C<type> is used, C<push_read> will invoke the code 1013Whenever the given C<type> is used, C<push_read> will invoke the code
829reference with the handle object, the callback and the remaining 1014reference with the handle object, the callback and the remaining
831 1016
832The code reference is supposed to return a callback (usually a closure) 1017The code reference is supposed to return a callback (usually a closure)
833that works as a plain read callback (see C<< ->push_read ($cb) >>). 1018that works as a plain read callback (see C<< ->push_read ($cb) >>).
834 1019
835It should invoke the passed callback when it is done reading (remember to 1020It should invoke the passed callback when it is done reading (remember to
836pass C<$self> as first argument as all other callbacks do that). 1021pass C<$handle> as first argument as all other callbacks do that).
837 1022
838Note that this is a function, and all types registered this way will be 1023Note that this is a function, and all types registered this way will be
839global, so try to use unique names. 1024global, so try to use unique names.
840 1025
841For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1026For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
867 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1052 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
868 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1053 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
869 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;
870 1055
871 if ($len > 0) { 1056 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now;
1058
872 $self->{filter_r} 1059 $self->{filter_r}
873 ? $self->{filter_r}->($self, $rbuf) 1060 ? $self->{filter_r}->($self, $rbuf)
874 : $self->_drain_rbuf; 1061 : $self->_drain_rbuf;
875 1062
876 } elsif (defined $len) { 1063 } elsif (defined $len) {
877 delete $self->{_rw}; 1064 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
878 $self->{_eof} = 1; 1067 $self->{_eof} = 1;
879 $self->_drain_rbuf; 1068 $self->_drain_rbuf;
880 1069
881 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
882 return $self->error; 1071 return $self->error;
883 } 1072 }
884 }); 1073 });
885 } 1074 }
886} 1075}

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines