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.144 by root, Mon Jul 6 21:38:25 2009 UTC vs.
Revision 1.158 by root, Fri Jul 24 08:40:35 2009 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings;
4use strict qw(subs vars);
5
6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 3use Scalar::Util ();
9use Carp (); 4use Carp ();
10use Fcntl ();
11use Errno qw(EAGAIN EINTR); 5use Errno qw(EAGAIN EINTR);
12 6
7use AnyEvent (); BEGIN { AnyEvent::common_sense }
8use AnyEvent::Util qw(WSAEWOULDBLOCK);
9
13=head1 NAME 10=head1 NAME
14 11
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 12AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 13
17=cut 14=cut
18 15
19our $VERSION = 4.452; 16our $VERSION = 4.86;
20 17
21=head1 SYNOPSIS 18=head1 SYNOPSIS
22 19
23 use AnyEvent; 20 use AnyEvent;
24 use AnyEvent::Handle; 21 use AnyEvent::Handle;
25 22
26 my $cv = AnyEvent->condvar; 23 my $cv = AnyEvent->condvar;
27 24
28 my $handle = 25 my $hdl; $hdl = new AnyEvent::Handle
29 AnyEvent::Handle->new (
30 fh => \*STDIN, 26 fh => \*STDIN,
31 on_eof => sub { 27 on_error => sub {
28 my ($hdl, $fatal, $msg) = @_;
29 warn "got error $msg\n";
30 $hdl->destroy;
32 $cv->send; 31 $cv->send;
33 },
34 ); 32 );
35 33
36 # send some request line 34 # send some request line
37 $handle->push_write ("getinfo\015\012"); 35 $hdl->push_write ("getinfo\015\012");
38 36
39 # read the response line 37 # read the response line
40 $handle->push_read (line => sub { 38 $hdl->push_read (line => sub {
41 my ($handle, $line) = @_; 39 my ($hdl, $line) = @_;
42 warn "read line <$line>\n"; 40 warn "got line <$line>\n";
43 $cv->send; 41 $cv->send;
44 }); 42 });
45 43
46 $cv->recv; 44 $cv->recv;
47 45
71 69
72=over 4 70=over 4
73 71
74=item fh => $filehandle [MANDATORY] 72=item fh => $filehandle [MANDATORY]
75 73
74#=item fh => $filehandle [C<fh> or C<connect> MANDATORY]
75
76The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
77
78NOTE: The filehandle will be set to non-blocking mode (using 77NOTE: The filehandle will be set to non-blocking mode (using
79C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in 78C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in
80that mode. 79that mode.
81 80
81#=item connect => [$host, $service]
82#
83# You have to specify either this parameter, or C<connect>, below.
84#Try to connect to the specified host and service (port), using
85#C<AnyEvent::Socket::tcp_connect>.
86#
87#When this
88
82=item on_eof => $cb->($handle) 89=item on_eof => $cb->($handle)
83 90
84Set the callback to be called when an end-of-file condition is detected, 91Set the callback to be called when an end-of-file condition is detected,
85i.e. in the case of a socket, when the other side has closed the 92i.e. in the case of a socket, when the other side has closed the
86connection cleanly. 93connection cleanly, and there are no outstanding read requests in the
94queue (if there are read requests, then an EOF counts as an unexpected
95connection close and will be flagged as an error).
87 96
88For sockets, this just means that the other side has stopped sending data, 97For 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 98you 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 99callback and continue writing data, as only the read part has been shut
91down. 100down.
92 101
93While not mandatory, it is I<highly> recommended to set an EOF callback,
94otherwise you might end up with a closed socket while you are still
95waiting for data.
96
97If an EOF condition has been detected but no C<on_eof> callback has been 102If 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>. 103set, then a fatal error will be raised with C<$!> set to <0>.
99 104
100=item on_error => $cb->($handle, $fatal, $message) 105=item on_error => $cb->($handle, $fatal, $message)
101 106
102This is the error callback, which is called when, well, some error 107This is the error callback, which is called when, well, some error
103occured, such as not being able to resolve the hostname, failure to 108occured, such as not being able to resolve the hostname, failure to
104connect or a read error. 109connect or a read error.
105 110
106Some errors are fatal (which is indicated by C<$fatal> being true). On 111Some errors are fatal (which is indicated by C<$fatal> being true). On
107fatal errors the handle object will be shut down and will not be usable 112fatal errors the handle object will be destroyed (by a call to C<< ->
108(but you are free to look at the current C<< ->rbuf >>). Examples of fatal 113destroy >>) after invoking the error callback (which means you are free to
109errors are an EOF condition with active (but unsatisifable) read watchers 114examine the handle object). Examples of fatal errors are an EOF condition
110(C<EPIPE>) or I/O errors. 115with active (but unsatisifable) read watchers (C<EPIPE>) or I/O errors.
111 116
112AnyEvent::Handle tries to find an appropriate error code for you to check 117AnyEvent::Handle tries to find an appropriate error code for you to check
113against, but in some cases (TLS errors), this does not work well. It is 118against, but in some cases (TLS errors), this does not work well. It is
114recommended to always output the C<$message> argument in human-readable 119recommended to always output the C<$message> argument in human-readable
115error messages (it's usually the same as C<"$!">). 120error messages (it's usually the same as C<"$!">).
141 146
142When an EOF condition is detected then AnyEvent::Handle will first try to 147When an EOF condition is detected then AnyEvent::Handle will first try to
143feed all the remaining data to the queued callbacks and C<on_read> before 148feed all the remaining data to the queued callbacks and C<on_read> before
144calling the C<on_eof> callback. If no progress can be made, then a fatal 149calling the C<on_eof> callback. If no progress can be made, then a fatal
145error will be raised (with C<$!> set to C<EPIPE>). 150error will be raised (with C<$!> set to C<EPIPE>).
151
152Note that, unlike requests in the read queue, an C<on_read> callback
153doesn't mean you I<require> some data: if there is an EOF and there
154are outstanding read requests then an error will be flagged. With an
155C<on_read> callback, the C<on_eof> callback will be invoked.
146 156
147=item on_drain => $cb->($handle) 157=item on_drain => $cb->($handle)
148 158
149This sets the callback that is called when the write buffer becomes empty 159This sets the callback that is called when the write buffer becomes empty
150(or when the callback is set and the buffer is empty already). 160(or when the callback is set and the buffer is empty already).
369 if $self->{on_read}; 379 if $self->{on_read};
370 380
371 $self->{fh} && $self 381 $self->{fh} && $self
372} 382}
373 383
374sub _shutdown { 384#sub _shutdown {
375 my ($self) = @_; 385# my ($self) = @_;
376 386#
377 delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)}; 387# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
378 $self->{_eof} = 1; # tell starttls et. al to stop trying 388# $self->{_eof} = 1; # tell starttls et. al to stop trying
379 389#
380 &_freetls; 390# &_freetls;
381} 391#}
382 392
383sub _error { 393sub _error {
384 my ($self, $errno, $fatal, $message) = @_; 394 my ($self, $errno, $fatal, $message) = @_;
385 395
386 $self->_shutdown
387 if $fatal;
388
389 $! = $errno; 396 $! = $errno;
390 $message ||= "$!"; 397 $message ||= "$!";
391 398
392 if ($self->{on_error}) { 399 if ($self->{on_error}) {
393 $self->{on_error}($self, $fatal, $message); 400 $self->{on_error}($self, $fatal, $message);
401 $self->destroy if $fatal;
394 } elsif ($self->{fh}) { 402 } elsif ($self->{fh}) {
403 $self->destroy;
395 Carp::croak "AnyEvent::Handle uncaught error: $message"; 404 Carp::croak "AnyEvent::Handle uncaught error: $message";
396 } 405 }
397} 406}
398 407
399=item $fh = $handle->fh 408=item $fh = $handle->fh
514 $self->{_activity} = $NOW; 523 $self->{_activity} = $NOW;
515 524
516 if ($self->{on_timeout}) { 525 if ($self->{on_timeout}) {
517 $self->{on_timeout}($self); 526 $self->{on_timeout}($self);
518 } else { 527 } else {
519 $self->_error (&Errno::ETIMEDOUT); 528 $self->_error (Errno::ETIMEDOUT);
520 } 529 }
521 530
522 # callback could have changed timeout value, optimise 531 # callback could have changed timeout value, optimise
523 return unless $self->{timeout}; 532 return unless $self->{timeout};
524 533
587 Scalar::Util::weaken $self; 596 Scalar::Util::weaken $self;
588 597
589 my $cb = sub { 598 my $cb = sub {
590 my $len = syswrite $self->{fh}, $self->{wbuf}; 599 my $len = syswrite $self->{fh}, $self->{wbuf};
591 600
592 if ($len >= 0) { 601 if (defined $len) {
593 substr $self->{wbuf}, 0, $len, ""; 602 substr $self->{wbuf}, 0, $len, "";
594 603
595 $self->{_activity} = AnyEvent->now; 604 $self->{_activity} = AnyEvent->now;
596 605
597 $self->{on_drain}($self) 606 $self->{on_drain}($self)
866 875
867 if ( 876 if (
868 defined $self->{rbuf_max} 877 defined $self->{rbuf_max}
869 && $self->{rbuf_max} < length $self->{rbuf} 878 && $self->{rbuf_max} < length $self->{rbuf}
870 ) { 879 ) {
871 $self->_error (&Errno::ENOSPC, 1), return; 880 $self->_error (Errno::ENOSPC, 1), return;
872 } 881 }
873 882
874 while () { 883 while () {
875 # we need to use a separate tls read buffer, as we must not receive data while 884 # we need to use a separate tls read buffer, as we must not receive data while
876 # we are draining the buffer, and this can only happen with TLS. 885 # we are draining the buffer, and this can only happen with TLS.
880 889
881 if (my $cb = shift @{ $self->{_queue} }) { 890 if (my $cb = shift @{ $self->{_queue} }) {
882 unless ($cb->($self)) { 891 unless ($cb->($self)) {
883 if ($self->{_eof}) { 892 if ($self->{_eof}) {
884 # no progress can be made (not enough data and no data forthcoming) 893 # no progress can be made (not enough data and no data forthcoming)
885 $self->_error (&Errno::EPIPE, 1), return; 894 $self->_error (Errno::EPIPE, 1), return;
886 } 895 }
887 896
888 unshift @{ $self->{_queue} }, $cb; 897 unshift @{ $self->{_queue} }, $cb;
889 last; 898 last;
890 } 899 }
898 && !@{ $self->{_queue} } # and the queue is still empty 907 && !@{ $self->{_queue} } # and the queue is still empty
899 && $self->{on_read} # but we still have on_read 908 && $self->{on_read} # but we still have on_read
900 ) { 909 ) {
901 # no further data will arrive 910 # no further data will arrive
902 # so no progress can be made 911 # so no progress can be made
903 $self->_error (&Errno::EPIPE, 1), return 912 $self->_error (Errno::EPIPE, 1), return
904 if $self->{_eof}; 913 if $self->{_eof};
905 914
906 last; # more data might arrive 915 last; # more data might arrive
907 } 916 }
908 } else { 917 } else {
1158 return 1; 1167 return 1;
1159 } 1168 }
1160 1169
1161 # reject 1170 # reject
1162 if ($reject && $$rbuf =~ $reject) { 1171 if ($reject && $$rbuf =~ $reject) {
1163 $self->_error (&Errno::EBADMSG); 1172 $self->_error (Errno::EBADMSG);
1164 } 1173 }
1165 1174
1166 # skip 1175 # skip
1167 if ($skip && $$rbuf =~ $skip) { 1176 if ($skip && $$rbuf =~ $skip) {
1168 $data .= substr $$rbuf, 0, $+[0], ""; 1177 $data .= substr $$rbuf, 0, $+[0], "";
1184 my ($self, $cb) = @_; 1193 my ($self, $cb) = @_;
1185 1194
1186 sub { 1195 sub {
1187 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 1196 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1188 if ($_[0]{rbuf} =~ /[^0-9]/) { 1197 if ($_[0]{rbuf} =~ /[^0-9]/) {
1189 $self->_error (&Errno::EBADMSG); 1198 $self->_error (Errno::EBADMSG);
1190 } 1199 }
1191 return; 1200 return;
1192 } 1201 }
1193 1202
1194 my $len = $1; 1203 my $len = $1;
1197 my $string = $_[1]; 1206 my $string = $_[1];
1198 $_[0]->unshift_read (chunk => 1, sub { 1207 $_[0]->unshift_read (chunk => 1, sub {
1199 if ($_[1] eq ",") { 1208 if ($_[1] eq ",") {
1200 $cb->($_[0], $string); 1209 $cb->($_[0], $string);
1201 } else { 1210 } else {
1202 $self->_error (&Errno::EBADMSG); 1211 $self->_error (Errno::EBADMSG);
1203 } 1212 }
1204 }); 1213 });
1205 }); 1214 });
1206 1215
1207 1 1216 1
1297 $json->incr_skip; 1306 $json->incr_skip;
1298 1307
1299 $self->{rbuf} = $json->incr_text; 1308 $self->{rbuf} = $json->incr_text;
1300 $json->incr_text = ""; 1309 $json->incr_text = "";
1301 1310
1302 $self->_error (&Errno::EBADMSG); 1311 $self->_error (Errno::EBADMSG);
1303 1312
1304 () 1313 ()
1305 } else { 1314 } else {
1306 $self->{rbuf} = ""; 1315 $self->{rbuf} = "";
1307 1316
1344 # read remaining chunk 1353 # read remaining chunk
1345 $_[0]->unshift_read (chunk => $len, sub { 1354 $_[0]->unshift_read (chunk => $len, sub {
1346 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1355 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1347 $cb->($_[0], $ref); 1356 $cb->($_[0], $ref);
1348 } else { 1357 } else {
1349 $self->_error (&Errno::EBADMSG); 1358 $self->_error (Errno::EBADMSG);
1350 } 1359 }
1351 }); 1360 });
1352 } 1361 }
1353 1362
1354 1 1363 1
1452 if ($self->{_on_starttls}) { 1461 if ($self->{_on_starttls}) {
1453 (delete $self->{_on_starttls})->($self, undef, $err); 1462 (delete $self->{_on_starttls})->($self, undef, $err);
1454 &_freetls; 1463 &_freetls;
1455 } else { 1464 } else {
1456 &_freetls; 1465 &_freetls;
1457 $self->_error (&Errno::EPROTO, 1, $err); 1466 $self->_error (Errno::EPROTO, 1, $err);
1458 } 1467 }
1459} 1468}
1460 1469
1461# poll the write BIO and send the data if applicable 1470# poll the write BIO and send the data if applicable
1462# also decode read data if possible 1471# also decode read data if possible
1519 1528
1520Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1529Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1521object is created, you can also do that at a later time by calling 1530object is created, you can also do that at a later time by calling
1522C<starttls>. 1531C<starttls>.
1523 1532
1533Starting TLS is currently an asynchronous operation - when you push some
1534write data and then call C<< ->starttls >> then TLS negotiation will start
1535immediately, after which the queued write data is then sent.
1536
1524The first argument is the same as the C<tls> constructor argument (either 1537The first argument is the same as the C<tls> constructor argument (either
1525C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1538C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1526 1539
1527The second argument is the optional C<AnyEvent::TLS> object that is used 1540The second argument is the optional C<AnyEvent::TLS> object that is used
1528when AnyEvent::Handle has to create its own TLS connection object, or 1541when AnyEvent::Handle has to create its own TLS connection object, or
1552 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1565 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1553 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1566 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1554 1567
1555 $ctx ||= $self->{tls_ctx}; 1568 $ctx ||= $self->{tls_ctx};
1556 1569
1570 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1571
1557 if ("HASH" eq ref $ctx) { 1572 if ("HASH" eq ref $ctx) {
1558 require AnyEvent::TLS; 1573 require AnyEvent::TLS;
1559
1560 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context
1561 1574
1562 if ($ctx->{cache}) { 1575 if ($ctx->{cache}) {
1563 my $key = $ctx+0; 1576 my $key = $ctx+0;
1564 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx; 1577 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx;
1565 } else { 1578 } else {
1637 1650
1638 &_freetls; 1651 &_freetls;
1639 1652
1640 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1653 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1641 1654
1642 if ($linger && length $self->{wbuf}) { 1655 if ($linger && length $self->{wbuf} && $self->{fh}) {
1643 my $fh = delete $self->{fh}; 1656 my $fh = delete $self->{fh};
1644 my $wbuf = delete $self->{wbuf}; 1657 my $wbuf = delete $self->{wbuf};
1645 1658
1646 my @linger; 1659 my @linger;
1647 1660
1671callbacks, as well as code outside. It does I<NOT> work in a read or write 1684callbacks, as well as code outside. It does I<NOT> work in a read or write
1672callback, so when you want to destroy the AnyEvent::Handle object from 1685callback, so when you want to destroy the AnyEvent::Handle object from
1673within such an callback. You I<MUST> call C<< ->destroy >> explicitly in 1686within such an callback. You I<MUST> call C<< ->destroy >> explicitly in
1674that case. 1687that case.
1675 1688
1689Destroying the handle object in this way has the advantage that callbacks
1690will be removed as well, so if those are the only reference holders (as
1691is common), then one doesn't need to do anything special to break any
1692reference cycles.
1693
1676The handle might still linger in the background and write out remaining 1694The handle might still linger in the background and write out remaining
1677data, as specified by the C<linger> option, however. 1695data, as specified by the C<linger> option, however.
1678 1696
1679=cut 1697=cut
1680 1698
1747 1765
1748 $handle->on_read (sub { }); 1766 $handle->on_read (sub { });
1749 $handle->on_eof (undef); 1767 $handle->on_eof (undef);
1750 $handle->on_error (sub { 1768 $handle->on_error (sub {
1751 my $data = delete $_[0]{rbuf}; 1769 my $data = delete $_[0]{rbuf};
1752 undef $handle;
1753 }); 1770 });
1754 1771
1755The reason to use C<on_error> is that TCP connections, due to latencies 1772The reason to use C<on_error> is that TCP connections, due to latencies
1756and packets loss, might get closed quite violently with an error, when in 1773and packets loss, might get closed quite violently with an error, when in
1757fact, all data has been received. 1774fact, all data has been received.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines