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.214 by root, Sun Jan 16 17:12:27 2011 UTC vs.
Revision 1.228 by root, Mon Feb 6 00:17:26 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 => "got error $msg\n";
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
128=item on_connect => $cb->($handle, $host, $port, $retry->()) 128=item on_connect => $cb->($handle, $host, $port, $retry->())
129 129
130This callback is called when a connection has been successfully established. 130This callback is called when a connection has been successfully established.
131 131
132The 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
133parameters, 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.
134 136
137It is not allowed to use the read or write queues while the handle object
138is connecting.
139
135If, for some reason, the handle is not acceptable, calling C<$retry> 140If, for some reason, the handle is not acceptable, calling C<$retry> will
136will continue with the next connection target (in case of multi-homed 141continue with the next connection target (in case of multi-homed hosts or
137hosts or SRV records there can be multiple connection endpoints). At the 142SRV records there can be multiple connection endpoints). The C<$retry>
138time 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
139similar properties of the handle will have been reset. 144start a handshake and then decide to retry with the next host if the
145handshake fails.
140 146
141In most cases, you should ignore the C<$retry> parameter. 147In most cases, you should ignore the C<$retry> parameter.
142 148
143=item on_connect_error => $cb->($handle, $message) 149=item on_connect_error => $cb->($handle, $message)
144 150
247many seconds pass without a successful read or write on the underlying 253many seconds pass without a successful read or write on the underlying
248file handle (or a call to C<timeout_reset>), the C<on_timeout> callback 254file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
249will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT> 255will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
250error will be raised). 256error will be raised).
251 257
252There are three variants of the timeouts that work independently 258There are three variants of the timeouts that work independently of each
253of each other, for both read and write, just read, and just write: 259other, for both read and write (triggered when nothing was read I<OR>
260written), just read (triggered when nothing was read), and just write:
254C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks 261C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
255C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions 262C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
256C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>. 263C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
257 264
258Note that timeout processing is active even when you do not have 265Note that timeout processing is active even when you do not have any
259any outstanding read or write requests: If you plan to keep the connection 266outstanding read or write requests: If you plan to keep the connection
260idle then you should disable the timeout temporarily or ignore the timeout 267idle then you should disable the timeout temporarily or ignore the
261in the C<on_timeout> callback, in which case AnyEvent::Handle will simply 268timeout in the corresponding C<on_timeout> callback, in which case
262restart the timeout. 269AnyEvent::Handle will simply restart the timeout.
263 270
264Zero (the default) disables this timeout. 271Zero (the default) disables the corresponding timeout.
265 272
266=item on_timeout => $cb->($handle) 273=item on_timeout => $cb->($handle)
274
275=item on_rtimeout => $cb->($handle)
276
277=item on_wtimeout => $cb->($handle)
267 278
268Called whenever the inactivity timeout passes. If you return from this 279Called whenever the inactivity timeout passes. If you return from this
269callback, then the timeout will be reset as if some activity had happened, 280callback, then the timeout will be reset as if some activity had happened,
270so this condition is not fatal in any way. 281so this condition is not fatal in any way.
271 282
354already have occured on BSD systems), but at least it will protect you 365already have occured on BSD systems), but at least it will protect you
355from most attacks. 366from most attacks.
356 367
357=item read_size => <bytes> 368=item read_size => <bytes>
358 369
359The initial read block size, the number of bytes this module will try to 370The initial read block size, the number of bytes this module will try
360read during each loop iteration. Each handle object will consume at least 371to read during each loop iteration. Each handle object will consume
361this amount of memory for the read buffer as well, so when handling many 372at least this amount of memory for the read buffer as well, so when
362connections requirements). See also C<max_read_size>. Default: C<2048>. 373handling many connections watch out for memory requirements). See also
374C<max_read_size>. Default: C<2048>.
363 375
364=item max_read_size => <bytes> 376=item max_read_size => <bytes>
365 377
366The maximum read buffer size used by the dynamic adjustment 378The maximum read buffer size used by the dynamic adjustment
367algorithm: Each time AnyEvent::Handle can read C<read_size> bytes in 379algorithm: Each time AnyEvent::Handle can read C<read_size> bytes in
536 }); 548 });
537 549
538 } else { 550 } else {
539 if ($self->{on_connect_error}) { 551 if ($self->{on_connect_error}) {
540 $self->{on_connect_error}($self, "$!"); 552 $self->{on_connect_error}($self, "$!");
541 $self->destroy; 553 $self->destroy if $self;
542 } else { 554 } else {
543 $self->_error ($!, 1); 555 $self->_error ($!, 1);
544 } 556 }
545 } 557 }
546 }, 558 },
765 777
766sub rbuf_max { 778sub rbuf_max {
767 $_[0]{rbuf_max} = $_[1]; 779 $_[0]{rbuf_max} = $_[1];
768} 780}
769 781
770sub rbuf_max { 782sub wbuf_max {
771 $_[0]{wbuf_max} = $_[1]; 783 $_[0]{wbuf_max} = $_[1];
772} 784}
773 785
774############################################################################# 786#############################################################################
775 787
778=item $handle->rtimeout ($seconds) 790=item $handle->rtimeout ($seconds)
779 791
780=item $handle->wtimeout ($seconds) 792=item $handle->wtimeout ($seconds)
781 793
782Configures (or disables) the inactivity timeout. 794Configures (or disables) the inactivity timeout.
795
796The timeout will be checked instantly, so this method might destroy the
797handle before it returns.
783 798
784=item $handle->timeout_reset 799=item $handle->timeout_reset
785 800
786=item $handle->rtimeout_reset 801=item $handle->rtimeout_reset
787 802
1072=cut 1087=cut
1073 1088
1074register_write_type storable => sub { 1089register_write_type storable => sub {
1075 my ($self, $ref) = @_; 1090 my ($self, $ref) = @_;
1076 1091
1077 require Storable; 1092 require Storable unless $Storable::VERSION;
1078 1093
1079 pack "w/a*", Storable::nfreeze ($ref) 1094 pack "w/a*", Storable::nfreeze ($ref)
1080}; 1095};
1081 1096
1082=back 1097=back
1119 1134
1120Whenever the given C<type> is used, C<push_write> will the function with 1135Whenever the given C<type> is used, C<push_write> will the function with
1121the handle object and the remaining arguments. 1136the handle object and the remaining arguments.
1122 1137
1123The function is supposed to return a single octet string that will be 1138The function is supposed to return a single octet string that will be
1124appended to the write buffer, so you cna mentally treat this function as a 1139appended to the write buffer, so you can mentally treat this function as a
1125"arguments to on-the-wire-format" converter. 1140"arguments to on-the-wire-format" converter.
1126 1141
1127Example: implement a custom write type C<join> that joins the remaining 1142Example: implement a custom write type C<join> that joins the remaining
1128arguments using the first one. 1143arguments using the first one.
1129 1144
1423data. 1438data.
1424 1439
1425Example: read 2 bytes. 1440Example: read 2 bytes.
1426 1441
1427 $handle->push_read (chunk => 2, sub { 1442 $handle->push_read (chunk => 2, sub {
1428 warn "yay ", unpack "H*", $_[1]; 1443 say "yay " . unpack "H*", $_[1];
1429 }); 1444 });
1430 1445
1431=cut 1446=cut
1432 1447
1433register_read_type chunk => sub { 1448register_read_type chunk => sub {
1467 if (@_ < 3) { 1482 if (@_ < 3) {
1468 # this is more than twice as fast as the generic code below 1483 # this is more than twice as fast as the generic code below
1469 sub { 1484 sub {
1470 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return; 1485 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
1471 1486
1472 $cb->($_[0], $1, $2); 1487 $cb->($_[0], "$1", "$2");
1473 1 1488 1
1474 } 1489 }
1475 } else { 1490 } else {
1476 $eol = quotemeta $eol unless ref $eol; 1491 $eol = quotemeta $eol unless ref $eol;
1477 $eol = qr|^(.*?)($eol)|s; 1492 $eol = qr|^(.*?)($eol)|s;
1478 1493
1479 sub { 1494 sub {
1480 $_[0]{rbuf} =~ s/$eol// or return; 1495 $_[0]{rbuf} =~ s/$eol// or return;
1481 1496
1482 $cb->($_[0], $1, $2); 1497 $cb->($_[0], "$1", "$2");
1483 1 1498 1
1484 } 1499 }
1485 } 1500 }
1486}; 1501};
1487 1502
1535 1550
1536 sub { 1551 sub {
1537 # accept 1552 # accept
1538 if ($$rbuf =~ $accept) { 1553 if ($$rbuf =~ $accept) {
1539 $data .= substr $$rbuf, 0, $+[0], ""; 1554 $data .= substr $$rbuf, 0, $+[0], "";
1540 $cb->($self, $data); 1555 $cb->($_[0], $data);
1541 return 1; 1556 return 1;
1542 } 1557 }
1543 1558
1544 # reject 1559 # reject
1545 if ($reject && $$rbuf =~ $reject) { 1560 if ($reject && $$rbuf =~ $reject) {
1546 $self->_error (Errno::EBADMSG); 1561 $_[0]->_error (Errno::EBADMSG);
1547 } 1562 }
1548 1563
1549 # skip 1564 # skip
1550 if ($skip && $$rbuf =~ $skip) { 1565 if ($skip && $$rbuf =~ $skip) {
1551 $data .= substr $$rbuf, 0, $+[0], ""; 1566 $data .= substr $$rbuf, 0, $+[0], "";
1567 my ($self, $cb) = @_; 1582 my ($self, $cb) = @_;
1568 1583
1569 sub { 1584 sub {
1570 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 1585 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1571 if ($_[0]{rbuf} =~ /[^0-9]/) { 1586 if ($_[0]{rbuf} =~ /[^0-9]/) {
1572 $self->_error (Errno::EBADMSG); 1587 $_[0]->_error (Errno::EBADMSG);
1573 } 1588 }
1574 return; 1589 return;
1575 } 1590 }
1576 1591
1577 my $len = $1; 1592 my $len = $1;
1578 1593
1579 $self->unshift_read (chunk => $len, sub { 1594 $_[0]->unshift_read (chunk => $len, sub {
1580 my $string = $_[1]; 1595 my $string = $_[1];
1581 $_[0]->unshift_read (chunk => 1, sub { 1596 $_[0]->unshift_read (chunk => 1, sub {
1582 if ($_[1] eq ",") { 1597 if ($_[1] eq ",") {
1583 $cb->($_[0], $string); 1598 $cb->($_[0], $string);
1584 } else { 1599 } else {
1585 $self->_error (Errno::EBADMSG); 1600 $_[0]->_error (Errno::EBADMSG);
1586 } 1601 }
1587 }); 1602 });
1588 }); 1603 });
1589 1604
1590 1 1605 1
1663 1678
1664 my $data; 1679 my $data;
1665 my $rbuf = \$self->{rbuf}; 1680 my $rbuf = \$self->{rbuf};
1666 1681
1667 sub { 1682 sub {
1668 my $ref = eval { $json->incr_parse ($self->{rbuf}) }; 1683 my $ref = eval { $json->incr_parse ($_[0]{rbuf}) };
1669 1684
1670 if ($ref) { 1685 if ($ref) {
1671 $self->{rbuf} = $json->incr_text; 1686 $_[0]{rbuf} = $json->incr_text;
1672 $json->incr_text = ""; 1687 $json->incr_text = "";
1673 $cb->($self, $ref); 1688 $cb->($_[0], $ref);
1674 1689
1675 1 1690 1
1676 } elsif ($@) { 1691 } elsif ($@) {
1677 # error case 1692 # error case
1678 $json->incr_skip; 1693 $json->incr_skip;
1679 1694
1680 $self->{rbuf} = $json->incr_text; 1695 $_[0]{rbuf} = $json->incr_text;
1681 $json->incr_text = ""; 1696 $json->incr_text = "";
1682 1697
1683 $self->_error (Errno::EBADMSG); 1698 $_[0]->_error (Errno::EBADMSG);
1684 1699
1685 () 1700 ()
1686 } else { 1701 } else {
1687 $self->{rbuf} = ""; 1702 $_[0]{rbuf} = "";
1688 1703
1689 () 1704 ()
1690 } 1705 }
1691 } 1706 }
1692}; 1707};
1702=cut 1717=cut
1703 1718
1704register_read_type storable => sub { 1719register_read_type storable => sub {
1705 my ($self, $cb) = @_; 1720 my ($self, $cb) = @_;
1706 1721
1707 require Storable; 1722 require Storable unless $Storable::VERSION;
1708 1723
1709 sub { 1724 sub {
1710 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1725 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1711 defined (my $len = eval { unpack "w", $_[0]{rbuf} }) 1726 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1712 or return; 1727 or return;
1725 # read remaining chunk 1740 # read remaining chunk
1726 $_[0]->unshift_read (chunk => $len, sub { 1741 $_[0]->unshift_read (chunk => $len, sub {
1727 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1742 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1728 $cb->($_[0], $ref); 1743 $cb->($_[0], $ref);
1729 } else { 1744 } else {
1730 $self->_error (Errno::EBADMSG); 1745 $_[0]->_error (Errno::EBADMSG);
1731 } 1746 }
1732 }); 1747 });
1733 } 1748 }
1734 1749
1735 1 1750 1
1783some readings of the the SSL/TLS specifications basically require this 1798some readings of the the SSL/TLS specifications basically require this
1784attack to be working, as SSL/TLS implementations might stall sending data 1799attack to be working, as SSL/TLS implementations might stall sending data
1785during a rehandshake. 1800during a rehandshake.
1786 1801
1787As a guideline, during the initial handshake, you should not stop reading, 1802As a guideline, during the initial handshake, you should not stop reading,
1788and as a client, it might cause problems, depending on your applciation. 1803and as a client, it might cause problems, depending on your application.
1789 1804
1790=cut 1805=cut
1791 1806
1792sub stop_read { 1807sub stop_read {
1793 my ($self) = @_; 1808 my ($self) = @_;
2002 Net::SSLeay::CTX_set_mode ($tls, 1|2); 2017 Net::SSLeay::CTX_set_mode ($tls, 1|2);
2003 2018
2004 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 2019 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
2005 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 2020 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
2006 2021
2007 Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf}); 2022 Net::SSLeay::BIO_write ($self->{_rbio}, $self->{rbuf});
2023 $self->{rbuf} = "";
2008 2024
2009 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio}); 2025 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
2010 2026
2011 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) } 2027 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
2012 if $self->{on_starttls}; 2028 if $self->{on_starttls};
2049 $self->{tls_ctx}->_put_session (delete $self->{tls}) 2065 $self->{tls_ctx}->_put_session (delete $self->{tls})
2050 if $self->{tls} > 0; 2066 if $self->{tls} > 0;
2051 2067
2052 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 2068 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
2053} 2069}
2070
2071=item $handle->resettls
2072
2073This rarely-used method simply resets and TLS state on the handle, usually
2074causing data loss.
2075
2076One case where it may be useful is when you want to skip over the data in
2077the stream but you are not interested in interpreting it, so data loss is
2078no concern.
2079
2080=cut
2081
2082*resettls = \&_freetls;
2054 2083
2055sub DESTROY { 2084sub DESTROY {
2056 my ($self) = @_; 2085 my ($self) = @_;
2057 2086
2058 &_freetls; 2087 &_freetls;
2181Probably because your C<on_error> callback is being called instead: When 2210Probably because your C<on_error> callback is being called instead: When
2182you have outstanding requests in your read queue, then an EOF is 2211you have outstanding requests in your read queue, then an EOF is
2183considered an error as you clearly expected some data. 2212considered an error as you clearly expected some data.
2184 2213
2185To avoid this, make sure you have an empty read queue whenever your handle 2214To avoid this, make sure you have an empty read queue whenever your handle
2186is supposed to be "idle" (i.e. connection closes are O.K.). You cna set 2215is supposed to be "idle" (i.e. connection closes are O.K.). You can set
2187an C<on_read> handler that simply pushes the first read requests in the 2216an C<on_read> handler that simply pushes the first read requests in the
2188queue. 2217queue.
2189 2218
2190See also the next question, which explains this in a bit more detail. 2219See also the next question, which explains this in a bit more detail.
2191 2220
2222some data and raises the C<EPIPE> error when the connction is dropped 2251some data and raises the C<EPIPE> error when the connction is dropped
2223unexpectedly. 2252unexpectedly.
2224 2253
2225The second variant is a protocol where the client can drop the connection 2254The second variant is a protocol where the client can drop the connection
2226at any time. For TCP, this means that the server machine may run out of 2255at any time. For TCP, this means that the server machine may run out of
2227sockets easier, and in general, it means you cnanot distinguish a protocl 2256sockets easier, and in general, it means you cannot distinguish a protocl
2228failure/client crash from a normal connection close. Nevertheless, these 2257failure/client crash from a normal connection close. Nevertheless, these
2229kinds of protocols are common (and sometimes even the best solution to the 2258kinds of protocols are common (and sometimes even the best solution to the
2230problem). 2259problem).
2231 2260
2232Having an outstanding read request at all times is possible if you ignore 2261Having an outstanding read request at all times is possible if you ignore
2284 $handle->on_eof (undef); 2313 $handle->on_eof (undef);
2285 $handle->on_error (sub { 2314 $handle->on_error (sub {
2286 my $data = delete $_[0]{rbuf}; 2315 my $data = delete $_[0]{rbuf};
2287 }); 2316 });
2288 2317
2318Note that this example removes the C<rbuf> member from the handle object,
2319which is not normally allowed by the API. It is expressly permitted in
2320this case only, as the handle object needs to be destroyed afterwards.
2321
2289The reason to use C<on_error> is that TCP connections, due to latencies 2322The reason to use C<on_error> is that TCP connections, due to latencies
2290and packets loss, might get closed quite violently with an error, when in 2323and packets loss, might get closed quite violently with an error, when in
2291fact all data has been received. 2324fact all data has been received.
2292 2325
2293It is usually better to use acknowledgements when transferring data, 2326It is usually better to use acknowledgements when transferring data,
2303C<low_water_mark> this will be called precisely when all data has been 2336C<low_water_mark> this will be called precisely when all data has been
2304written to the socket: 2337written to the socket:
2305 2338
2306 $handle->push_write (...); 2339 $handle->push_write (...);
2307 $handle->on_drain (sub { 2340 $handle->on_drain (sub {
2308 warn "all data submitted to the kernel\n"; 2341 AE::log debug => "all data submitted to the kernel\n";
2309 undef $handle; 2342 undef $handle;
2310 }); 2343 });
2311 2344
2312If you just want to queue some data and then signal EOF to the other side, 2345If you just want to queue some data and then signal EOF to the other side,
2313consider using C<< ->push_shutdown >> instead. 2346consider using C<< ->push_shutdown >> instead.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines