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.67 by root, Fri Jun 6 15:33:10 2008 UTC vs.
Revision 1.85 by root, Thu Aug 21 19:53:19 2008 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict qw(subs vars);
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
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
17=cut 17=cut
18 18
19our $VERSION = 4.15; 19our $VERSION = 4.232;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
49 49
50This module is a helper module to make it easier to do event-based I/O on 50This module is a helper module to make it easier to do event-based I/O on
51filehandles. For utility functions for doing non-blocking connects and accepts 51filehandles. For utility functions for doing non-blocking connects and accepts
52on sockets see L<AnyEvent::Util>. 52on sockets see L<AnyEvent::Util>.
53 53
54The L<AnyEvent::Intro> tutorial contains some well-documented
55AnyEvent::Handle examples.
56
54In the following, when the documentation refers to of "bytes" then this 57In the following, when the documentation refers to of "bytes" then this
55means characters. As sysread and syswrite are used for all I/O, their 58means characters. As sysread and syswrite are used for all I/O, their
56treatment of characters applies to this module as well. 59treatment of characters applies to this module as well.
57 60
58All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
70 73
71=item fh => $filehandle [MANDATORY] 74=item fh => $filehandle [MANDATORY]
72 75
73The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
74 77
75NOTE: The filehandle will be set to non-blocking (using 78NOTE: The filehandle will be set to non-blocking mode (using
76AnyEvent::Util::fh_nonblocking). 79C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in
80that mode.
77 81
78=item on_eof => $cb->($handle) 82=item on_eof => $cb->($handle)
79 83
80Set the callback to be called when an end-of-file condition is detcted, 84Set the callback to be called when an end-of-file condition is detected,
81i.e. in the case of a socket, when the other side has closed the 85i.e. in the case of a socket, when the other side has closed the
82connection cleanly. 86connection cleanly.
83 87
88For sockets, this just means that the other side has stopped sending data,
89you can still try to write data, and, in fact, one can return from the eof
90callback and continue writing data, as only the read part has been shut
91down.
92
84While not mandatory, it is highly recommended to set an eof callback, 93While not mandatory, it is I<highly> recommended to set an eof callback,
85otherwise you might end up with a closed socket while you are still 94otherwise you might end up with a closed socket while you are still
86waiting for data. 95waiting for data.
96
97If an EOF condition has been detected but no C<on_eof> callback has been
98set, then a fatal error will be raised with C<$!> set to <0>.
87 99
88=item on_error => $cb->($handle, $fatal) 100=item on_error => $cb->($handle, $fatal)
89 101
90This is the error callback, which is called when, well, some error 102This is the error callback, which is called when, well, some error
91occured, such as not being able to resolve the hostname, failure to 103occured, such as not being able to resolve the hostname, failure to
92connect or a read error. 104connect or a read error.
93 105
94Some errors are fatal (which is indicated by C<$fatal> being true). On 106Some errors are fatal (which is indicated by C<$fatal> being true). On
95fatal errors the handle object will be shut down and will not be 107fatal errors the handle object will be shut down and will not be usable
108(but you are free to look at the current C< ->rbuf >). Examples of fatal
109errors are an EOF condition with active (but unsatisifable) read watchers
110(C<EPIPE>) or I/O errors.
111
96usable. Non-fatal errors can be retried by simply returning, but it is 112Non-fatal errors can be retried by simply returning, but it is recommended
97recommended to simply ignore this parameter and instead abondon the handle 113to simply ignore this parameter and instead abondon the handle object
98object when this callback is invoked. 114when this callback is invoked. Examples of non-fatal errors are timeouts
115C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
99 116
100On callback entrance, the value of C<$!> contains the operating system 117On callback entrance, the value of C<$!> contains the operating system
101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 118error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
102 119
103While not mandatory, it is I<highly> recommended to set this callback, as 120While not mandatory, it is I<highly> recommended to set this callback, as
123 140
124This sets the callback that is called when the write buffer becomes empty 141This sets the callback that is called when the write buffer becomes empty
125(or when the callback is set and the buffer is empty already). 142(or when the callback is set and the buffer is empty already).
126 143
127To append to the write buffer, use the C<< ->push_write >> method. 144To append to the write buffer, use the C<< ->push_write >> method.
145
146This callback is useful when you don't want to put all of your write data
147into the queue at once, for example, when you want to write the contents
148of some file to the socket you might not want to read the whole file into
149memory and push it into the queue, but instead only read more data from
150the file when the write queue becomes empty.
128 151
129=item timeout => $fractional_seconds 152=item timeout => $fractional_seconds
130 153
131If non-zero, then this enables an "inactivity" timeout: whenever this many 154If non-zero, then this enables an "inactivity" timeout: whenever this many
132seconds pass without a successful read or write on the underlying file 155seconds pass without a successful read or write on the underlying file
156be configured to accept only so-and-so much data that it cannot act on 179be configured to accept only so-and-so much data that it cannot act on
157(for example, when expecting a line, an attacker could send an unlimited 180(for example, when expecting a line, an attacker could send an unlimited
158amount of data without a callback ever being called as long as the line 181amount of data without a callback ever being called as long as the line
159isn't finished). 182isn't finished).
160 183
184=item autocork => <boolean>
185
186When disabled (the default), then C<push_write> will try to immediately
187write the data to the handle if possible. This avoids having to register
188a write watcher and wait for the next event loop iteration, but can be
189inefficient if you write multiple small chunks (this disadvantage is
190usually avoided by your kernel's nagle algorithm, see C<low_delay>).
191
192When enabled, then writes will always be queued till the next event loop
193iteration. This is efficient when you do many small writes per iteration,
194but less efficient when you do a single write only.
195
196=item no_delay => <boolean>
197
198When doing small writes on sockets, your operating system kernel might
199wait a bit for more data before actually sending it out. This is called
200the Nagle algorithm, and usually it is beneficial.
201
202In some situations you want as low a delay as possible, which cna be
203accomplishd by setting this option to true.
204
205The default is your opertaing system's default behaviour, this option
206explicitly enables or disables it, if possible.
207
161=item read_size => <bytes> 208=item read_size => <bytes>
162 209
163The default read block size (the amount of bytes this module will try to read 210The default read block size (the amount of bytes this module will try to read
164during each (loop iteration). Default: C<8192>. 211during each (loop iteration). Default: C<8192>.
165 212
180This will not work for partial TLS data that could not yet been 227This will not work for partial TLS data that could not yet been
181encoded. This data will be lost. 228encoded. This data will be lost.
182 229
183=item tls => "accept" | "connect" | Net::SSLeay::SSL object 230=item tls => "accept" | "connect" | Net::SSLeay::SSL object
184 231
185When this parameter is given, it enables TLS (SSL) mode, that means it 232When this parameter is given, it enables TLS (SSL) mode, that means
186will start making tls handshake and will transparently encrypt/decrypt 233AnyEvent will start a TLS handshake and will transparently encrypt/decrypt
187data. 234data.
188 235
189TLS mode requires Net::SSLeay to be installed (it will be loaded 236TLS mode requires Net::SSLeay to be installed (it will be loaded
190automatically when you try to create a TLS handle). 237automatically when you try to create a TLS handle).
191 238
192For the TLS server side, use C<accept>, and for the TLS client side of a 239Unlike TCP, TLS has a server and client side: for the TLS server side, use
193connection, use C<connect> mode. 240C<accept>, and for the TLS client side of a connection, use C<connect>
241mode.
194 242
195You can also provide your own TLS connection object, but you have 243You can also provide your own TLS connection object, but you have
196to make sure that you call either C<Net::SSLeay::set_connect_state> 244to make sure that you call either C<Net::SSLeay::set_connect_state>
197or C<Net::SSLeay::set_accept_state> on it before you pass it to 245or C<Net::SSLeay::set_accept_state> on it before you pass it to
198AnyEvent::Handle. 246AnyEvent::Handle.
199 247
200See the C<starttls> method if you need to start TLs negotiation later. 248See the C<starttls> method for when need to start TLS negotiation later.
201 249
202=item tls_ctx => $ssl_ctx 250=item tls_ctx => $ssl_ctx
203 251
204Use the given Net::SSLeay::CTX object to create the new TLS connection 252Use the given Net::SSLeay::CTX object to create the new TLS connection
205(unless a connection object was specified directly). If this parameter is 253(unless a connection object was specified directly). If this parameter is
240 } 288 }
241 289
242 $self->{_activity} = AnyEvent->now; 290 $self->{_activity} = AnyEvent->now;
243 $self->_timeout; 291 $self->_timeout;
244 292
245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 293 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
294 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
246 295
247 $self->start_read 296 $self->start_read
248 if $self->{on_read}; 297 if $self->{on_read};
249 298
250 $self 299 $self
257 delete $self->{_rw}; 306 delete $self->{_rw};
258 delete $self->{_ww}; 307 delete $self->{_ww};
259 delete $self->{fh}; 308 delete $self->{fh};
260 309
261 $self->stoptls; 310 $self->stoptls;
311
312 delete $self->{on_read};
313 delete $self->{_queue};
262} 314}
263 315
264sub _error { 316sub _error {
265 my ($self, $errno, $fatal) = @_; 317 my ($self, $errno, $fatal) = @_;
266 318
312 364
313=cut 365=cut
314 366
315sub on_timeout { 367sub on_timeout {
316 $_[0]{on_timeout} = $_[1]; 368 $_[0]{on_timeout} = $_[1];
369}
370
371=item $handle->autocork ($boolean)
372
373Enables or disables the current autocork behaviour (see C<autocork>
374constructor argument).
375
376=cut
377
378=item $handle->no_delay ($boolean)
379
380Enables or disables the C<no_delay> setting (see constructor argument of
381the same name for details).
382
383=cut
384
385sub no_delay {
386 $_[0]{no_delay} = $_[1];
387
388 eval {
389 local $SIG{__DIE__};
390 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
391 };
317} 392}
318 393
319############################################################################# 394#############################################################################
320 395
321=item $handle->timeout ($seconds) 396=item $handle->timeout ($seconds)
436 $self->_error ($!, 1); 511 $self->_error ($!, 1);
437 } 512 }
438 }; 513 };
439 514
440 # try to write data immediately 515 # try to write data immediately
441 $cb->(); 516 $cb->() unless $self->{autocork};
442 517
443 # if still data left in wbuf, we need to poll 518 # if still data left in wbuf, we need to poll
444 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 519 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
445 if length $self->{wbuf}; 520 if length $self->{wbuf};
446 }; 521 };
593ways, the "simple" way, using only C<on_read> and the "complex" way, using 668ways, the "simple" way, using only C<on_read> and the "complex" way, using
594a queue. 669a queue.
595 670
596In the simple case, you just install an C<on_read> callback and whenever 671In the simple case, you just install an C<on_read> callback and whenever
597new data arrives, it will be called. You can then remove some data (if 672new data arrives, it will be called. You can then remove some data (if
598enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 673enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
599or not. 674leave the data there if you want to accumulate more (e.g. when only a
675partial message has been received so far).
600 676
601In the more complex case, you want to queue multiple callbacks. In this 677In the more complex case, you want to queue multiple callbacks. In this
602case, AnyEvent::Handle will call the first queued callback each time new 678case, AnyEvent::Handle will call the first queued callback each time new
603data arrives (also the first time it is queued) and removes it when it has 679data arrives (also the first time it is queued) and removes it when it has
604done its job (see C<push_read>, below). 680done its job (see C<push_read>, below).
622 # handle xml 698 # handle xml
623 }); 699 });
624 }); 700 });
625 }); 701 });
626 702
627Example 2: Implement a client for a protocol that replies either with 703Example 2: Implement a client for a protocol that replies either with "OK"
628"OK" and another line or "ERROR" for one request, and 64 bytes for the 704and another line or "ERROR" for the first request that is sent, and 64
629second request. Due tot he availability of a full queue, we can just 705bytes for the second request. Due to the availability of a queue, we can
630pipeline sending both requests and manipulate the queue as necessary in 706just pipeline sending both requests and manipulate the queue as necessary
631the callbacks: 707in the callbacks.
632 708
633 # request one 709When the first callback is called and sees an "OK" response, it will
710C<unshift> another line-read. This line-read will be queued I<before> the
71164-byte chunk callback.
712
713 # request one, returns either "OK + extra line" or "ERROR"
634 $handle->push_write ("request 1\015\012"); 714 $handle->push_write ("request 1\015\012");
635 715
636 # we expect "ERROR" or "OK" as response, so push a line read 716 # we expect "ERROR" or "OK" as response, so push a line read
637 $handle->push_read (line => sub { 717 $handle->push_read (line => sub {
638 # if we got an "OK", we have to _prepend_ another line, 718 # if we got an "OK", we have to _prepend_ another line,
645 ... 725 ...
646 }); 726 });
647 } 727 }
648 }); 728 });
649 729
650 # request two 730 # request two, simply returns 64 octets
651 $handle->push_write ("request 2\015\012"); 731 $handle->push_write ("request 2\015\012");
652 732
653 # simply read 64 bytes, always 733 # simply read 64 bytes, always
654 $handle->push_read (chunk => 64, sub { 734 $handle->push_read (chunk => 64, sub {
655 my $response = $_[1]; 735 my $response = $_[1];
667 747
668 if ( 748 if (
669 defined $self->{rbuf_max} 749 defined $self->{rbuf_max}
670 && $self->{rbuf_max} < length $self->{rbuf} 750 && $self->{rbuf_max} < length $self->{rbuf}
671 ) { 751 ) {
672 return $self->_error (&Errno::ENOSPC, 1); 752 $self->_error (&Errno::ENOSPC, 1), return;
673 } 753 }
674 754
675 while () { 755 while () {
676 no strict 'refs';
677
678 my $len = length $self->{rbuf}; 756 my $len = length $self->{rbuf};
679 757
680 if (my $cb = shift @{ $self->{_queue} }) { 758 if (my $cb = shift @{ $self->{_queue} }) {
681 unless ($cb->($self)) { 759 unless ($cb->($self)) {
682 if ($self->{_eof}) { 760 if ($self->{_eof}) {
683 # no progress can be made (not enough data and no data forthcoming) 761 # no progress can be made (not enough data and no data forthcoming)
684 $self->_error (&Errno::EPIPE, 1), last; 762 $self->_error (&Errno::EPIPE, 1), return;
685 } 763 }
686 764
687 unshift @{ $self->{_queue} }, $cb; 765 unshift @{ $self->{_queue} }, $cb;
688 last; 766 last;
689 } 767 }
697 && !@{ $self->{_queue} } # and the queue is still empty 775 && !@{ $self->{_queue} } # and the queue is still empty
698 && $self->{on_read} # but we still have on_read 776 && $self->{on_read} # but we still have on_read
699 ) { 777 ) {
700 # no further data will arrive 778 # no further data will arrive
701 # so no progress can be made 779 # so no progress can be made
702 $self->_error (&Errno::EPIPE, 1), last 780 $self->_error (&Errno::EPIPE, 1), return
703 if $self->{_eof}; 781 if $self->{_eof};
704 782
705 last; # more data might arrive 783 last; # more data might arrive
706 } 784 }
707 } else { 785 } else {
709 delete $self->{_rw}; 787 delete $self->{_rw};
710 last; 788 last;
711 } 789 }
712 } 790 }
713 791
792 if ($self->{_eof}) {
793 if ($self->{on_eof}) {
714 $self->{on_eof}($self) 794 $self->{on_eof}($self)
715 if $self->{_eof} && $self->{on_eof}; 795 } else {
796 $self->_error (0, 1);
797 }
798 }
716 799
717 # may need to restart read watcher 800 # may need to restart read watcher
718 unless ($self->{_rw}) { 801 unless ($self->{_rw}) {
719 $self->start_read 802 $self->start_read
720 if $self->{on_read} || @{ $self->{_queue} }; 803 if $self->{on_read} || @{ $self->{_queue} };
846 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 929 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
847 1 930 1
848 } 931 }
849}; 932};
850 933
851# compatibility with older API
852sub push_read_chunk {
853 $_[0]->push_read (chunk => $_[1], $_[2]);
854}
855
856sub unshift_read_chunk {
857 $_[0]->unshift_read (chunk => $_[1], $_[2]);
858}
859
860=item line => [$eol, ]$cb->($handle, $line, $eol) 934=item line => [$eol, ]$cb->($handle, $line, $eol)
861 935
862The callback will be called only once a full line (including the end of 936The callback will be called only once a full line (including the end of
863line marker, C<$eol>) has been read. This line (excluding the end of line 937line marker, C<$eol>) has been read. This line (excluding the end of line
864marker) will be passed to the callback as second argument (C<$line>), and 938marker) will be passed to the callback as second argument (C<$line>), and
879=cut 953=cut
880 954
881register_read_type line => sub { 955register_read_type line => sub {
882 my ($self, $cb, $eol) = @_; 956 my ($self, $cb, $eol) = @_;
883 957
884 $eol = qr|(\015?\012)| if @_ < 3; 958 if (@_ < 3) {
959 # this is more than twice as fast as the generic code below
960 sub {
961 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
962
963 $cb->($_[0], $1, $2);
964 1
965 }
966 } else {
885 $eol = quotemeta $eol unless ref $eol; 967 $eol = quotemeta $eol unless ref $eol;
886 $eol = qr|^(.*?)($eol)|s; 968 $eol = qr|^(.*?)($eol)|s;
887 969
888 sub { 970 sub {
889 $_[0]{rbuf} =~ s/$eol// or return; 971 $_[0]{rbuf} =~ s/$eol// or return;
890 972
891 $cb->($_[0], $1, $2); 973 $cb->($_[0], $1, $2);
974 1
892 1 975 }
893 } 976 }
894}; 977};
895
896# compatibility with older API
897sub push_read_line {
898 my $self = shift;
899 $self->push_read (line => @_);
900}
901
902sub unshift_read_line {
903 my $self = shift;
904 $self->unshift_read (line => @_);
905}
906 978
907=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 979=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
908 980
909Makes a regex match against the regex object C<$accept> and returns 981Makes a regex match against the regex object C<$accept> and returns
910everything up to and including the match. 982everything up to and including the match.
1031register_read_type packstring => sub { 1103register_read_type packstring => sub {
1032 my ($self, $cb, $format) = @_; 1104 my ($self, $cb, $format) = @_;
1033 1105
1034 sub { 1106 sub {
1035 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1107 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1036 defined (my $len = eval { unpack $format, $_[0]->{rbuf} }) 1108 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1037 or return; 1109 or return;
1038 1110
1111 $format = length pack $format, $len;
1112
1113 # bypass unshift if we already have the remaining chunk
1114 if ($format + $len <= length $_[0]{rbuf}) {
1115 my $data = substr $_[0]{rbuf}, $format, $len;
1116 substr $_[0]{rbuf}, 0, $format + $len, "";
1117 $cb->($_[0], $data);
1118 } else {
1039 # remove prefix 1119 # remove prefix
1040 substr $_[0]->{rbuf}, 0, (length pack $format, $len), ""; 1120 substr $_[0]{rbuf}, 0, $format, "";
1041 1121
1042 # read rest 1122 # read remaining chunk
1043 $_[0]->unshift_read (chunk => $len, $cb); 1123 $_[0]->unshift_read (chunk => $len, $cb);
1124 }
1044 1125
1045 1 1126 1
1046 } 1127 }
1047}; 1128};
1048 1129
1105 1186
1106 require Storable; 1187 require Storable;
1107 1188
1108 sub { 1189 sub {
1109 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1190 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1110 defined (my $len = eval { unpack "w", $_[0]->{rbuf} }) 1191 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1111 or return; 1192 or return;
1112 1193
1194 my $format = length pack "w", $len;
1195
1196 # bypass unshift if we already have the remaining chunk
1197 if ($format + $len <= length $_[0]{rbuf}) {
1198 my $data = substr $_[0]{rbuf}, $format, $len;
1199 substr $_[0]{rbuf}, 0, $format + $len, "";
1200 $cb->($_[0], Storable::thaw ($data));
1201 } else {
1113 # remove prefix 1202 # remove prefix
1114 substr $_[0]->{rbuf}, 0, (length pack "w", $len), ""; 1203 substr $_[0]{rbuf}, 0, $format, "";
1115 1204
1116 # read rest 1205 # read remaining chunk
1117 $_[0]->unshift_read (chunk => $len, sub { 1206 $_[0]->unshift_read (chunk => $len, sub {
1118 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1207 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1119 $cb->($_[0], $ref); 1208 $cb->($_[0], $ref);
1120 } else { 1209 } else {
1121 $self->_error (&Errno::EBADMSG); 1210 $self->_error (&Errno::EBADMSG);
1211 }
1122 } 1212 });
1123 }); 1213 }
1214
1215 1
1124 } 1216 }
1125}; 1217};
1126 1218
1127=back 1219=back
1128 1220
1389=over 4 1481=over 4
1390 1482
1391=item * all constructor arguments become object members. 1483=item * all constructor arguments become object members.
1392 1484
1393At least initially, when you pass a C<tls>-argument to the constructor it 1485At least initially, when you pass a C<tls>-argument to the constructor it
1394will end up in C<< $handle->{tls} >>. Those members might be changes or 1486will end up in C<< $handle->{tls} >>. Those members might be changed or
1395mutated later on (for example C<tls> will hold the TLS connection object). 1487mutated later on (for example C<tls> will hold the TLS connection object).
1396 1488
1397=item * other object member names are prefixed with an C<_>. 1489=item * other object member names are prefixed with an C<_>.
1398 1490
1399All object members not explicitly documented (internal use) are prefixed 1491All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines