--- AnyEvent/lib/AnyEvent/Handle.pm 2008/05/29 00:20:39 1.45 +++ AnyEvent/lib/AnyEvent/Handle.pm 2008/06/05 16:53:11 1.59 @@ -16,7 +16,7 @@ =cut -our $VERSION = '0.04'; +our $VERSION = 4.13; =head1 SYNOPSIS @@ -77,30 +77,32 @@ =item on_eof => $cb->($handle) -Set the callback to be called on EOF. +Set the callback to be called when an end-of-file condition is detcted, +i.e. in the case of a socket, when the other side has closed the +connection cleanly. While not mandatory, it is highly recommended to set an eof callback, otherwise you might end up with a closed socket while you are still waiting for data. -=item on_error => $cb->($handle) +=item on_error => $cb->($handle, $fatal) -This is the fatal error callback, that is called when, well, a fatal error -occurs, such as not being able to resolve the hostname, failure to connect -or a read error. - -The object will not be in a usable state when this callback has been -called. +This is the error callback, which is called when, well, some error +occured, such as not being able to resolve the hostname, failure to +connect or a read error. + +Some errors are fatal (which is indicated by C<$fatal> being true). On +fatal errors the handle object will be shut down and will not be +usable. Non-fatal errors can be retried by simply returning, but it is +recommended to simply ignore this parameter and instead abondon the handle +object when this callback is invoked. On callback entrance, the value of C<$!> contains the operating system error (or C, C, C or C). -The callback should throw an exception. If it returns, then -AnyEvent::Handle will C for you. - While not mandatory, it is I recommended to set this callback, as you will not be notified of errors otherwise. The default simply calls -die. +C. =item on_read => $cb->($handle) @@ -157,7 +159,7 @@ =item read_size => The default read block size (the amount of bytes this module will try to read -on each [loop iteration). Default: C<4096>. +during each (loop iteration). Default: C<8192>. =item low_water_mark => @@ -224,15 +226,11 @@ $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); } -# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop -# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop -# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop - $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; - $self->{_activity} = AnyEvent->now; $self->_timeout; - $self->start_read; + $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; + $self->on_read (delete $self->{on_read} ) if $self->{on_read}; $self } @@ -240,23 +238,27 @@ sub _shutdown { my ($self) = @_; + delete $self->{_tw}; delete $self->{_rw}; delete $self->{_ww}; delete $self->{fh}; + + $self->stoptls; } -sub error { - my ($self) = @_; +sub _error { + my ($self, $errno, $fatal) = @_; - { - local $!; - $self->_shutdown; - } + $self->_shutdown + if $fatal; - $self->{on_error}($self) - if $self->{on_error}; + $! = $errno; - Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; + if ($self->{on_error}) { + $self->{on_error}($self, $fatal); + } else { + Carp::croak "AnyEvent::Handle uncaught error: $!"; + } } =item $fh = $handle->fh @@ -330,13 +332,12 @@ $self->{_activity} = $NOW; if ($self->{on_timeout}) { - $self->{on_timeout}->($self); + $self->{on_timeout}($self); } else { - $! = Errno::ETIMEDOUT; - $self->error; + $self->_error (&Errno::ETIMEDOUT); } - # callbakx could have changed timeout value, optimise + # callback could have changed timeout value, optimise return unless $self->{timeout}; # calculate new after @@ -344,6 +345,7 @@ } Scalar::Util::weaken $self; + return unless $self; # ->error could have destroyed $self $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { delete $self->{_tw}; @@ -416,7 +418,7 @@ delete $self->{_ww} unless length $self->{wbuf}; } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { - $self->error; + $self->_error ($!, 1); } }; @@ -446,7 +448,7 @@ } if ($self->{filter_w}) { - $self->{filter_w}->($self, \$_[0]); + $self->{filter_w}($self, \$_[0]); } else { $self->{wbuf} .= $_[0]; $self->_drain_wbuf; @@ -455,8 +457,6 @@ =item $handle->push_write (type => @args) -=item $handle->unshift_write (type => @args) - Instead of formatting your data yourself, you can also let this module do the job by specifying a type and type-specific arguments. @@ -470,8 +470,6 @@ Formats the given value as netstring (http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). -=back - =cut register_write_type netstring => sub { @@ -521,6 +519,8 @@ : JSON::encode_json ($ref) }; +=back + =item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) This function (not method) lets you add your own types to C. @@ -567,12 +567,12 @@ # in the default state, expect some header bytes $handle->on_read (sub { # some data is here, now queue the length-header-read (4 octets) - shift->unshift_read_chunk (4, sub { + shift->unshift_read (chunk => 4, sub { # header arrived, decode my $len = unpack "N", $_[1]; # now read the payload - shift->unshift_read_chunk ($len, sub { + shift->unshift_read (chunk => $len, sub { my $xml = $_[1]; # handle xml }); @@ -589,13 +589,13 @@ $handle->push_write ("request 1\015\012"); # we expect "ERROR" or "OK" as response, so push a line read - $handle->push_read_line (sub { + $handle->push_read (line => sub { # if we got an "OK", we have to _prepend_ another line, # so it will be read before the second request reads its 64 bytes # which are already in the queue when this callback is called # we don't do this in case we got an error if ($_[1] eq "OK") { - $_[0]->unshift_read_line (sub { + $_[0]->unshift_read (line => sub { my $response = $_[1]; ... }); @@ -606,7 +606,7 @@ $handle->push_write ("request 2\015\012"); # simply read 64 bytes, always - $handle->push_read_chunk (64, sub { + $handle->push_read (chunk => 64, sub { my $response = $_[1]; ... }); @@ -618,54 +618,59 @@ sub _drain_rbuf { my ($self) = @_; + local $self->{_in_drain} = 1; + if ( defined $self->{rbuf_max} && $self->{rbuf_max} < length $self->{rbuf} ) { - $! = &Errno::ENOSPC; - $self->error; + return $self->_error (&Errno::ENOSPC, 1); } - return if $self->{in_drain}; - local $self->{in_drain} = 1; - - while (my $len = length $self->{rbuf}) { + while () { no strict 'refs'; + + my $len = length $self->{rbuf}; + if (my $cb = shift @{ $self->{_queue} }) { unless ($cb->($self)) { if ($self->{_eof}) { # no progress can be made (not enough data and no data forthcoming) - $! = &Errno::EPIPE; - $self->error; + return $self->_error (&Errno::EPIPE, 1); } unshift @{ $self->{_queue} }, $cb; - return; + last; } } elsif ($self->{on_read}) { $self->{on_read}($self); if ( - $self->{_eof} # if no further data will arrive - && $len == length $self->{rbuf} # and no data has been consumed - && !@{ $self->{_queue} } # and the queue is still empty - && $self->{on_read} # and we still want to read data + $len == length $self->{rbuf} # if no data has been consumed + && !@{ $self->{_queue} } # and the queue is still empty + && $self->{on_read} # but we still have on_read ) { - # then no progress can be made - $! = &Errno::EPIPE; - $self->error; + # no further data will arrive + # so no progress can be made + return $self->_error (&Errno::EPIPE, 1) + if $self->{_eof}; + + last; # more data might arrive } } else { # read side becomes idle delete $self->{_rw}; - return; + last; } } - if ($self->{_eof}) { - $self->_shutdown; - $self->{on_eof}($self) - if $self->{on_eof}; + $self->{on_eof}($self) + if $self->{_eof} && $self->{on_eof}; + + # may need to restart read watcher + unless ($self->{_rw}) { + $self->start_read + if $self->{on_read} || @{ $self->{_queue} }; } } @@ -681,6 +686,7 @@ my ($self, $cb) = @_; $self->{on_read} = $cb; + $self->_drain_rbuf if $cb && !$self->{_in_drain}; } =item $handle->rbuf @@ -739,7 +745,7 @@ } push @{ $self->{_queue} }, $cb; - $self->_drain_rbuf; + $self->_drain_rbuf unless $self->{_in_drain}; } sub unshift_read { @@ -755,7 +761,7 @@ unshift @{ $self->{_queue} }, $cb; - $self->_drain_rbuf; + $self->_drain_rbuf unless $self->{_in_drain}; } =item $handle->push_read (type => @args, $cb) @@ -865,8 +871,7 @@ sub { unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { if ($_[0]{rbuf} =~ /[^0-9]/) { - $! = &Errno::EBADMSG; - $self->error; + $self->_error (&Errno::EBADMSG); } return; } @@ -879,8 +884,7 @@ if ($_[1] eq ",") { $cb->($_[0], $string); } else { - $! = &Errno::EBADMSG; - $self->error; + $self->_error (&Errno::EBADMSG); } }); }); @@ -947,8 +951,7 @@ # reject if ($reject && $$rbuf =~ $reject) { - $! = &Errno::EBADMSG; - $self->error; + $self->_error (&Errno::EBADMSG); } # skip @@ -1031,10 +1034,15 @@ =item $handle->start_read In rare cases you actually do not want to read anything from the -socket. In this case you can call C. Neither C no +socket. In this case you can call C. Neither C nor any queued callbacks will be executed then. To start reading again, call C. +Note that AnyEvent::Handle will automatically C for you when +you change the C callback or push/unshift a read callback, and it +will automatically C for you when neither C is set nor +there are any read requests in the queue. + =cut sub stop_read { @@ -1057,18 +1065,16 @@ $self->{_activity} = AnyEvent->now; $self->{filter_r} - ? $self->{filter_r}->($self, $rbuf) - : $self->_drain_rbuf; + ? $self->{filter_r}($self, $rbuf) + : $self->{_in_drain} || $self->_drain_rbuf; } elsif (defined $len) { delete $self->{_rw}; - delete $self->{_ww}; - delete $self->{_tw}; $self->{_eof} = 1; - $self->_drain_rbuf; + $self->_drain_rbuf unless $self->{_in_drain}; } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { - return $self->error; + return $self->_error ($!, 1); } }); } @@ -1077,30 +1083,38 @@ sub _dotls { my ($self) = @_; + my $buf; + if (length $self->{_tls_wbuf}) { while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { substr $self->{_tls_wbuf}, 0, $len, ""; } } - if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { + if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { $self->{wbuf} .= $buf; $self->_drain_wbuf; } - while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { - $self->{rbuf} .= $buf; - $self->_drain_rbuf; + while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { + if (length $buf) { + $self->{rbuf} .= $buf; + $self->_drain_rbuf unless $self->{_in_drain}; + } else { + # let's treat SSL-eof as we treat normal EOF + $self->{_eof} = 1; + $self->_shutdown; + return; + } } my $err = Net::SSLeay::get_error ($self->{tls}, -1); if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { if ($err == Net::SSLeay::ERROR_SYSCALL ()) { - $self->error; + return $self->_error ($!, 1); } elsif ($err == Net::SSLeay::ERROR_SSL ()) { - $! = &Errno::EIO; - $self->error; + return $self->_error (&Errno::EIO, 1); } # all others are fine for our purposes @@ -1125,7 +1139,6 @@ =cut -# TODO: maybe document... sub starttls { my ($self, $ssl, $ctx) = @_;