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.66 by root, Fri Jun 6 15:32:54 2008 UTC vs.
Revision 1.84 by root, Thu Aug 21 19:13:05 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
195You can also provide your own TLS connection object, but you have 242You can also provide your own TLS connection object, but you have
196to make sure that you call either C<Net::SSLeay::set_connect_state> 243to 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 244or C<Net::SSLeay::set_accept_state> on it before you pass it to
198AnyEvent::Handle. 245AnyEvent::Handle.
199 246
200See the C<starttls> method if you need to start TLs negotiation later. 247See the C<starttls> method if you need to start TLS negotiation later.
201 248
202=item tls_ctx => $ssl_ctx 249=item tls_ctx => $ssl_ctx
203 250
204Use the given Net::SSLeay::CTX object to create the new TLS connection 251Use the given Net::SSLeay::CTX object to create the new TLS connection
205(unless a connection object was specified directly). If this parameter is 252(unless a connection object was specified directly). If this parameter is
240 } 287 }
241 288
242 $self->{_activity} = AnyEvent->now; 289 $self->{_activity} = AnyEvent->now;
243 $self->_timeout; 290 $self->_timeout;
244 291
245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 292 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
293 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
246 294
247 $self->start_read 295 $self->start_read
248 if $self->{on_read} || @{ $self->{_queue} }; 296 if $self->{on_read};
249 297
250 $self 298 $self
251} 299}
252 300
253sub _shutdown { 301sub _shutdown {
257 delete $self->{_rw}; 305 delete $self->{_rw};
258 delete $self->{_ww}; 306 delete $self->{_ww};
259 delete $self->{fh}; 307 delete $self->{fh};
260 308
261 $self->stoptls; 309 $self->stoptls;
310
311 delete $self->{on_read};
312 delete $self->{_queue};
262} 313}
263 314
264sub _error { 315sub _error {
265 my ($self, $errno, $fatal) = @_; 316 my ($self, $errno, $fatal) = @_;
266 317
312 363
313=cut 364=cut
314 365
315sub on_timeout { 366sub on_timeout {
316 $_[0]{on_timeout} = $_[1]; 367 $_[0]{on_timeout} = $_[1];
368}
369
370=item $handle->autocork ($boolean)
371
372Enables or disables the current autocork behaviour (see C<autocork>
373constructor argument).
374
375=cut
376
377=item $handle->no_delay ($boolean)
378
379Enables or disables the C<no_delay> setting (see constructor argument of
380the same name for details).
381
382=cut
383
384sub no_delay {
385 $_[0]{no_delay} = $_[1];
386
387 eval {
388 local $SIG{__DIE__};
389 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
390 };
317} 391}
318 392
319############################################################################# 393#############################################################################
320 394
321=item $handle->timeout ($seconds) 395=item $handle->timeout ($seconds)
436 $self->_error ($!, 1); 510 $self->_error ($!, 1);
437 } 511 }
438 }; 512 };
439 513
440 # try to write data immediately 514 # try to write data immediately
441 $cb->(); 515 $cb->() unless $self->{autocork};
442 516
443 # if still data left in wbuf, we need to poll 517 # if still data left in wbuf, we need to poll
444 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 518 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
445 if length $self->{wbuf}; 519 if length $self->{wbuf};
446 }; 520 };
593ways, the "simple" way, using only C<on_read> and the "complex" way, using 667ways, the "simple" way, using only C<on_read> and the "complex" way, using
594a queue. 668a queue.
595 669
596In the simple case, you just install an C<on_read> callback and whenever 670In 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 671new 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 672enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
599or not. 673leave the data there if you want to accumulate more (e.g. when only a
674partial message has been received so far).
600 675
601In the more complex case, you want to queue multiple callbacks. In this 676In the more complex case, you want to queue multiple callbacks. In this
602case, AnyEvent::Handle will call the first queued callback each time new 677case, 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 678data arrives (also the first time it is queued) and removes it when it has
604done its job (see C<push_read>, below). 679done its job (see C<push_read>, below).
622 # handle xml 697 # handle xml
623 }); 698 });
624 }); 699 });
625 }); 700 });
626 701
627Example 2: Implement a client for a protocol that replies either with 702Example 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 703and 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 704bytes for the second request. Due to the availability of a queue, we can
630pipeline sending both requests and manipulate the queue as necessary in 705just pipeline sending both requests and manipulate the queue as necessary
631the callbacks: 706in the callbacks.
632 707
633 # request one 708When the first callback is called and sees an "OK" response, it will
709C<unshift> another line-read. This line-read will be queued I<before> the
71064-byte chunk callback.
711
712 # request one, returns either "OK + extra line" or "ERROR"
634 $handle->push_write ("request 1\015\012"); 713 $handle->push_write ("request 1\015\012");
635 714
636 # we expect "ERROR" or "OK" as response, so push a line read 715 # we expect "ERROR" or "OK" as response, so push a line read
637 $handle->push_read (line => sub { 716 $handle->push_read (line => sub {
638 # if we got an "OK", we have to _prepend_ another line, 717 # if we got an "OK", we have to _prepend_ another line,
645 ... 724 ...
646 }); 725 });
647 } 726 }
648 }); 727 });
649 728
650 # request two 729 # request two, simply returns 64 octets
651 $handle->push_write ("request 2\015\012"); 730 $handle->push_write ("request 2\015\012");
652 731
653 # simply read 64 bytes, always 732 # simply read 64 bytes, always
654 $handle->push_read (chunk => 64, sub { 733 $handle->push_read (chunk => 64, sub {
655 my $response = $_[1]; 734 my $response = $_[1];
667 746
668 if ( 747 if (
669 defined $self->{rbuf_max} 748 defined $self->{rbuf_max}
670 && $self->{rbuf_max} < length $self->{rbuf} 749 && $self->{rbuf_max} < length $self->{rbuf}
671 ) { 750 ) {
672 return $self->_error (&Errno::ENOSPC, 1); 751 $self->_error (&Errno::ENOSPC, 1), return;
673 } 752 }
674 753
675 while () { 754 while () {
676 no strict 'refs';
677
678 my $len = length $self->{rbuf}; 755 my $len = length $self->{rbuf};
679 756
680 if (my $cb = shift @{ $self->{_queue} }) { 757 if (my $cb = shift @{ $self->{_queue} }) {
681 unless ($cb->($self)) { 758 unless ($cb->($self)) {
682 if ($self->{_eof}) { 759 if ($self->{_eof}) {
683 # no progress can be made (not enough data and no data forthcoming) 760 # no progress can be made (not enough data and no data forthcoming)
684 $self->_error (&Errno::EPIPE, 1), last; 761 $self->_error (&Errno::EPIPE, 1), return;
685 } 762 }
686 763
687 unshift @{ $self->{_queue} }, $cb; 764 unshift @{ $self->{_queue} }, $cb;
688 last; 765 last;
689 } 766 }
697 && !@{ $self->{_queue} } # and the queue is still empty 774 && !@{ $self->{_queue} } # and the queue is still empty
698 && $self->{on_read} # but we still have on_read 775 && $self->{on_read} # but we still have on_read
699 ) { 776 ) {
700 # no further data will arrive 777 # no further data will arrive
701 # so no progress can be made 778 # so no progress can be made
702 $self->_error (&Errno::EPIPE, 1), last 779 $self->_error (&Errno::EPIPE, 1), return
703 if $self->{_eof}; 780 if $self->{_eof};
704 781
705 last; # more data might arrive 782 last; # more data might arrive
706 } 783 }
707 } else { 784 } else {
709 delete $self->{_rw}; 786 delete $self->{_rw};
710 last; 787 last;
711 } 788 }
712 } 789 }
713 790
791 if ($self->{_eof}) {
792 if ($self->{on_eof}) {
714 $self->{on_eof}($self) 793 $self->{on_eof}($self)
715 if $self->{_eof} && $self->{on_eof}; 794 } else {
795 $self->_error (0, 1);
796 }
797 }
716 798
717 # may need to restart read watcher 799 # may need to restart read watcher
718 unless ($self->{_rw}) { 800 unless ($self->{_rw}) {
719 $self->start_read 801 $self->start_read
720 if $self->{on_read} || @{ $self->{_queue} }; 802 if $self->{on_read} || @{ $self->{_queue} };
846 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 928 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
847 1 929 1
848 } 930 }
849}; 931};
850 932
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) 933=item line => [$eol, ]$cb->($handle, $line, $eol)
861 934
862The callback will be called only once a full line (including the end of 935The 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 936line 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 937marker) will be passed to the callback as second argument (C<$line>), and
879=cut 952=cut
880 953
881register_read_type line => sub { 954register_read_type line => sub {
882 my ($self, $cb, $eol) = @_; 955 my ($self, $cb, $eol) = @_;
883 956
884 $eol = qr|(\015?\012)| if @_ < 3; 957 if (@_ < 3) {
958 # this is more than twice as fast as the generic code below
959 sub {
960 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
961
962 $cb->($_[0], $1, $2);
963 1
964 }
965 } else {
885 $eol = quotemeta $eol unless ref $eol; 966 $eol = quotemeta $eol unless ref $eol;
886 $eol = qr|^(.*?)($eol)|s; 967 $eol = qr|^(.*?)($eol)|s;
887 968
888 sub { 969 sub {
889 $_[0]{rbuf} =~ s/$eol// or return; 970 $_[0]{rbuf} =~ s/$eol// or return;
890 971
891 $cb->($_[0], $1, $2); 972 $cb->($_[0], $1, $2);
973 1
892 1 974 }
893 } 975 }
894}; 976};
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 977
907=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 978=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
908 979
909Makes a regex match against the regex object C<$accept> and returns 980Makes a regex match against the regex object C<$accept> and returns
910everything up to and including the match. 981everything up to and including the match.
1031register_read_type packstring => sub { 1102register_read_type packstring => sub {
1032 my ($self, $cb, $format) = @_; 1103 my ($self, $cb, $format) = @_;
1033 1104
1034 sub { 1105 sub {
1035 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1106 # 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} }) 1107 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1037 or return; 1108 or return;
1038 1109
1110 $format = length pack $format, $len;
1111
1112 # bypass unshift if we already have the remaining chunk
1113 if ($format + $len <= length $_[0]{rbuf}) {
1114 my $data = substr $_[0]{rbuf}, $format, $len;
1115 substr $_[0]{rbuf}, 0, $format + $len, "";
1116 $cb->($_[0], $data);
1117 } else {
1039 # remove prefix 1118 # remove prefix
1040 substr $_[0]->{rbuf}, 0, (length pack $format, $len), ""; 1119 substr $_[0]{rbuf}, 0, $format, "";
1041 1120
1042 # read rest 1121 # read remaining chunk
1043 $_[0]->unshift_read (chunk => $len, $cb); 1122 $_[0]->unshift_read (chunk => $len, $cb);
1123 }
1044 1124
1045 1 1125 1
1046 } 1126 }
1047}; 1127};
1048 1128
1105 1185
1106 require Storable; 1186 require Storable;
1107 1187
1108 sub { 1188 sub {
1109 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1189 # 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} }) 1190 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1111 or return; 1191 or return;
1112 1192
1193 my $format = length pack "w", $len;
1194
1195 # bypass unshift if we already have the remaining chunk
1196 if ($format + $len <= length $_[0]{rbuf}) {
1197 my $data = substr $_[0]{rbuf}, $format, $len;
1198 substr $_[0]{rbuf}, 0, $format + $len, "";
1199 $cb->($_[0], Storable::thaw ($data));
1200 } else {
1113 # remove prefix 1201 # remove prefix
1114 substr $_[0]->{rbuf}, 0, (length pack "w", $len), ""; 1202 substr $_[0]{rbuf}, 0, $format, "";
1115 1203
1116 # read rest 1204 # read remaining chunk
1117 $_[0]->unshift_read (chunk => $len, sub { 1205 $_[0]->unshift_read (chunk => $len, sub {
1118 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1206 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1119 $cb->($_[0], $ref); 1207 $cb->($_[0], $ref);
1120 } else { 1208 } else {
1121 $self->_error (&Errno::EBADMSG); 1209 $self->_error (&Errno::EBADMSG);
1210 }
1122 } 1211 });
1123 }); 1212 }
1213
1214 1
1124 } 1215 }
1125}; 1216};
1126 1217
1127=back 1218=back
1128 1219
1389=over 4 1480=over 4
1390 1481
1391=item * all constructor arguments become object members. 1482=item * all constructor arguments become object members.
1392 1483
1393At least initially, when you pass a C<tls>-argument to the constructor it 1484At 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 1485will 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). 1486mutated later on (for example C<tls> will hold the TLS connection object).
1396 1487
1397=item * other object member names are prefixed with an C<_>. 1488=item * other object member names are prefixed with an C<_>.
1398 1489
1399All object members not explicitly documented (internal use) are prefixed 1490All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines