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.201 by root, Wed Oct 13 01:15:57 2010 UTC vs.
Revision 1.234 by root, Wed Apr 18 09:44:10 2012 UTC

11 11
12 my $hdl; $hdl = new AnyEvent::Handle 12 my $hdl; $hdl = new AnyEvent::Handle
13 fh => \*STDIN, 13 fh => \*STDIN,
14 on_error => sub { 14 on_error => sub {
15 my ($hdl, $fatal, $msg) = @_; 15 my ($hdl, $fatal, $msg) = @_;
16 warn "got error $msg\n"; 16 AE::log error => $msg;
17 $hdl->destroy; 17 $hdl->destroy;
18 $cv->send; 18 $cv->send;
19 }; 19 };
20 20
21 # send some request line 21 # send some request line
22 $hdl->push_write ("getinfo\015\012"); 22 $hdl->push_write ("getinfo\015\012");
23 23
24 # read the response line 24 # read the response line
25 $hdl->push_read (line => sub { 25 $hdl->push_read (line => sub {
26 my ($hdl, $line) = @_; 26 my ($hdl, $line) = @_;
27 warn "got line <$line>\n"; 27 say "got line <$line>";
28 $cv->send; 28 $cv->send;
29 }); 29 });
30 30
31 $cv->recv; 31 $cv->recv;
32 32
75 } 75 }
76 76
77 \&$func 77 \&$func
78} 78}
79 79
80sub MAX_READ_SIZE() { 131072 }
81
80=head1 METHODS 82=head1 METHODS
81 83
82=over 4 84=over 4
83 85
84=item $handle = B<new> AnyEvent::Handle fh => $filehandle, key => value... 86=item $handle = B<new> AnyEvent::Handle fh => $filehandle, key => value...
112=over 4 114=over 4
113 115
114=item on_prepare => $cb->($handle) 116=item on_prepare => $cb->($handle)
115 117
116This (rarely used) callback is called before a new connection is 118This (rarely used) callback is called before a new connection is
117attempted, but after the file handle has been created. It could be used to 119attempted, but after the file handle has been created (you can access that
120file handle via C<< $handle->{fh} >>). It could be used to prepare the
118prepare the file handle with parameters required for the actual connect 121file handle with parameters required for the actual connect (as opposed to
119(as opposed to settings that can be changed when the connection is already 122settings that can be changed when the connection is already established).
120established).
121 123
122The return value of this callback should be the connect timeout value in 124The return value of this callback should be the connect timeout value in
123seconds (or C<0>, or C<undef>, or the empty list, to indicate that the 125seconds (or C<0>, or C<undef>, or the empty list, to indicate that the
124default timeout is to be used). 126default timeout is to be used).
125 127
126=item on_connect => $cb->($handle, $host, $port, $retry->()) 128=item on_connect => $cb->($handle, $host, $port, $retry->())
127 129
128This callback is called when a connection has been successfully established. 130This callback is called when a connection has been successfully established.
129 131
130The peer's numeric host and port (the socket peername) are passed as 132The peer's numeric host and port (the socket peername) are passed as
131parameters, together with a retry callback. 133parameters, together with a retry callback. At the time it is called the
134read and write queues, EOF status, TLS status and similar properties of
135the handle will have been reset.
132 136
137It is not allowed to use the read or write queues while the handle object
138is connecting.
139
133If, for some reason, the handle is not acceptable, calling C<$retry> 140If, for some reason, the handle is not acceptable, calling C<$retry> will
134will continue with the next connection target (in case of multi-homed 141continue with the next connection target (in case of multi-homed hosts or
135hosts or SRV records there can be multiple connection endpoints). At the 142SRV records there can be multiple connection endpoints). The C<$retry>
136time it is called the read and write queues, eof status, tls status and 143callback can be invoked after the connect callback returns, i.e. one can
137similar properties of the handle will have been reset. 144start a handshake and then decide to retry with the next host if the
145handshake fails.
138 146
139In most cases, you should ignore the C<$retry> parameter. 147In most cases, you should ignore the C<$retry> parameter.
140 148
141=item on_connect_error => $cb->($handle, $message) 149=item on_connect_error => $cb->($handle, $message)
142 150
157 165
158Some errors are fatal (which is indicated by C<$fatal> being true). On 166Some errors are fatal (which is indicated by C<$fatal> being true). On
159fatal errors the handle object will be destroyed (by a call to C<< -> 167fatal errors the handle object will be destroyed (by a call to C<< ->
160destroy >>) after invoking the error callback (which means you are free to 168destroy >>) after invoking the error callback (which means you are free to
161examine the handle object). Examples of fatal errors are an EOF condition 169examine the handle object). Examples of fatal errors are an EOF condition
162with active (but unsatisifable) read watchers (C<EPIPE>) or I/O errors. In 170with active (but unsatisfiable) read watchers (C<EPIPE>) or I/O errors. In
163cases where the other side can close the connection at will, it is 171cases where the other side can close the connection at will, it is
164often easiest to not report C<EPIPE> errors in this callback. 172often easiest to not report C<EPIPE> errors in this callback.
165 173
166AnyEvent::Handle tries to find an appropriate error code for you to check 174AnyEvent::Handle tries to find an appropriate error code for you to check
167against, but in some cases (TLS errors), this does not work well. It is 175against, but in some cases (TLS errors), this does not work well.
168recommended to always output the C<$message> argument in human-readable 176
169error messages (it's usually the same as C<"$!">). 177If you report the error to the user, it is recommended to always output
178the C<$message> argument in human-readable error messages (you don't need
179to report C<"$!"> if you report C<$message>).
180
181If you want to react programmatically to the error, then looking at C<$!>
182and comparing it against some of the documented C<Errno> values is usually
183better than looking at the C<$message>.
170 184
171Non-fatal errors can be retried by returning, but it is recommended 185Non-fatal errors can be retried by returning, but it is recommended
172to simply ignore this parameter and instead abondon the handle object 186to simply ignore this parameter and instead abondon the handle object
173when this callback is invoked. Examples of non-fatal errors are timeouts 187when this callback is invoked. Examples of non-fatal errors are timeouts
174C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>). 188C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
222If an EOF condition has been detected but no C<on_eof> callback has been 236If an EOF condition has been detected but no C<on_eof> callback has been
223set, then a fatal error will be raised with C<$!> set to <0>. 237set, then a fatal error will be raised with C<$!> set to <0>.
224 238
225=item on_drain => $cb->($handle) 239=item on_drain => $cb->($handle)
226 240
227This sets the callback that is called when the write buffer becomes empty 241This sets the callback that is called once when the write buffer becomes
228(or immediately if the buffer is empty already). 242empty (and immediately when the handle object is created).
229 243
230To append to the write buffer, use the C<< ->push_write >> method. 244To append to the write buffer, use the C<< ->push_write >> method.
231 245
232This callback is useful when you don't want to put all of your write data 246This callback is useful when you don't want to put all of your write data
233into the queue at once, for example, when you want to write the contents 247into the queue at once, for example, when you want to write the contents
245many seconds pass without a successful read or write on the underlying 259many seconds pass without a successful read or write on the underlying
246file handle (or a call to C<timeout_reset>), the C<on_timeout> callback 260file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
247will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT> 261will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
248error will be raised). 262error will be raised).
249 263
250There are three variants of the timeouts that work independently 264There are three variants of the timeouts that work independently of each
251of each other, for both read and write, just read, and just write: 265other, for both read and write (triggered when nothing was read I<OR>
266written), just read (triggered when nothing was read), and just write:
252C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks 267C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
253C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions 268C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
254C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>. 269C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
255 270
256Note that timeout processing is active even when you do not have 271Note that timeout processing is active even when you do not have any
257any outstanding read or write requests: If you plan to keep the connection 272outstanding read or write requests: If you plan to keep the connection
258idle then you should disable the timeout temporarily or ignore the timeout 273idle then you should disable the timeout temporarily or ignore the
259in the C<on_timeout> callback, in which case AnyEvent::Handle will simply 274timeout in the corresponding C<on_timeout> callback, in which case
260restart the timeout. 275AnyEvent::Handle will simply restart the timeout.
261 276
262Zero (the default) disables this timeout. 277Zero (the default) disables the corresponding timeout.
263 278
264=item on_timeout => $cb->($handle) 279=item on_timeout => $cb->($handle)
280
281=item on_rtimeout => $cb->($handle)
282
283=item on_wtimeout => $cb->($handle)
265 284
266Called whenever the inactivity timeout passes. If you return from this 285Called whenever the inactivity timeout passes. If you return from this
267callback, then the timeout will be reset as if some activity had happened, 286callback, then the timeout will be reset as if some activity had happened,
268so this condition is not fatal in any way. 287so this condition is not fatal in any way.
269 288
276For example, a server accepting connections from untrusted sources should 295For example, a server accepting connections from untrusted sources should
277be configured to accept only so-and-so much data that it cannot act on 296be configured to accept only so-and-so much data that it cannot act on
278(for example, when expecting a line, an attacker could send an unlimited 297(for example, when expecting a line, an attacker could send an unlimited
279amount of data without a callback ever being called as long as the line 298amount of data without a callback ever being called as long as the line
280isn't finished). 299isn't finished).
300
301=item wbuf_max => <bytes>
302
303If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
304when the write buffer ever (strictly) exceeds this size. This is useful to
305avoid some forms of denial-of-service attacks.
306
307Although the units of this parameter is bytes, this is the I<raw> number
308of bytes not yet accepted by the kernel. This can make a difference when
309you e.g. use TLS, as TLS typically makes your write data larger (but it
310can also make it smaller due to compression).
311
312As an example of when this limit is useful, take a chat server that sends
313chat messages to a client. If the client does not read those in a timely
314manner then the send buffer in the server would grow unbounded.
281 315
282=item autocork => <boolean> 316=item autocork => <boolean>
283 317
284When disabled (the default), C<push_write> will try to immediately 318When disabled (the default), C<push_write> will try to immediately
285write the data to the handle if possible. This avoids having to register 319write the data to the handle if possible. This avoids having to register
337already have occured on BSD systems), but at least it will protect you 371already have occured on BSD systems), but at least it will protect you
338from most attacks. 372from most attacks.
339 373
340=item read_size => <bytes> 374=item read_size => <bytes>
341 375
342The default read block size (the number of bytes this module will 376The initial read block size, the number of bytes this module will try
343try to read during each loop iteration, which affects memory 377to read during each loop iteration. Each handle object will consume
344requirements). Default: C<8192>. 378at least this amount of memory for the read buffer as well, so when
379handling many connections watch out for memory requirements). See also
380C<max_read_size>. Default: C<2048>.
381
382=item max_read_size => <bytes>
383
384The maximum read buffer size used by the dynamic adjustment
385algorithm: Each time AnyEvent::Handle can read C<read_size> bytes in
386one go it will double C<read_size> up to the maximum given by this
387option. Default: C<131072> or C<read_size>, whichever is higher.
345 388
346=item low_water_mark => <bytes> 389=item low_water_mark => <bytes>
347 390
348Sets the number of bytes (default: C<0>) that make up an "empty" write 391Sets the number of bytes (default: C<0>) that make up an "empty" write
349buffer: If the buffer reaches this size or gets even samller it is 392buffer: If the buffer reaches this size or gets even samller it is
386appropriate error message. 429appropriate error message.
387 430
388TLS mode requires Net::SSLeay to be installed (it will be loaded 431TLS mode requires Net::SSLeay to be installed (it will be loaded
389automatically when you try to create a TLS handle): this module doesn't 432automatically when you try to create a TLS handle): this module doesn't
390have a dependency on that module, so if your module requires it, you have 433have a dependency on that module, so if your module requires it, you have
391to add the dependency yourself. 434to add the dependency yourself. If Net::SSLeay cannot be loaded or is too
435old, you get an C<EPROTO> error.
392 436
393Unlike TCP, TLS has a server and client side: for the TLS server side, use 437Unlike TCP, TLS has a server and client side: for the TLS server side, use
394C<accept>, and for the TLS client side of a connection, use C<connect> 438C<accept>, and for the TLS client side of a connection, use C<connect>
395mode. 439mode.
396 440
412Use the C<< ->starttls >> method if you need to start TLS negotiation later. 456Use the C<< ->starttls >> method if you need to start TLS negotiation later.
413 457
414=item tls_ctx => $anyevent_tls 458=item tls_ctx => $anyevent_tls
415 459
416Use the given C<AnyEvent::TLS> object to create the new TLS connection 460Use the given C<AnyEvent::TLS> object to create the new TLS connection
417(unless a connection object was specified directly). If this parameter is 461(unless a connection object was specified directly). If this
418missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 462parameter is missing (or C<undef>), then AnyEvent::Handle will use
463C<AnyEvent::Handle::TLS_CTX>.
419 464
420Instead of an object, you can also specify a hash reference with C<< key 465Instead of an object, you can also specify a hash reference with C<< key
421=> value >> pairs. Those will be passed to L<AnyEvent::TLS> to create a 466=> value >> pairs. Those will be passed to L<AnyEvent::TLS> to create a
422new TLS context object. 467new TLS context object.
423 468
492 $self->{connect}[0], 537 $self->{connect}[0],
493 $self->{connect}[1], 538 $self->{connect}[1],
494 sub { 539 sub {
495 my ($fh, $host, $port, $retry) = @_; 540 my ($fh, $host, $port, $retry) = @_;
496 541
542 delete $self->{_connect}; # no longer needed
543
497 if ($fh) { 544 if ($fh) {
498 $self->{fh} = $fh; 545 $self->{fh} = $fh;
499 546
500 delete $self->{_skip_drain_rbuf}; 547 delete $self->{_skip_drain_rbuf};
501 $self->_start; 548 $self->_start;
508 }); 555 });
509 556
510 } else { 557 } else {
511 if ($self->{on_connect_error}) { 558 if ($self->{on_connect_error}) {
512 $self->{on_connect_error}($self, "$!"); 559 $self->{on_connect_error}($self, "$!");
513 $self->destroy; 560 $self->destroy if $self;
514 } else { 561 } else {
515 $self->_error ($!, 1); 562 $self->_error ($!, 1);
516 } 563 }
517 } 564 }
518 }, 565 },
519 sub { 566 sub {
520 local $self->{fh} = $_[0]; 567 local $self->{fh} = $_[0];
521 568
522 $self->{on_prepare} 569 $self->{on_prepare}
523 ? $self->{on_prepare}->($self) 570 ? $self->{on_prepare}->($self)
524 : () 571 : ()
525 } 572 }
526 ); 573 );
527 } 574 }
528 575
545 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 592 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
546 593
547 $self->{_activity} = 594 $self->{_activity} =
548 $self->{_ractivity} = 595 $self->{_ractivity} =
549 $self->{_wactivity} = AE::now; 596 $self->{_wactivity} = AE::now;
597
598 $self->{read_size} ||= 2048;
599 $self->{max_read_size} = $self->{read_size}
600 if $self->{read_size} > ($self->{max_read_size} || MAX_READ_SIZE);
550 601
551 $self->timeout (delete $self->{timeout} ) if $self->{timeout}; 602 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
552 $self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout}; 603 $self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout};
553 $self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout}; 604 $self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout};
554 605
723 774
724=item $handle->rbuf_max ($max_octets) 775=item $handle->rbuf_max ($max_octets)
725 776
726Configures the C<rbuf_max> setting (C<undef> disables it). 777Configures the C<rbuf_max> setting (C<undef> disables it).
727 778
779=item $handle->wbuf_max ($max_octets)
780
781Configures the C<wbuf_max> setting (C<undef> disables it).
782
728=cut 783=cut
729 784
730sub rbuf_max { 785sub rbuf_max {
731 $_[0]{rbuf_max} = $_[1]; 786 $_[0]{rbuf_max} = $_[1];
732} 787}
733 788
789sub wbuf_max {
790 $_[0]{wbuf_max} = $_[1];
791}
792
734############################################################################# 793#############################################################################
735 794
736=item $handle->timeout ($seconds) 795=item $handle->timeout ($seconds)
737 796
738=item $handle->rtimeout ($seconds) 797=item $handle->rtimeout ($seconds)
739 798
740=item $handle->wtimeout ($seconds) 799=item $handle->wtimeout ($seconds)
741 800
742Configures (or disables) the inactivity timeout. 801Configures (or disables) the inactivity timeout.
802
803The timeout will be checked instantly, so this method might destroy the
804handle before it returns.
743 805
744=item $handle->timeout_reset 806=item $handle->timeout_reset
745 807
746=item $handle->rtimeout_reset 808=item $handle->rtimeout_reset
747 809
831 893
832The write queue is very simple: you can add data to its end, and 894The write queue is very simple: you can add data to its end, and
833AnyEvent::Handle will automatically try to get rid of it for you. 895AnyEvent::Handle will automatically try to get rid of it for you.
834 896
835When data could be written and the write buffer is shorter then the low 897When data could be written and the write buffer is shorter then the low
836water mark, the C<on_drain> callback will be invoked. 898water mark, the C<on_drain> callback will be invoked once.
837 899
838=over 4 900=over 4
839 901
840=item $handle->on_drain ($cb) 902=item $handle->on_drain ($cb)
841 903
856 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}); 918 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
857} 919}
858 920
859=item $handle->push_write ($data) 921=item $handle->push_write ($data)
860 922
861Queues the given scalar to be written. You can push as much data as you 923Queues the given scalar to be written. You can push as much data as
862want (only limited by the available memory), as C<AnyEvent::Handle> 924you want (only limited by the available memory and C<wbuf_max>), as
863buffers it independently of the kernel. 925C<AnyEvent::Handle> buffers it independently of the kernel.
864 926
865This method may invoke callbacks (and therefore the handle might be 927This method may invoke callbacks (and therefore the handle might be
866destroyed after it returns). 928destroyed after it returns).
867 929
868=cut 930=cut
896 $cb->() unless $self->{autocork}; 958 $cb->() unless $self->{autocork};
897 959
898 # if still data left in wbuf, we need to poll 960 # if still data left in wbuf, we need to poll
899 $self->{_ww} = AE::io $self->{fh}, 1, $cb 961 $self->{_ww} = AE::io $self->{fh}, 1, $cb
900 if length $self->{wbuf}; 962 if length $self->{wbuf};
963
964 if (
965 defined $self->{wbuf_max}
966 && $self->{wbuf_max} < length $self->{wbuf}
967 ) {
968 $self->_error (Errno::ENOSPC, 1), return;
969 }
901 }; 970 };
902} 971}
903 972
904our %WH; 973our %WH;
905 974
1025=cut 1094=cut
1026 1095
1027register_write_type storable => sub { 1096register_write_type storable => sub {
1028 my ($self, $ref) = @_; 1097 my ($self, $ref) = @_;
1029 1098
1030 require Storable; 1099 require Storable unless $Storable::VERSION;
1031 1100
1032 pack "w/a*", Storable::nfreeze ($ref) 1101 pack "w/a*", Storable::nfreeze ($ref)
1033}; 1102};
1034 1103
1035=back 1104=back
1040before it was actually written. One way to do that is to replace your 1109before it was actually written. One way to do that is to replace your
1041C<on_drain> handler by a callback that shuts down the socket (and set 1110C<on_drain> handler by a callback that shuts down the socket (and set
1042C<low_water_mark> to C<0>). This method is a shorthand for just that, and 1111C<low_water_mark> to C<0>). This method is a shorthand for just that, and
1043replaces the C<on_drain> callback with: 1112replaces the C<on_drain> callback with:
1044 1113
1045 sub { shutdown $_[0]{fh}, 1 } # for push_shutdown 1114 sub { shutdown $_[0]{fh}, 1 }
1046 1115
1047This simply shuts down the write side and signals an EOF condition to the 1116This simply shuts down the write side and signals an EOF condition to the
1048the peer. 1117the peer.
1049 1118
1050You can rely on the normal read queue and C<on_eof> handling 1119You can rely on the normal read queue and C<on_eof> handling
1072 1141
1073Whenever the given C<type> is used, C<push_write> will the function with 1142Whenever the given C<type> is used, C<push_write> will the function with
1074the handle object and the remaining arguments. 1143the handle object and the remaining arguments.
1075 1144
1076The function is supposed to return a single octet string that will be 1145The function is supposed to return a single octet string that will be
1077appended to the write buffer, so you cna mentally treat this function as a 1146appended to the write buffer, so you can mentally treat this function as a
1078"arguments to on-the-wire-format" converter. 1147"arguments to on-the-wire-format" converter.
1079 1148
1080Example: implement a custom write type C<join> that joins the remaining 1149Example: implement a custom write type C<join> that joins the remaining
1081arguments using the first one. 1150arguments using the first one.
1082 1151
1376data. 1445data.
1377 1446
1378Example: read 2 bytes. 1447Example: read 2 bytes.
1379 1448
1380 $handle->push_read (chunk => 2, sub { 1449 $handle->push_read (chunk => 2, sub {
1381 warn "yay ", unpack "H*", $_[1]; 1450 say "yay " . unpack "H*", $_[1];
1382 }); 1451 });
1383 1452
1384=cut 1453=cut
1385 1454
1386register_read_type chunk => sub { 1455register_read_type chunk => sub {
1420 if (@_ < 3) { 1489 if (@_ < 3) {
1421 # this is more than twice as fast as the generic code below 1490 # this is more than twice as fast as the generic code below
1422 sub { 1491 sub {
1423 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return; 1492 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
1424 1493
1425 $cb->($_[0], $1, $2); 1494 $cb->($_[0], "$1", "$2");
1426 1 1495 1
1427 } 1496 }
1428 } else { 1497 } else {
1429 $eol = quotemeta $eol unless ref $eol; 1498 $eol = quotemeta $eol unless ref $eol;
1430 $eol = qr|^(.*?)($eol)|s; 1499 $eol = qr|^(.*?)($eol)|s;
1431 1500
1432 sub { 1501 sub {
1433 $_[0]{rbuf} =~ s/$eol// or return; 1502 $_[0]{rbuf} =~ s/$eol// or return;
1434 1503
1435 $cb->($_[0], $1, $2); 1504 $cb->($_[0], "$1", "$2");
1436 1 1505 1
1437 } 1506 }
1438 } 1507 }
1439}; 1508};
1440 1509
1488 1557
1489 sub { 1558 sub {
1490 # accept 1559 # accept
1491 if ($$rbuf =~ $accept) { 1560 if ($$rbuf =~ $accept) {
1492 $data .= substr $$rbuf, 0, $+[0], ""; 1561 $data .= substr $$rbuf, 0, $+[0], "";
1493 $cb->($self, $data); 1562 $cb->($_[0], $data);
1494 return 1; 1563 return 1;
1495 } 1564 }
1496 1565
1497 # reject 1566 # reject
1498 if ($reject && $$rbuf =~ $reject) { 1567 if ($reject && $$rbuf =~ $reject) {
1499 $self->_error (Errno::EBADMSG); 1568 $_[0]->_error (Errno::EBADMSG);
1500 } 1569 }
1501 1570
1502 # skip 1571 # skip
1503 if ($skip && $$rbuf =~ $skip) { 1572 if ($skip && $$rbuf =~ $skip) {
1504 $data .= substr $$rbuf, 0, $+[0], ""; 1573 $data .= substr $$rbuf, 0, $+[0], "";
1520 my ($self, $cb) = @_; 1589 my ($self, $cb) = @_;
1521 1590
1522 sub { 1591 sub {
1523 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 1592 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1524 if ($_[0]{rbuf} =~ /[^0-9]/) { 1593 if ($_[0]{rbuf} =~ /[^0-9]/) {
1525 $self->_error (Errno::EBADMSG); 1594 $_[0]->_error (Errno::EBADMSG);
1526 } 1595 }
1527 return; 1596 return;
1528 } 1597 }
1529 1598
1530 my $len = $1; 1599 my $len = $1;
1531 1600
1532 $self->unshift_read (chunk => $len, sub { 1601 $_[0]->unshift_read (chunk => $len, sub {
1533 my $string = $_[1]; 1602 my $string = $_[1];
1534 $_[0]->unshift_read (chunk => 1, sub { 1603 $_[0]->unshift_read (chunk => 1, sub {
1535 if ($_[1] eq ",") { 1604 if ($_[1] eq ",") {
1536 $cb->($_[0], $string); 1605 $cb->($_[0], $string);
1537 } else { 1606 } else {
1538 $self->_error (Errno::EBADMSG); 1607 $_[0]->_error (Errno::EBADMSG);
1539 } 1608 }
1540 }); 1609 });
1541 }); 1610 });
1542 1611
1543 1 1612 1
1616 1685
1617 my $data; 1686 my $data;
1618 my $rbuf = \$self->{rbuf}; 1687 my $rbuf = \$self->{rbuf};
1619 1688
1620 sub { 1689 sub {
1621 my $ref = eval { $json->incr_parse ($self->{rbuf}) }; 1690 my $ref = eval { $json->incr_parse ($_[0]{rbuf}) };
1622 1691
1623 if ($ref) { 1692 if ($ref) {
1624 $self->{rbuf} = $json->incr_text; 1693 $_[0]{rbuf} = $json->incr_text;
1625 $json->incr_text = ""; 1694 $json->incr_text = "";
1626 $cb->($self, $ref); 1695 $cb->($_[0], $ref);
1627 1696
1628 1 1697 1
1629 } elsif ($@) { 1698 } elsif ($@) {
1630 # error case 1699 # error case
1631 $json->incr_skip; 1700 $json->incr_skip;
1632 1701
1633 $self->{rbuf} = $json->incr_text; 1702 $_[0]{rbuf} = $json->incr_text;
1634 $json->incr_text = ""; 1703 $json->incr_text = "";
1635 1704
1636 $self->_error (Errno::EBADMSG); 1705 $_[0]->_error (Errno::EBADMSG);
1637 1706
1638 () 1707 ()
1639 } else { 1708 } else {
1640 $self->{rbuf} = ""; 1709 $_[0]{rbuf} = "";
1641 1710
1642 () 1711 ()
1643 } 1712 }
1644 } 1713 }
1645}; 1714};
1655=cut 1724=cut
1656 1725
1657register_read_type storable => sub { 1726register_read_type storable => sub {
1658 my ($self, $cb) = @_; 1727 my ($self, $cb) = @_;
1659 1728
1660 require Storable; 1729 require Storable unless $Storable::VERSION;
1661 1730
1662 sub { 1731 sub {
1663 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1732 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1664 defined (my $len = eval { unpack "w", $_[0]{rbuf} }) 1733 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1665 or return; 1734 or return;
1668 1737
1669 # bypass unshift if we already have the remaining chunk 1738 # bypass unshift if we already have the remaining chunk
1670 if ($format + $len <= length $_[0]{rbuf}) { 1739 if ($format + $len <= length $_[0]{rbuf}) {
1671 my $data = substr $_[0]{rbuf}, $format, $len; 1740 my $data = substr $_[0]{rbuf}, $format, $len;
1672 substr $_[0]{rbuf}, 0, $format + $len, ""; 1741 substr $_[0]{rbuf}, 0, $format + $len, "";
1742
1673 $cb->($_[0], Storable::thaw ($data)); 1743 eval { $cb->($_[0], Storable::thaw ($data)); 1 }
1744 or return $_[0]->_error (Errno::EBADMSG);
1674 } else { 1745 } else {
1675 # remove prefix 1746 # remove prefix
1676 substr $_[0]{rbuf}, 0, $format, ""; 1747 substr $_[0]{rbuf}, 0, $format, "";
1677 1748
1678 # read remaining chunk 1749 # read remaining chunk
1679 $_[0]->unshift_read (chunk => $len, sub { 1750 $_[0]->unshift_read (chunk => $len, sub {
1680 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1751 eval { $cb->($_[0], Storable::thaw ($_[1])); 1 }
1681 $cb->($_[0], $ref);
1682 } else {
1683 $self->_error (Errno::EBADMSG); 1752 or $_[0]->_error (Errno::EBADMSG);
1684 }
1685 }); 1753 });
1686 } 1754 }
1687 1755
1688 1 1756 1
1689 } 1757 }
1726Note that AnyEvent::Handle will automatically C<start_read> for you when 1794Note that AnyEvent::Handle will automatically C<start_read> for you when
1727you change the C<on_read> callback or push/unshift a read callback, and it 1795you change the C<on_read> callback or push/unshift a read callback, and it
1728will automatically C<stop_read> for you when neither C<on_read> is set nor 1796will automatically C<stop_read> for you when neither C<on_read> is set nor
1729there are any read requests in the queue. 1797there are any read requests in the queue.
1730 1798
1731These methods will have no effect when in TLS mode (as TLS doesn't support 1799In older versions of this module (<= 5.3), these methods had no effect,
1732half-duplex connections). 1800as TLS does not support half-duplex connections. In current versions they
1801work as expected, as this behaviour is required to avoid certain resource
1802attacks, where the program would be forced to read (and buffer) arbitrary
1803amounts of data before being able to send some data. The drawback is that
1804some readings of the the SSL/TLS specifications basically require this
1805attack to be working, as SSL/TLS implementations might stall sending data
1806during a rehandshake.
1807
1808As a guideline, during the initial handshake, you should not stop reading,
1809and as a client, it might cause problems, depending on your application.
1733 1810
1734=cut 1811=cut
1735 1812
1736sub stop_read { 1813sub stop_read {
1737 my ($self) = @_; 1814 my ($self) = @_;
1738 1815
1739 delete $self->{_rw} unless $self->{tls}; 1816 delete $self->{_rw};
1740} 1817}
1741 1818
1742sub start_read { 1819sub start_read {
1743 my ($self) = @_; 1820 my ($self) = @_;
1744 1821
1745 unless ($self->{_rw} || $self->{_eof} || !$self->{fh}) { 1822 unless ($self->{_rw} || $self->{_eof} || !$self->{fh}) {
1746 Scalar::Util::weaken $self; 1823 Scalar::Util::weaken $self;
1747 1824
1748 $self->{_rw} = AE::io $self->{fh}, 0, sub { 1825 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1749 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1826 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1750 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1827 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size}, length $$rbuf;
1751 1828
1752 if ($len > 0) { 1829 if ($len > 0) {
1753 $self->{_activity} = $self->{_ractivity} = AE::now; 1830 $self->{_activity} = $self->{_ractivity} = AE::now;
1754 1831
1755 if ($self->{tls}) { 1832 if ($self->{tls}) {
1758 &_dotls ($self); 1835 &_dotls ($self);
1759 } else { 1836 } else {
1760 $self->_drain_rbuf; 1837 $self->_drain_rbuf;
1761 } 1838 }
1762 1839
1840 if ($len == $self->{read_size}) {
1841 $self->{read_size} *= 2;
1842 $self->{read_size} = $self->{max_read_size} || MAX_READ_SIZE
1843 if $self->{read_size} > ($self->{max_read_size} || MAX_READ_SIZE);
1844 }
1845
1763 } elsif (defined $len) { 1846 } elsif (defined $len) {
1764 delete $self->{_rw}; 1847 delete $self->{_rw};
1765 $self->{_eof} = 1; 1848 $self->{_eof} = 1;
1766 $self->_drain_rbuf; 1849 $self->_drain_rbuf;
1767 1850
1779 my ($self, $err) = @_; 1862 my ($self, $err) = @_;
1780 1863
1781 return $self->_error ($!, 1) 1864 return $self->_error ($!, 1)
1782 if $err == Net::SSLeay::ERROR_SYSCALL (); 1865 if $err == Net::SSLeay::ERROR_SYSCALL ();
1783 1866
1784 my $err =Net::SSLeay::ERR_error_string (Net::SSLeay::ERR_get_error ()); 1867 my $err = Net::SSLeay::ERR_error_string (Net::SSLeay::ERR_get_error ());
1785 1868
1786 # reduce error string to look less scary 1869 # reduce error string to look less scary
1787 $err =~ s/^error:[0-9a-fA-F]{8}:[^:]+:([^:]+):/\L$1: /; 1870 $err =~ s/^error:[0-9a-fA-F]{8}:[^:]+:([^:]+):/\L$1: /;
1788 1871
1789 if ($self->{_on_starttls}) { 1872 if ($self->{_on_starttls}) {
1855 1938
1856=item $handle->starttls ($tls[, $tls_ctx]) 1939=item $handle->starttls ($tls[, $tls_ctx])
1857 1940
1858Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1941Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1859object is created, you can also do that at a later time by calling 1942object is created, you can also do that at a later time by calling
1860C<starttls>. 1943C<starttls>. See the C<tls> constructor argument for general info.
1861 1944
1862Starting TLS is currently an asynchronous operation - when you push some 1945Starting TLS is currently an asynchronous operation - when you push some
1863write data and then call C<< ->starttls >> then TLS negotiation will start 1946write data and then call C<< ->starttls >> then TLS negotiation will start
1864immediately, after which the queued write data is then sent. 1947immediately, after which the queued write data is then sent. This might
1948change in future versions, so best make sure you have no outstanding write
1949data when calling this method.
1865 1950
1866The first argument is the same as the C<tls> constructor argument (either 1951The first argument is the same as the C<tls> constructor argument (either
1867C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1952C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1868 1953
1869The second argument is the optional C<AnyEvent::TLS> object that is used 1954The second argument is the optional C<AnyEvent::TLS> object that is used
1891 my ($self, $tls, $ctx) = @_; 1976 my ($self, $tls, $ctx) = @_;
1892 1977
1893 Carp::croak "It is an error to call starttls on an AnyEvent::Handle object while TLS is already active, caught" 1978 Carp::croak "It is an error to call starttls on an AnyEvent::Handle object while TLS is already active, caught"
1894 if $self->{tls}; 1979 if $self->{tls};
1895 1980
1981 unless (defined $AnyEvent::TLS::VERSION) {
1982 eval {
1983 require Net::SSLeay;
1984 require AnyEvent::TLS;
1985 1
1986 } or return $self->_error (Errno::EPROTO, 1, "TLS support not available on this system");
1987 }
1988
1896 $self->{tls} = $tls; 1989 $self->{tls} = $tls;
1897 $self->{tls_ctx} = $ctx if @_ > 2; 1990 $self->{tls_ctx} = $ctx if @_ > 2;
1898 1991
1899 return unless $self->{fh}; 1992 return unless $self->{fh};
1900 1993
1901 require Net::SSLeay;
1902
1903 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1994 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1904 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1995 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1905 1996
1906 $tls = delete $self->{tls}; 1997 $tls = delete $self->{tls};
1907 $ctx = $self->{tls_ctx}; 1998 $ctx = $self->{tls_ctx};
1908 1999
1909 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 2000 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1910 2001
1911 if ("HASH" eq ref $ctx) { 2002 if ("HASH" eq ref $ctx) {
1912 require AnyEvent::TLS;
1913
1914 if ($ctx->{cache}) { 2003 if ($ctx->{cache}) {
1915 my $key = $ctx+0; 2004 my $key = $ctx+0;
1916 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx; 2005 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx;
1917 } else { 2006 } else {
1918 $ctx = new AnyEvent::TLS %$ctx; 2007 $ctx = new AnyEvent::TLS %$ctx;
1940 Net::SSLeay::CTX_set_mode ($tls, 1|2); 2029 Net::SSLeay::CTX_set_mode ($tls, 1|2);
1941 2030
1942 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 2031 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1943 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 2032 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1944 2033
1945 Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf}); 2034 Net::SSLeay::BIO_write ($self->{_rbio}, $self->{rbuf});
2035 $self->{rbuf} = "";
1946 2036
1947 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio}); 2037 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1948 2038
1949 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) } 2039 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1950 if $self->{on_starttls}; 2040 if $self->{on_starttls};
1988 if $self->{tls} > 0; 2078 if $self->{tls} > 0;
1989 2079
1990 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 2080 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1991} 2081}
1992 2082
2083=item $handle->resettls
2084
2085This rarely-used method simply resets and TLS state on the handle, usually
2086causing data loss.
2087
2088One case where it may be useful is when you want to skip over the data in
2089the stream but you are not interested in interpreting it, so data loss is
2090no concern.
2091
2092=cut
2093
2094*resettls = \&_freetls;
2095
1993sub DESTROY { 2096sub DESTROY {
1994 my ($self) = @_; 2097 my ($self) = @_;
1995 2098
1996 &_freetls; 2099 &_freetls;
1997 2100
2006 push @linger, AE::io $fh, 1, sub { 2109 push @linger, AE::io $fh, 1, sub {
2007 my $len = syswrite $fh, $wbuf, length $wbuf; 2110 my $len = syswrite $fh, $wbuf, length $wbuf;
2008 2111
2009 if ($len > 0) { 2112 if ($len > 0) {
2010 substr $wbuf, 0, $len, ""; 2113 substr $wbuf, 0, $len, "";
2011 } else { 2114 } elsif (defined $len || ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK)) {
2012 @linger = (); # end 2115 @linger = (); # end
2013 } 2116 }
2014 }; 2117 };
2015 push @linger, AE::timer $linger, 0, sub { 2118 push @linger, AE::timer $linger, 0, sub {
2016 @linger = (); 2119 @linger = ();
2112 2215
2113It is only safe to "forget" the reference inside EOF or error callbacks, 2216It is only safe to "forget" the reference inside EOF or error callbacks,
2114from within all other callbacks, you need to explicitly call the C<< 2217from within all other callbacks, you need to explicitly call the C<<
2115->destroy >> method. 2218->destroy >> method.
2116 2219
2220=item Why is my C<on_eof> callback never called?
2221
2222Probably because your C<on_error> callback is being called instead: When
2223you have outstanding requests in your read queue, then an EOF is
2224considered an error as you clearly expected some data.
2225
2226To avoid this, make sure you have an empty read queue whenever your handle
2227is supposed to be "idle" (i.e. connection closes are O.K.). You can set
2228an C<on_read> handler that simply pushes the first read requests in the
2229queue.
2230
2231See also the next question, which explains this in a bit more detail.
2232
2233=item How can I serve requests in a loop?
2234
2235Most protocols consist of some setup phase (authentication for example)
2236followed by a request handling phase, where the server waits for requests
2237and handles them, in a loop.
2238
2239There are two important variants: The first (traditional, better) variant
2240handles requests until the server gets some QUIT command, causing it to
2241close the connection first (highly desirable for a busy TCP server). A
2242client dropping the connection is an error, which means this variant can
2243detect an unexpected detection close.
2244
2245To handle this case, always make sure you have a on-empty read queue, by
2246pushing the "read request start" handler on it:
2247
2248 # we assume a request starts with a single line
2249 my @start_request; @start_request = (line => sub {
2250 my ($hdl, $line) = @_;
2251
2252 ... handle request
2253
2254 # push next request read, possibly from a nested callback
2255 $hdl->push_read (@start_request);
2256 });
2257
2258 # auth done, now go into request handling loop
2259 # now push the first @start_request
2260 $hdl->push_read (@start_request);
2261
2262By always having an outstanding C<push_read>, the handle always expects
2263some data and raises the C<EPIPE> error when the connction is dropped
2264unexpectedly.
2265
2266The second variant is a protocol where the client can drop the connection
2267at any time. For TCP, this means that the server machine may run out of
2268sockets easier, and in general, it means you cannot distinguish a protocl
2269failure/client crash from a normal connection close. Nevertheless, these
2270kinds of protocols are common (and sometimes even the best solution to the
2271problem).
2272
2273Having an outstanding read request at all times is possible if you ignore
2274C<EPIPE> errors, but this doesn't help with when the client drops the
2275connection during a request, which would still be an error.
2276
2277A better solution is to push the initial request read in an C<on_read>
2278callback. This avoids an error, as when the server doesn't expect data
2279(i.e. is idly waiting for the next request, an EOF will not raise an
2280error, but simply result in an C<on_eof> callback. It is also a bit slower
2281and simpler:
2282
2283 # auth done, now go into request handling loop
2284 $hdl->on_read (sub {
2285 my ($hdl) = @_;
2286
2287 # called each time we receive data but the read queue is empty
2288 # simply start read the request
2289
2290 $hdl->push_read (line => sub {
2291 my ($hdl, $line) = @_;
2292
2293 ... handle request
2294
2295 # do nothing special when the request has been handled, just
2296 # let the request queue go empty.
2297 });
2298 });
2299
2117=item I get different callback invocations in TLS mode/Why can't I pause 2300=item I get different callback invocations in TLS mode/Why can't I pause
2118reading? 2301reading?
2119 2302
2120Unlike, say, TCP, TLS connections do not consist of two independent 2303Unlike, say, TCP, TLS connections do not consist of two independent
2121communication channels, one for each direction. Or put differently, the 2304communication channels, one for each direction. Or put differently, the
2142 $handle->on_eof (undef); 2325 $handle->on_eof (undef);
2143 $handle->on_error (sub { 2326 $handle->on_error (sub {
2144 my $data = delete $_[0]{rbuf}; 2327 my $data = delete $_[0]{rbuf};
2145 }); 2328 });
2146 2329
2330Note that this example removes the C<rbuf> member from the handle object,
2331which is not normally allowed by the API. It is expressly permitted in
2332this case only, as the handle object needs to be destroyed afterwards.
2333
2147The reason to use C<on_error> is that TCP connections, due to latencies 2334The reason to use C<on_error> is that TCP connections, due to latencies
2148and packets loss, might get closed quite violently with an error, when in 2335and packets loss, might get closed quite violently with an error, when in
2149fact all data has been received. 2336fact all data has been received.
2150 2337
2151It is usually better to use acknowledgements when transferring data, 2338It is usually better to use acknowledgements when transferring data,
2161C<low_water_mark> this will be called precisely when all data has been 2348C<low_water_mark> this will be called precisely when all data has been
2162written to the socket: 2349written to the socket:
2163 2350
2164 $handle->push_write (...); 2351 $handle->push_write (...);
2165 $handle->on_drain (sub { 2352 $handle->on_drain (sub {
2166 warn "all data submitted to the kernel\n"; 2353 AE::log debug => "All data submitted to the kernel.";
2167 undef $handle; 2354 undef $handle;
2168 }); 2355 });
2169 2356
2170If you just want to queue some data and then signal EOF to the other side, 2357If you just want to queue some data and then signal EOF to the other side,
2171consider using C<< ->push_shutdown >> instead. 2358consider using C<< ->push_shutdown >> instead.
2255When you have intermediate CA certificates that your clients might not 2442When you have intermediate CA certificates that your clients might not
2256know about, just append them to the C<cert_file>. 2443know about, just append them to the C<cert_file>.
2257 2444
2258=back 2445=back
2259 2446
2260
2261=head1 SUBCLASSING AnyEvent::Handle 2447=head1 SUBCLASSING AnyEvent::Handle
2262 2448
2263In many cases, you might want to subclass AnyEvent::Handle. 2449In many cases, you might want to subclass AnyEvent::Handle.
2264 2450
2265To make this easier, a given version of AnyEvent::Handle uses these 2451To make this easier, a given version of AnyEvent::Handle uses these
2291 2477
2292Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 2478Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
2293 2479
2294=cut 2480=cut
2295 2481
22961; # End of AnyEvent::Handle 24821
2483

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines