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.151 by root, Thu Jul 16 04:20:23 2009 UTC vs.
Revision 1.159 by root, Fri Jul 24 12:35:58 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.82; 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;
… …
47 $cv->recv; 44 $cv->recv;
48 45
49=head1 DESCRIPTION 46=head1 DESCRIPTION
50 47
51This module is a helper module to make it easier to do event-based I/O on 48This module is a helper module to make it easier to do event-based I/O on
52filehandles. For utility functions for doing non-blocking connects and accepts 49filehandles.
53on sockets see L<AnyEvent::Util>.
54 50
55The L<AnyEvent::Intro> tutorial contains some well-documented 51The L<AnyEvent::Intro> tutorial contains some well-documented
56AnyEvent::Handle examples. 52AnyEvent::Handle examples.
57 53
58In the following, when the documentation refers to of "bytes" then this 54In the following, when the documentation refers to of "bytes" then this
59means characters. As sysread and syswrite are used for all I/O, their 55means characters. As sysread and syswrite are used for all I/O, their
60treatment of characters applies to this module as well. 56treatment of characters applies to this module as well.
61 57
58At the very minimum, you should specify C<fh> or C<connect>, and the
59C<on_error> callback.
60
62All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
63argument. 62argument.
64 63
65=head1 METHODS 64=head1 METHODS
66 65
… …
70 69
71The constructor supports these arguments (all as C<< key => value >> pairs). 70The constructor supports these arguments (all as C<< key => value >> pairs).
72 71
73=over 4 72=over 4
74 73
75=item fh => $filehandle [MANDATORY] 74=item fh => $filehandle [C<fh> or C<connect> MANDATORY]
76 75
77The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
78
79NOTE: The filehandle will be set to non-blocking mode (using 77NOTE: The filehandle will be set to non-blocking mode (using
80C<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
81that mode. 79that mode.
80
81=item connect => [$host, $service] [C<fh> or C<connect> MANDATORY]
82
83Try to connect to the specified host and service (port), using
84C<AnyEvent::Socket::tcp_connect>. The C<$host> additionally becomes the
85default C<peername>.
86
87You have to specify either this parameter, or C<fh>, above.
88
89When this parameter is specified, then the C<on_prepare>,
90C<on_connect_error> and C<on_connect> callbacks will be called under the
91appropriate circumstances:
92
93=over 4
94
95=item on_prepare => $cb->($handle)
96
97This (rarely used) callback is called before a new connection is
98attempted, but after the file handle has been created. It could be used to
99prepare the file handle with parameters required for the actual connect
100(as opposed to settings that can be changed when the connection is already
101established).
102
103=item on_connect => $cb->($handle, $host, $port, $retry->())
104
105This callback is called when a connection has been successfully established.
106
107The actual numeric host and port (the socket peername) are passed as
108parameters, together with a retry callback.
109
110When, for some reason, the handle is not acceptable, then calling
111C<$retry> will continue with the next conenction target (in case of
112multi-homed hosts or SRV records there can be multiple connection
113endpoints). When it is called then the read and write queues, eof status,
114tls status and similar properties of the handle are being reset.
115
116In most cases, ignoring the C<$retry> parameter is the way to go.
117
118=item on_connect_error => $cb->($handle, $message)
119
120This callback is called when the conenction could not be
121established. C<$!> will contain the relevant error code, and C<$message> a
122message describing it (usually the same as C<"$!">).
123
124If this callback isn't specified, then C<on_error> will be called with a
125fatal error instead.
126
127=back
128
129=item on_error => $cb->($handle, $fatal, $message)
130
131This is the error callback, which is called when, well, some error
132occured, such as not being able to resolve the hostname, failure to
133connect or a read error.
134
135Some errors are fatal (which is indicated by C<$fatal> being true). On
136fatal errors the handle object will be destroyed (by a call to C<< ->
137destroy >>) after invoking the error callback (which means you are free to
138examine the handle object). Examples of fatal errors are an EOF condition
139with active (but unsatisifable) read watchers (C<EPIPE>) or I/O errors. In
140cases where the other side can close the connection at their will it is
141often easiest to not report C<EPIPE> errors in this callback.
142
143AnyEvent::Handle tries to find an appropriate error code for you to check
144against, but in some cases (TLS errors), this does not work well. It is
145recommended to always output the C<$message> argument in human-readable
146error messages (it's usually the same as C<"$!">).
147
148Non-fatal errors can be retried by simply returning, but it is recommended
149to simply ignore this parameter and instead abondon the handle object
150when this callback is invoked. Examples of non-fatal errors are timeouts
151C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
152
153On callback entrance, the value of C<$!> contains the operating system
154error code (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT>, C<EBADMSG> or
155C<EPROTO>).
156
157While not mandatory, it is I<highly> recommended to set this callback, as
158you will not be notified of errors otherwise. The default simply calls
159C<croak>.
160
161=item on_read => $cb->($handle)
162
163This sets the default read callback, which is called when data arrives
164and no read request is in the queue (unlike read queue callbacks, this
165callback will only be called when at least one octet of data is in the
166read buffer).
167
168To access (and remove data from) the read buffer, use the C<< ->rbuf >>
169method or access the C<< $handle->{rbuf} >> member directly. Note that you
170must not enlarge or modify the read buffer, you can only remove data at
171the beginning from it.
172
173When an EOF condition is detected then AnyEvent::Handle will first try to
174feed all the remaining data to the queued callbacks and C<on_read> before
175calling the C<on_eof> callback. If no progress can be made, then a fatal
176error will be raised (with C<$!> set to C<EPIPE>).
177
178Note that, unlike requests in the read queue, an C<on_read> callback
179doesn't mean you I<require> some data: if there is an EOF and there
180are outstanding read requests then an error will be flagged. With an
181C<on_read> callback, the C<on_eof> callback will be invoked.
82 182
83=item on_eof => $cb->($handle) 183=item on_eof => $cb->($handle)
84 184
85Set the callback to be called when an end-of-file condition is detected, 185Set the callback to be called when an end-of-file condition is detected,
86i.e. in the case of a socket, when the other side has closed the 186i.e. in the case of a socket, when the other side has closed the
… …
93callback and continue writing data, as only the read part has been shut 193callback and continue writing data, as only the read part has been shut
94down. 194down.
95 195
96If an EOF condition has been detected but no C<on_eof> callback has been 196If an EOF condition has been detected but no C<on_eof> callback has been
97set, then a fatal error will be raised with C<$!> set to <0>. 197set, then a fatal error will be raised with C<$!> set to <0>.
98
99=item on_error => $cb->($handle, $fatal, $message)
100
101This is the error callback, which is called when, well, some error
102occured, such as not being able to resolve the hostname, failure to
103connect or a read error.
104
105Some errors are fatal (which is indicated by C<$fatal> being true). On
106fatal errors the handle object will be destroyed (by a call to C<< ->
107destroy >>) after invoking the error callback (which means you are free to
108examine the handle object). Examples of fatal errors are an EOF condition
109with active (but unsatisifable) read watchers (C<EPIPE>) or I/O errors.
110
111AnyEvent::Handle tries to find an appropriate error code for you to check
112against, but in some cases (TLS errors), this does not work well. It is
113recommended to always output the C<$message> argument in human-readable
114error messages (it's usually the same as C<"$!">).
115
116Non-fatal errors can be retried by simply returning, but it is recommended
117to simply ignore this parameter and instead abondon the handle object
118when this callback is invoked. Examples of non-fatal errors are timeouts
119C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
120
121On callback entrance, the value of C<$!> contains the operating system
122error code (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT>, C<EBADMSG> or
123C<EPROTO>).
124
125While not mandatory, it is I<highly> recommended to set this callback, as
126you will not be notified of errors otherwise. The default simply calls
127C<croak>.
128
129=item on_read => $cb->($handle)
130
131This sets the default read callback, which is called when data arrives
132and no read request is in the queue (unlike read queue callbacks, this
133callback will only be called when at least one octet of data is in the
134read buffer).
135
136To access (and remove data from) the read buffer, use the C<< ->rbuf >>
137method or access the C<< $handle->{rbuf} >> member directly. Note that you
138must not enlarge or modify the read buffer, you can only remove data at
139the beginning from it.
140
141When an EOF condition is detected then AnyEvent::Handle will first try to
142feed all the remaining data to the queued callbacks and C<on_read> before
143calling the C<on_eof> callback. If no progress can be made, then a fatal
144error will be raised (with C<$!> set to C<EPIPE>).
145
146Note that, unlike requests in the read queue, an C<on_read> callback
147doesn't mean you I<require> some data: if there is an EOF and there
148are outstanding read requests then an error will be flagged. With an
149C<on_read> callback, the C<on_eof> callback will be invoked.
150 198
151=item on_drain => $cb->($handle) 199=item on_drain => $cb->($handle)
152 200
153This sets the callback that is called when the write buffer becomes empty 201This sets the callback that is called when the write buffer becomes empty
154(or when the callback is set and the buffer is empty already). 202(or when the callback is set and the buffer is empty already).
… …
353 401
354sub new { 402sub new {
355 my $class = shift; 403 my $class = shift;
356 my $self = bless { @_ }, $class; 404 my $self = bless { @_ }, $class;
357 405
358 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 406 if ($self->{fh}) {
407 $self->_start;
408 return unless $self->{fh}; # could be gone by now
409
410 } elsif ($self->{connect}) {
411 require AnyEvent::Socket;
412
413 $self->{peername} = $self->{connect}[0]
414 unless exists $self->{peername};
415
416 $self->{_skip_drain_rbuf} = 1;
417
418 {
419 Scalar::Util::weaken (my $self = $self);
420
421 $self->{_connect} =
422 AnyEvent::Socket::tcp_connect (
423 $self->{connect}[0],
424 $self->{connect}[1],
425 sub {
426 my ($fh, $host, $port, $retry) = @_;
427
428 if ($fh) {
429 $self->{fh} = $fh;
430
431 delete $self->{_skip_drain_rbuf};
432 $self->_start;
433
434 $self->{on_connect}
435 and $self->{on_connect}($self, $host, $port, sub {
436 delete @$self{qw(fh _tw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
437 $self->{_skip_drain_rbuf} = 1;
438 &$retry;
439 });
440
441 } else {
442 if ($self->{on_connect_error}) {
443 $self->{on_connect_error}($self, "$!");
444 $self->destroy;
445 } else {
446 $self->fatal ($!, 1);
447 }
448 }
449 },
450 sub {
451 local $self->{fh} = $_[0];
452
453 $self->{on_prepare}->($self)
454 if $self->{on_prepare};
455 }
456 );
457 }
458
459 } else {
460 Carp::croak "AnyEvent::Handle: either an existing fh or the connect parameter must be specified";
461 }
462
463 $self
464}
465
466sub _start {
467 my ($self) = @_;
359 468
360 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 469 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
361 470
362 $self->{_activity} = AnyEvent->now; 471 $self->{_activity} = AnyEvent->now;
363 $self->_timeout; 472 $self->_timeout;
… …
368 if $self->{tls}; 477 if $self->{tls};
369 478
370 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 479 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
371 480
372 $self->start_read 481 $self->start_read
373 if $self->{on_read}; 482 if $self->{on_read} || @{ $self->{_queue} };
374
375 $self->{fh} && $self
376} 483}
377 484
378#sub _shutdown { 485#sub _shutdown {
379# my ($self) = @_; 486# my ($self) = @_;
380# 487#
… …
460sub no_delay { 567sub no_delay {
461 $_[0]{no_delay} = $_[1]; 568 $_[0]{no_delay} = $_[1];
462 569
463 eval { 570 eval {
464 local $SIG{__DIE__}; 571 local $SIG{__DIE__};
465 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1]; 572 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1]
573 if $_[0]{fh};
466 }; 574 };
467} 575}
468 576
469=item $handle->on_starttls ($cb) 577=item $handle->on_starttls ($cb)
470 578
… …
504# reset the timeout watcher, as neccessary 612# reset the timeout watcher, as neccessary
505# also check for time-outs 613# also check for time-outs
506sub _timeout { 614sub _timeout {
507 my ($self) = @_; 615 my ($self) = @_;
508 616
509 if ($self->{timeout}) { 617 if ($self->{timeout} && $self->{fh}) {
510 my $NOW = AnyEvent->now; 618 my $NOW = AnyEvent->now;
511 619
512 # when would the timeout trigger? 620 # when would the timeout trigger?
513 my $after = $self->{_activity} + $self->{timeout} - $NOW; 621 my $after = $self->{_activity} + $self->{timeout} - $NOW;
514 622
… …
636 $self->{_tls_wbuf} .= $_[0]; 744 $self->{_tls_wbuf} .= $_[0];
637 745
638 &_dotls ($self); 746 &_dotls ($self);
639 } else { 747 } else {
640 $self->{wbuf} .= $_[0]; 748 $self->{wbuf} .= $_[0];
641 $self->_drain_wbuf; 749 $self->_drain_wbuf if $self->{fh};
642 } 750 }
643} 751}
644 752
645=item $handle->push_write (type => @args) 753=item $handle->push_write (type => @args)
646 754
… …
863=cut 971=cut
864 972
865sub _drain_rbuf { 973sub _drain_rbuf {
866 my ($self) = @_; 974 my ($self) = @_;
867 975
976 # avoid recursion
977 return if exists $self->{_skip_drain_rbuf};
868 local $self->{_in_drain} = 1; 978 local $self->{_skip_drain_rbuf} = 1;
869 979
870 if ( 980 if (
871 defined $self->{rbuf_max} 981 defined $self->{rbuf_max}
872 && $self->{rbuf_max} < length $self->{rbuf} 982 && $self->{rbuf_max} < length $self->{rbuf}
873 ) { 983 ) {
… …
940 1050
941sub on_read { 1051sub on_read {
942 my ($self, $cb) = @_; 1052 my ($self, $cb) = @_;
943 1053
944 $self->{on_read} = $cb; 1054 $self->{on_read} = $cb;
945 $self->_drain_rbuf if $cb && !$self->{_in_drain}; 1055 $self->_drain_rbuf if $cb;
946} 1056}
947 1057
948=item $handle->rbuf 1058=item $handle->rbuf
949 1059
950Returns the read buffer (as a modifiable lvalue). 1060Returns the read buffer (as a modifiable lvalue).
… …
1002 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 1112 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
1003 ->($self, $cb, @_); 1113 ->($self, $cb, @_);
1004 } 1114 }
1005 1115
1006 push @{ $self->{_queue} }, $cb; 1116 push @{ $self->{_queue} }, $cb;
1007 $self->_drain_rbuf unless $self->{_in_drain}; 1117 $self->_drain_rbuf;
1008} 1118}
1009 1119
1010sub unshift_read { 1120sub unshift_read {
1011 my $self = shift; 1121 my $self = shift;
1012 my $cb = pop; 1122 my $cb = pop;
… …
1018 ->($self, $cb, @_); 1128 ->($self, $cb, @_);
1019 } 1129 }
1020 1130
1021 1131
1022 unshift @{ $self->{_queue} }, $cb; 1132 unshift @{ $self->{_queue} }, $cb;
1023 $self->_drain_rbuf unless $self->{_in_drain}; 1133 $self->_drain_rbuf;
1024} 1134}
1025 1135
1026=item $handle->push_read (type => @args, $cb) 1136=item $handle->push_read (type => @args, $cb)
1027 1137
1028=item $handle->unshift_read (type => @args, $cb) 1138=item $handle->unshift_read (type => @args, $cb)
… …
1421 if ($self->{tls}) { 1531 if ($self->{tls}) {
1422 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1532 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1423 1533
1424 &_dotls ($self); 1534 &_dotls ($self);
1425 } else { 1535 } else {
1426 $self->_drain_rbuf unless $self->{_in_drain}; 1536 $self->_drain_rbuf;
1427 } 1537 }
1428 1538
1429 } elsif (defined $len) { 1539 } elsif (defined $len) {
1430 delete $self->{_rw}; 1540 delete $self->{_rw};
1431 $self->{_eof} = 1; 1541 $self->{_eof} = 1;
1432 $self->_drain_rbuf unless $self->{_in_drain}; 1542 $self->_drain_rbuf;
1433 1543
1434 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1544 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1435 return $self->_error ($!, 1); 1545 return $self->_error ($!, 1);
1436 } 1546 }
1437 }); 1547 });
… …
1497 $self->{_eof} = 1; 1607 $self->{_eof} = 1;
1498 } 1608 }
1499 } 1609 }
1500 1610
1501 $self->{_tls_rbuf} .= $tmp; 1611 $self->{_tls_rbuf} .= $tmp;
1502 $self->_drain_rbuf unless $self->{_in_drain}; 1612 $self->_drain_rbuf;
1503 $self->{tls} or return; # tls session might have gone away in callback 1613 $self->{tls} or return; # tls session might have gone away in callback
1504 } 1614 }
1505 1615
1506 $tmp = Net::SSLeay::get_error ($self->{tls}, -1); 1616 $tmp = Net::SSLeay::get_error ($self->{tls}, -1);
1507 return $self->_tls_error ($tmp) 1617 return $self->_tls_error ($tmp)
… …
1522 1632
1523Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1633Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1524object is created, you can also do that at a later time by calling 1634object is created, you can also do that at a later time by calling
1525C<starttls>. 1635C<starttls>.
1526 1636
1637Starting TLS is currently an asynchronous operation - when you push some
1638write data and then call C<< ->starttls >> then TLS negotiation will start
1639immediately, after which the queued write data is then sent.
1640
1527The first argument is the same as the C<tls> constructor argument (either 1641The first argument is the same as the C<tls> constructor argument (either
1528C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1642C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1529 1643
1530The second argument is the optional C<AnyEvent::TLS> object that is used 1644The second argument is the optional C<AnyEvent::TLS> object that is used
1531when AnyEvent::Handle has to create its own TLS connection object, or 1645when AnyEvent::Handle has to create its own TLS connection object, or
… …
1555 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1669 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1556 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1670 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1557 1671
1558 $ctx ||= $self->{tls_ctx}; 1672 $ctx ||= $self->{tls_ctx};
1559 1673
1674 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1675
1560 if ("HASH" eq ref $ctx) { 1676 if ("HASH" eq ref $ctx) {
1561 require AnyEvent::TLS; 1677 require AnyEvent::TLS;
1562
1563 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context
1564 1678
1565 if ($ctx->{cache}) { 1679 if ($ctx->{cache}) {
1566 my $key = $ctx+0; 1680 my $key = $ctx+0;
1567 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx; 1681 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx;
1568 } else { 1682 } else {
… …
1640 1754
1641 &_freetls; 1755 &_freetls;
1642 1756
1643 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1757 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1644 1758
1645 if ($linger && length $self->{wbuf}) { 1759 if ($linger && length $self->{wbuf} && $self->{fh}) {
1646 my $fh = delete $self->{fh}; 1760 my $fh = delete $self->{fh};
1647 my $wbuf = delete $self->{wbuf}; 1761 my $wbuf = delete $self->{wbuf};
1648 1762
1649 my @linger; 1763 my @linger;
1650 1764

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines