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.154 by root, Sat Jul 18 05:19:09 2009 UTC vs.
Revision 1.159 by root, Fri Jul 24 12:35:58 2009 UTC

… …
11 11
12AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 12AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
13 13
14=cut 14=cut
15 15
16our $VERSION = 4.85; 16our $VERSION = 4.86;
17 17
18=head1 SYNOPSIS 18=head1 SYNOPSIS
19 19
20 use AnyEvent; 20 use AnyEvent;
21 use AnyEvent::Handle; 21 use AnyEvent::Handle;
… …
44 $cv->recv; 44 $cv->recv;
45 45
46=head1 DESCRIPTION 46=head1 DESCRIPTION
47 47
48This 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
49filehandles. For utility functions for doing non-blocking connects and accepts 49filehandles.
50on sockets see L<AnyEvent::Util>.
51 50
52The L<AnyEvent::Intro> tutorial contains some well-documented 51The L<AnyEvent::Intro> tutorial contains some well-documented
53AnyEvent::Handle examples. 52AnyEvent::Handle examples.
54 53
55In the following, when the documentation refers to of "bytes" then this 54In the following, when the documentation refers to of "bytes" then this
56means 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
57treatment of characters applies to this module as well. 56treatment of characters applies to this module as well.
58 57
58At the very minimum, you should specify C<fh> or C<connect>, and the
59C<on_error> callback.
60
59All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
60argument. 62argument.
61 63
62=head1 METHODS 64=head1 METHODS
63 65
… …
67 69
68The constructor supports these arguments (all as C<< key => value >> pairs). 70The constructor supports these arguments (all as C<< key => value >> pairs).
69 71
70=over 4 72=over 4
71 73
72=item fh => $filehandle [MANDATORY] 74=item fh => $filehandle [C<fh> or C<connect> MANDATORY]
73 75
74The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
75
76NOTE: The filehandle will be set to non-blocking mode (using 77NOTE: The filehandle will be set to non-blocking mode (using
77C<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
78that 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.
79 182
80=item on_eof => $cb->($handle) 183=item on_eof => $cb->($handle)
81 184
82Set 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,
83i.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
… …
90callback 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
91down. 194down.
92 195
93If 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
94set, 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>.
95
96=item on_error => $cb->($handle, $fatal, $message)
97
98This is the error callback, which is called when, well, some error
99occured, such as not being able to resolve the hostname, failure to
100connect or a read error.
101
102Some errors are fatal (which is indicated by C<$fatal> being true). On
103fatal errors the handle object will be destroyed (by a call to C<< ->
104destroy >>) after invoking the error callback (which means you are free to
105examine the handle object). Examples of fatal errors are an EOF condition
106with active (but unsatisifable) read watchers (C<EPIPE>) or I/O errors.
107
108AnyEvent::Handle tries to find an appropriate error code for you to check
109against, but in some cases (TLS errors), this does not work well. It is
110recommended to always output the C<$message> argument in human-readable
111error messages (it's usually the same as C<"$!">).
112
113Non-fatal errors can be retried by simply returning, but it is recommended
114to simply ignore this parameter and instead abondon the handle object
115when this callback is invoked. Examples of non-fatal errors are timeouts
116C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
117
118On callback entrance, the value of C<$!> contains the operating system
119error code (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT>, C<EBADMSG> or
120C<EPROTO>).
121
122While not mandatory, it is I<highly> recommended to set this callback, as
123you will not be notified of errors otherwise. The default simply calls
124C<croak>.
125
126=item on_read => $cb->($handle)
127
128This sets the default read callback, which is called when data arrives
129and no read request is in the queue (unlike read queue callbacks, this
130callback will only be called when at least one octet of data is in the
131read buffer).
132
133To access (and remove data from) the read buffer, use the C<< ->rbuf >>
134method or access the C<< $handle->{rbuf} >> member directly. Note that you
135must not enlarge or modify the read buffer, you can only remove data at
136the beginning from it.
137
138When an EOF condition is detected then AnyEvent::Handle will first try to
139feed all the remaining data to the queued callbacks and C<on_read> before
140calling the C<on_eof> callback. If no progress can be made, then a fatal
141error will be raised (with C<$!> set to C<EPIPE>).
142
143Note that, unlike requests in the read queue, an C<on_read> callback
144doesn't mean you I<require> some data: if there is an EOF and there
145are outstanding read requests then an error will be flagged. With an
146C<on_read> callback, the C<on_eof> callback will be invoked.
147 198
148=item on_drain => $cb->($handle) 199=item on_drain => $cb->($handle)
149 200
150This 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
151(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).
… …
350 401
351sub new { 402sub new {
352 my $class = shift; 403 my $class = shift;
353 my $self = bless { @_ }, $class; 404 my $self = bless { @_ }, $class;
354 405
355 $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) = @_;
356 468
357 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 469 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
358 470
359 $self->{_activity} = AnyEvent->now; 471 $self->{_activity} = AnyEvent->now;
360 $self->_timeout; 472 $self->_timeout;
… …
365 if $self->{tls}; 477 if $self->{tls};
366 478
367 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 479 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
368 480
369 $self->start_read 481 $self->start_read
370 if $self->{on_read}; 482 if $self->{on_read} || @{ $self->{_queue} };
371
372 $self->{fh} && $self
373} 483}
374 484
375#sub _shutdown { 485#sub _shutdown {
376# my ($self) = @_; 486# my ($self) = @_;
377# 487#
… …
457sub no_delay { 567sub no_delay {
458 $_[0]{no_delay} = $_[1]; 568 $_[0]{no_delay} = $_[1];
459 569
460 eval { 570 eval {
461 local $SIG{__DIE__}; 571 local $SIG{__DIE__};
462 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};
463 }; 574 };
464} 575}
465 576
466=item $handle->on_starttls ($cb) 577=item $handle->on_starttls ($cb)
467 578
… …
501# reset the timeout watcher, as neccessary 612# reset the timeout watcher, as neccessary
502# also check for time-outs 613# also check for time-outs
503sub _timeout { 614sub _timeout {
504 my ($self) = @_; 615 my ($self) = @_;
505 616
506 if ($self->{timeout}) { 617 if ($self->{timeout} && $self->{fh}) {
507 my $NOW = AnyEvent->now; 618 my $NOW = AnyEvent->now;
508 619
509 # when would the timeout trigger? 620 # when would the timeout trigger?
510 my $after = $self->{_activity} + $self->{timeout} - $NOW; 621 my $after = $self->{_activity} + $self->{timeout} - $NOW;
511 622
… …
633 $self->{_tls_wbuf} .= $_[0]; 744 $self->{_tls_wbuf} .= $_[0];
634 745
635 &_dotls ($self); 746 &_dotls ($self);
636 } else { 747 } else {
637 $self->{wbuf} .= $_[0]; 748 $self->{wbuf} .= $_[0];
638 $self->_drain_wbuf; 749 $self->_drain_wbuf if $self->{fh};
639 } 750 }
640} 751}
641 752
642=item $handle->push_write (type => @args) 753=item $handle->push_write (type => @args)
643 754
… …
860=cut 971=cut
861 972
862sub _drain_rbuf { 973sub _drain_rbuf {
863 my ($self) = @_; 974 my ($self) = @_;
864 975
976 # avoid recursion
977 return if exists $self->{_skip_drain_rbuf};
865 local $self->{_in_drain} = 1; 978 local $self->{_skip_drain_rbuf} = 1;
866 979
867 if ( 980 if (
868 defined $self->{rbuf_max} 981 defined $self->{rbuf_max}
869 && $self->{rbuf_max} < length $self->{rbuf} 982 && $self->{rbuf_max} < length $self->{rbuf}
870 ) { 983 ) {
… …
937 1050
938sub on_read { 1051sub on_read {
939 my ($self, $cb) = @_; 1052 my ($self, $cb) = @_;
940 1053
941 $self->{on_read} = $cb; 1054 $self->{on_read} = $cb;
942 $self->_drain_rbuf if $cb && !$self->{_in_drain}; 1055 $self->_drain_rbuf if $cb;
943} 1056}
944 1057
945=item $handle->rbuf 1058=item $handle->rbuf
946 1059
947Returns the read buffer (as a modifiable lvalue). 1060Returns the read buffer (as a modifiable lvalue).
… …
999 $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")
1000 ->($self, $cb, @_); 1113 ->($self, $cb, @_);
1001 } 1114 }
1002 1115
1003 push @{ $self->{_queue} }, $cb; 1116 push @{ $self->{_queue} }, $cb;
1004 $self->_drain_rbuf unless $self->{_in_drain}; 1117 $self->_drain_rbuf;
1005} 1118}
1006 1119
1007sub unshift_read { 1120sub unshift_read {
1008 my $self = shift; 1121 my $self = shift;
1009 my $cb = pop; 1122 my $cb = pop;
… …
1015 ->($self, $cb, @_); 1128 ->($self, $cb, @_);
1016 } 1129 }
1017 1130
1018 1131
1019 unshift @{ $self->{_queue} }, $cb; 1132 unshift @{ $self->{_queue} }, $cb;
1020 $self->_drain_rbuf unless $self->{_in_drain}; 1133 $self->_drain_rbuf;
1021} 1134}
1022 1135
1023=item $handle->push_read (type => @args, $cb) 1136=item $handle->push_read (type => @args, $cb)
1024 1137
1025=item $handle->unshift_read (type => @args, $cb) 1138=item $handle->unshift_read (type => @args, $cb)
… …
1418 if ($self->{tls}) { 1531 if ($self->{tls}) {
1419 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1532 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1420 1533
1421 &_dotls ($self); 1534 &_dotls ($self);
1422 } else { 1535 } else {
1423 $self->_drain_rbuf unless $self->{_in_drain}; 1536 $self->_drain_rbuf;
1424 } 1537 }
1425 1538
1426 } elsif (defined $len) { 1539 } elsif (defined $len) {
1427 delete $self->{_rw}; 1540 delete $self->{_rw};
1428 $self->{_eof} = 1; 1541 $self->{_eof} = 1;
1429 $self->_drain_rbuf unless $self->{_in_drain}; 1542 $self->_drain_rbuf;
1430 1543
1431 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1544 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1432 return $self->_error ($!, 1); 1545 return $self->_error ($!, 1);
1433 } 1546 }
1434 }); 1547 });
… …
1494 $self->{_eof} = 1; 1607 $self->{_eof} = 1;
1495 } 1608 }
1496 } 1609 }
1497 1610
1498 $self->{_tls_rbuf} .= $tmp; 1611 $self->{_tls_rbuf} .= $tmp;
1499 $self->_drain_rbuf unless $self->{_in_drain}; 1612 $self->_drain_rbuf;
1500 $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
1501 } 1614 }
1502 1615
1503 $tmp = Net::SSLeay::get_error ($self->{tls}, -1); 1616 $tmp = Net::SSLeay::get_error ($self->{tls}, -1);
1504 return $self->_tls_error ($tmp) 1617 return $self->_tls_error ($tmp)
… …
1519 1632
1520Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1633Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1521object 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
1522C<starttls>. 1635C<starttls>.
1523 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
1524The 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
1525C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1642C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1526 1643
1527The 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
1528when AnyEvent::Handle has to create its own TLS connection object, or 1645when AnyEvent::Handle has to create its own TLS connection object, or
… …
1552 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1669 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1553 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1670 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1554 1671
1555 $ctx ||= $self->{tls_ctx}; 1672 $ctx ||= $self->{tls_ctx};
1556 1673
1674 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1675
1557 if ("HASH" eq ref $ctx) { 1676 if ("HASH" eq ref $ctx) {
1558 require AnyEvent::TLS; 1677 require AnyEvent::TLS;
1559
1560 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context
1561 1678
1562 if ($ctx->{cache}) { 1679 if ($ctx->{cache}) {
1563 my $key = $ctx+0; 1680 my $key = $ctx+0;
1564 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx; 1681 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx;
1565 } else { 1682 } else {
… …
1637 1754
1638 &_freetls; 1755 &_freetls;
1639 1756
1640 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1757 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1641 1758
1642 if ($linger && length $self->{wbuf}) { 1759 if ($linger && length $self->{wbuf} && $self->{fh}) {
1643 my $fh = delete $self->{fh}; 1760 my $fh = delete $self->{fh};
1644 my $wbuf = delete $self->{wbuf}; 1761 my $wbuf = delete $self->{wbuf};
1645 1762
1646 my @linger; 1763 my @linger;
1647 1764

Diff Legend

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