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.82 by root, Thu Aug 21 18:45:16 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;
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->($handle) 78=item on_eof => $cb->($handle)
79 79
80Set the callback to be called when an end-of-file condition is detcted, 80Set 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 81i.e. in the case of a socket, when the other side has closed the
82connection cleanly. 82connection cleanly.
83 83
84For sockets, this just means that the other side has stopped sending data,
85you can still try to write data, and, in fact, one can return from the eof
86callback and continue writing data, as only the read part has been shut
87down.
88
84While not mandatory, it is highly recommended to set an eof callback, 89While 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 90otherwise you might end up with a closed socket while you are still
86waiting for data. 91waiting for data.
92
93If an EOF condition has been detected but no C<on_eof> callback has been
94set, then a fatal error will be raised with C<$!> set to <0>.
87 95
88=item on_error => $cb->($handle, $fatal) 96=item on_error => $cb->($handle, $fatal)
89 97
90This is the error callback, which is called when, well, some error 98This is the error callback, which is called when, well, some error
91occured, such as not being able to resolve the hostname, failure to 99occured, such as not being able to resolve the hostname, failure to
92connect or a read error. 100connect or a read error.
93 101
94Some errors are fatal (which is indicated by C<$fatal> being true). On 102Some 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 103fatal errors the handle object will be shut down and will not be usable
104(but you are free to look at the current C< ->rbuf >). Examples of fatal
105errors are an EOF condition with active (but unsatisifable) read watchers
106(C<EPIPE>) or I/O errors.
107
96usable. Non-fatal errors can be retried by simply returning, but it is 108Non-fatal errors can be retried by simply returning, but it is recommended
97recommended to simply ignore this parameter and instead abondon the handle 109to simply ignore this parameter and instead abondon the handle object
98object when this callback is invoked. 110when this callback is invoked. Examples of non-fatal errors are timeouts
111C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
99 112
100On callback entrance, the value of C<$!> contains the operating system 113On callback entrance, the value of C<$!> contains the operating system
101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 114error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
102 115
103While not mandatory, it is I<highly> recommended to set this callback, as 116While not mandatory, it is I<highly> recommended to set this callback, as
123 136
124This sets the callback that is called when the write buffer becomes empty 137This 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). 138(or when the callback is set and the buffer is empty already).
126 139
127To append to the write buffer, use the C<< ->push_write >> method. 140To append to the write buffer, use the C<< ->push_write >> method.
141
142This callback is useful when you don't want to put all of your write data
143into the queue at once, for example, when you want to write the contents
144of some file to the socket you might not want to read the whole file into
145memory and push it into the queue, but instead only read more data from
146the file when the write queue becomes empty.
128 147
129=item timeout => $fractional_seconds 148=item timeout => $fractional_seconds
130 149
131If non-zero, then this enables an "inactivity" timeout: whenever this many 150If non-zero, then this enables an "inactivity" timeout: whenever this many
132seconds pass without a successful read or write on the underlying file 151seconds 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 175be 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 176(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 177amount of data without a callback ever being called as long as the line
159isn't finished). 178isn't finished).
160 179
180=item autocork => <boolean>
181
182When disabled (the default), then C<push_write> will try to immediately
183write the data to the handle if possible. This avoids having to register
184a write watcher and wait for the next event loop iteration, but can be
185inefficient if you write multiple small chunks (this disadvantage is
186usually avoided by your kernel's nagle algorithm, see C<low_delay>).
187
188When enabled, then writes will always be queued till the next event loop
189iteration. This is efficient when you do many small writes per iteration,
190but less efficient when you do a single write only.
191
192=item no_delay => <boolean>
193
194When doing small writes on sockets, your operating system kernel might
195wait a bit for more data before actually sending it out. This is called
196the Nagle algorithm, and usually it is beneficial.
197
198In some situations you want as low a delay as possible, which cna be
199accomplishd by setting this option to true.
200
201The default is your opertaing system's default behaviour, this option
202explicitly enables or disables it, if possible.
203
161=item read_size => <bytes> 204=item read_size => <bytes>
162 205
163The default read block size (the amount of bytes this module will try to read 206The default read block size (the amount of bytes this module will try to read
164during each (loop iteration). Default: C<8192>. 207during each (loop iteration). Default: C<8192>.
165 208
195You can also provide your own TLS connection object, but you have 238You can also provide your own TLS connection object, but you have
196to make sure that you call either C<Net::SSLeay::set_connect_state> 239to 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 240or C<Net::SSLeay::set_accept_state> on it before you pass it to
198AnyEvent::Handle. 241AnyEvent::Handle.
199 242
200See the C<starttls> method if you need to start TLs negotiation later. 243See the C<starttls> method if you need to start TLS negotiation later.
201 244
202=item tls_ctx => $ssl_ctx 245=item tls_ctx => $ssl_ctx
203 246
204Use the given Net::SSLeay::CTX object to create the new TLS connection 247Use the given Net::SSLeay::CTX object to create the new TLS connection
205(unless a connection object was specified directly). If this parameter is 248(unless a connection object was specified directly). If this parameter is
240 } 283 }
241 284
242 $self->{_activity} = AnyEvent->now; 285 $self->{_activity} = AnyEvent->now;
243 $self->_timeout; 286 $self->_timeout;
244 287
245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 288 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
289 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
246 290
247 $self->start_read 291 $self->start_read
248 if $self->{on_read}; 292 if $self->{on_read};
249 293
250 $self 294 $self
257 delete $self->{_rw}; 301 delete $self->{_rw};
258 delete $self->{_ww}; 302 delete $self->{_ww};
259 delete $self->{fh}; 303 delete $self->{fh};
260 304
261 $self->stoptls; 305 $self->stoptls;
306
307 delete $self->{on_read};
308 delete $self->{_queue};
262} 309}
263 310
264sub _error { 311sub _error {
265 my ($self, $errno, $fatal) = @_; 312 my ($self, $errno, $fatal) = @_;
266 313
312 359
313=cut 360=cut
314 361
315sub on_timeout { 362sub on_timeout {
316 $_[0]{on_timeout} = $_[1]; 363 $_[0]{on_timeout} = $_[1];
364}
365
366=item $handle->autocork ($boolean)
367
368Enables or disables the current autocork behaviour (see C<autocork>
369constructor argument).
370
371=cut
372
373=item $handle->no_delay ($boolean)
374
375Enables or disables the C<no_delay> setting (see constructor argument of
376the same name for details).
377
378=cut
379
380sub no_delay {
381 $_[0]{no_delay} = $_[1];
382
383 eval {
384 local $SIG{__DIE__};
385 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
386 };
317} 387}
318 388
319############################################################################# 389#############################################################################
320 390
321=item $handle->timeout ($seconds) 391=item $handle->timeout ($seconds)
436 $self->_error ($!, 1); 506 $self->_error ($!, 1);
437 } 507 }
438 }; 508 };
439 509
440 # try to write data immediately 510 # try to write data immediately
441 $cb->(); 511 $cb->() unless $self->{autocork};
442 512
443 # if still data left in wbuf, we need to poll 513 # if still data left in wbuf, we need to poll
444 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 514 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
445 if length $self->{wbuf}; 515 if length $self->{wbuf};
446 }; 516 };
593ways, the "simple" way, using only C<on_read> and the "complex" way, using 663ways, the "simple" way, using only C<on_read> and the "complex" way, using
594a queue. 664a queue.
595 665
596In the simple case, you just install an C<on_read> callback and whenever 666In 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 667new 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 668enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
599or not. 669leave the data there if you want to accumulate more (e.g. when only a
670partial message has been received so far).
600 671
601In the more complex case, you want to queue multiple callbacks. In this 672In the more complex case, you want to queue multiple callbacks. In this
602case, AnyEvent::Handle will call the first queued callback each time new 673case, 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 674data arrives (also the first time it is queued) and removes it when it has
604done its job (see C<push_read>, below). 675done its job (see C<push_read>, below).
622 # handle xml 693 # handle xml
623 }); 694 });
624 }); 695 });
625 }); 696 });
626 697
627Example 2: Implement a client for a protocol that replies either with 698Example 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 699and 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 700bytes for the second request. Due to the availability of a queue, we can
630pipeline sending both requests and manipulate the queue as necessary in 701just pipeline sending both requests and manipulate the queue as necessary
631the callbacks: 702in the callbacks.
632 703
633 # request one 704When the first callback is called and sees an "OK" response, it will
705C<unshift> another line-read. This line-read will be queued I<before> the
70664-byte chunk callback.
707
708 # request one, returns either "OK + extra line" or "ERROR"
634 $handle->push_write ("request 1\015\012"); 709 $handle->push_write ("request 1\015\012");
635 710
636 # we expect "ERROR" or "OK" as response, so push a line read 711 # we expect "ERROR" or "OK" as response, so push a line read
637 $handle->push_read (line => sub { 712 $handle->push_read (line => sub {
638 # if we got an "OK", we have to _prepend_ another line, 713 # if we got an "OK", we have to _prepend_ another line,
645 ... 720 ...
646 }); 721 });
647 } 722 }
648 }); 723 });
649 724
650 # request two 725 # request two, simply returns 64 octets
651 $handle->push_write ("request 2\015\012"); 726 $handle->push_write ("request 2\015\012");
652 727
653 # simply read 64 bytes, always 728 # simply read 64 bytes, always
654 $handle->push_read (chunk => 64, sub { 729 $handle->push_read (chunk => 64, sub {
655 my $response = $_[1]; 730 my $response = $_[1];
667 742
668 if ( 743 if (
669 defined $self->{rbuf_max} 744 defined $self->{rbuf_max}
670 && $self->{rbuf_max} < length $self->{rbuf} 745 && $self->{rbuf_max} < length $self->{rbuf}
671 ) { 746 ) {
672 return $self->_error (&Errno::ENOSPC, 1); 747 $self->_error (&Errno::ENOSPC, 1), return;
673 } 748 }
674 749
675 while () { 750 while () {
676 no strict 'refs';
677
678 my $len = length $self->{rbuf}; 751 my $len = length $self->{rbuf};
679 752
680 if (my $cb = shift @{ $self->{_queue} }) { 753 if (my $cb = shift @{ $self->{_queue} }) {
681 unless ($cb->($self)) { 754 unless ($cb->($self)) {
682 if ($self->{_eof}) { 755 if ($self->{_eof}) {
683 # no progress can be made (not enough data and no data forthcoming) 756 # no progress can be made (not enough data and no data forthcoming)
684 $self->_error (&Errno::EPIPE, 1), last; 757 $self->_error (&Errno::EPIPE, 1), return;
685 } 758 }
686 759
687 unshift @{ $self->{_queue} }, $cb; 760 unshift @{ $self->{_queue} }, $cb;
688 last; 761 last;
689 } 762 }
697 && !@{ $self->{_queue} } # and the queue is still empty 770 && !@{ $self->{_queue} } # and the queue is still empty
698 && $self->{on_read} # but we still have on_read 771 && $self->{on_read} # but we still have on_read
699 ) { 772 ) {
700 # no further data will arrive 773 # no further data will arrive
701 # so no progress can be made 774 # so no progress can be made
702 $self->_error (&Errno::EPIPE, 1), last 775 $self->_error (&Errno::EPIPE, 1), return
703 if $self->{_eof}; 776 if $self->{_eof};
704 777
705 last; # more data might arrive 778 last; # more data might arrive
706 } 779 }
707 } else { 780 } else {
709 delete $self->{_rw}; 782 delete $self->{_rw};
710 last; 783 last;
711 } 784 }
712 } 785 }
713 786
787 if ($self->{_eof}) {
788 if ($self->{on_eof}) {
714 $self->{on_eof}($self) 789 $self->{on_eof}($self)
715 if $self->{_eof} && $self->{on_eof}; 790 } else {
791 $self->_error (0, 1);
792 }
793 }
716 794
717 # may need to restart read watcher 795 # may need to restart read watcher
718 unless ($self->{_rw}) { 796 unless ($self->{_rw}) {
719 $self->start_read 797 $self->start_read
720 if $self->{on_read} || @{ $self->{_queue} }; 798 if $self->{on_read} || @{ $self->{_queue} };
846 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 924 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
847 1 925 1
848 } 926 }
849}; 927};
850 928
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) 929=item line => [$eol, ]$cb->($handle, $line, $eol)
861 930
862The callback will be called only once a full line (including the end of 931The 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 932line 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 933marker) will be passed to the callback as second argument (C<$line>), and
879=cut 948=cut
880 949
881register_read_type line => sub { 950register_read_type line => sub {
882 my ($self, $cb, $eol) = @_; 951 my ($self, $cb, $eol) = @_;
883 952
884 $eol = qr|(\015?\012)| if @_ < 3; 953 if (@_ < 3) {
954 # this is more than twice as fast as the generic code below
955 sub {
956 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
957
958 $cb->($_[0], $1, $2);
959 1
960 }
961 } else {
885 $eol = quotemeta $eol unless ref $eol; 962 $eol = quotemeta $eol unless ref $eol;
886 $eol = qr|^(.*?)($eol)|s; 963 $eol = qr|^(.*?)($eol)|s;
887 964
888 sub { 965 sub {
889 $_[0]{rbuf} =~ s/$eol// or return; 966 $_[0]{rbuf} =~ s/$eol// or return;
890 967
891 $cb->($_[0], $1, $2); 968 $cb->($_[0], $1, $2);
969 1
892 1 970 }
893 } 971 }
894}; 972};
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 973
907=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 974=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
908 975
909Makes a regex match against the regex object C<$accept> and returns 976Makes a regex match against the regex object C<$accept> and returns
910everything up to and including the match. 977everything up to and including the match.
1031register_read_type packstring => sub { 1098register_read_type packstring => sub {
1032 my ($self, $cb, $format) = @_; 1099 my ($self, $cb, $format) = @_;
1033 1100
1034 sub { 1101 sub {
1035 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1102 # 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} }) 1103 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1037 or return; 1104 or return;
1038 1105
1106 $format = length pack $format, $len;
1107
1108 # bypass unshift if we already have the remaining chunk
1109 if ($format + $len <= length $_[0]{rbuf}) {
1110 my $data = substr $_[0]{rbuf}, $format, $len;
1111 substr $_[0]{rbuf}, 0, $format + $len, "";
1112 $cb->($_[0], $data);
1113 } else {
1039 # remove prefix 1114 # remove prefix
1040 substr $_[0]->{rbuf}, 0, (length pack $format, $len), ""; 1115 substr $_[0]{rbuf}, 0, $format, "";
1041 1116
1042 # read rest 1117 # read remaining chunk
1043 $_[0]->unshift_read (chunk => $len, $cb); 1118 $_[0]->unshift_read (chunk => $len, $cb);
1119 }
1044 1120
1045 1 1121 1
1046 } 1122 }
1047}; 1123};
1048 1124
1105 1181
1106 require Storable; 1182 require Storable;
1107 1183
1108 sub { 1184 sub {
1109 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1185 # 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} }) 1186 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1111 or return; 1187 or return;
1112 1188
1189 my $format = length pack "w", $len;
1190
1191 # bypass unshift if we already have the remaining chunk
1192 if ($format + $len <= length $_[0]{rbuf}) {
1193 my $data = substr $_[0]{rbuf}, $format, $len;
1194 substr $_[0]{rbuf}, 0, $format + $len, "";
1195 $cb->($_[0], Storable::thaw ($data));
1196 } else {
1113 # remove prefix 1197 # remove prefix
1114 substr $_[0]->{rbuf}, 0, (length pack "w", $len), ""; 1198 substr $_[0]{rbuf}, 0, $format, "";
1115 1199
1116 # read rest 1200 # read remaining chunk
1117 $_[0]->unshift_read (chunk => $len, sub { 1201 $_[0]->unshift_read (chunk => $len, sub {
1118 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1202 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1119 $cb->($_[0], $ref); 1203 $cb->($_[0], $ref);
1120 } else { 1204 } else {
1121 $self->_error (&Errno::EBADMSG); 1205 $self->_error (&Errno::EBADMSG);
1206 }
1122 } 1207 });
1123 }); 1208 }
1209
1210 1
1124 } 1211 }
1125}; 1212};
1126 1213
1127=back 1214=back
1128 1215
1389=over 4 1476=over 4
1390 1477
1391=item * all constructor arguments become object members. 1478=item * all constructor arguments become object members.
1392 1479
1393At least initially, when you pass a C<tls>-argument to the constructor it 1480At 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 1481will 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). 1482mutated later on (for example C<tls> will hold the TLS connection object).
1396 1483
1397=item * other object member names are prefixed with an C<_>. 1484=item * other object member names are prefixed with an C<_>.
1398 1485
1399All object members not explicitly documented (internal use) are prefixed 1486All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines