[ViewVC] Diff of: cvs/AnyEvent/lib/AnyEvent/Handle.pm

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.165 by root, Mon Jul 27 22:49:23 2009 UTC vs.
Revision 1.196 by root, Tue Jun 8 10:04:17 2010 UTC

-package AnyEvent::Handle;
-use Scalar::Util ();
-use Carp ();
-use Errno qw(EAGAIN EINTR);
-use AnyEvent (); BEGIN { AnyEvent::common_sense }
-use AnyEvent::Util qw(WSAEWOULDBLOCK);
 =head1 NAME
-AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
+AnyEvent::Handle - non-blocking I/O on streaming handles via AnyEvent
-=cut
-our $VERSION = 4.87;
 =head1 SYNOPSIS
    use AnyEvent;
    use AnyEvent::Handle;
       on_error => sub {
          my ($hdl, $fatal, $msg) = @_;
          warn "got error $msg\n";
          $hdl->destroy;
          $cv->send;
-      );
+      };
    # send some request line
    $hdl->push_write ("getinfo\015\012");
    # read the response line
    $cv->recv;
 =head1 DESCRIPTION
 This module is a helper module to make it easier to do event-based I/O on
-filehandles.
+stream-based filehandles (sockets, pipes or other stream things).
 The L<AnyEvent::Intro> tutorial contains some well-documented
 AnyEvent::Handle examples.
 In the following, when the documentation refers to of "bytes" then this
 C<on_error> callback.
 All callbacks will be invoked with the handle object as their first
 argument.
+=cut
+package AnyEvent::Handle;
+use Scalar::Util ();
+use List::Util ();
+use Carp ();
+use Errno qw(EAGAIN EINTR);
+use AnyEvent (); BEGIN { AnyEvent::common_sense }
+use AnyEvent::Util qw(WSAEWOULDBLOCK);
+our $VERSION = $AnyEvent::VERSION;
+sub _load_func($) {
+   my $func = $_[0];
+   unless (defined &$func) {
+      my $pkg = $func;
+      do {
+         $pkg =~ s/::[^:]+$//
+            or return;
+         eval "require $pkg";
+      } until defined &$func;
+   }
+   \&$func
+}
 =head1 METHODS
 =over 4
-=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...
+=item $handle = B<new> AnyEvent::Handle fh => $filehandle, key => value...
 The constructor supports these arguments (all as C<< key => value >> pairs).
 =over 4
 The actual numeric host and port (the socket peername) are passed as
 parameters, together with a retry callback.
 When, for some reason, the handle is not acceptable, then calling
-C<$retry> will continue with the next conenction target (in case of
+C<$retry> will continue with the next connection target (in case of
 multi-homed hosts or SRV records there can be multiple connection
-endpoints). When it is called then the read and write queues, eof status,
+endpoints). At the time it is called the read and write queues, eof
-tls status and similar properties of the handle are being reset.
+status, tls status and similar properties of the handle will have been
+reset.
 In most cases, ignoring the C<$retry> parameter is the way to go.
 =item on_connect_error => $cb->($handle, $message)
-This callback is called when the conenction could not be
+This callback is called when the connection could not be
 established. C<$!> will contain the relevant error code, and C<$message> a
 message describing it (usually the same as C<"$!">).
 If this callback isn't specified, then C<on_error> will be called with a
 fatal error instead.
 memory and push it into the queue, but instead only read more data from
 the file when the write queue becomes empty.
 =item timeout => $fractional_seconds
+=item rtimeout => $fractional_seconds
+=item wtimeout => $fractional_seconds
-If non-zero, then this enables an "inactivity" timeout: whenever this many
+If non-zero, then these enables an "inactivity" timeout: whenever this
-seconds pass without a successful read or write on the underlying file
+many seconds pass without a successful read or write on the underlying
-handle, the C<on_timeout> callback will be invoked (and if that one is
+file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
-missing, a non-fatal C<ETIMEDOUT> error will be raised).
+will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
+error will be raised).
+There are three variants of the timeouts that work fully independent
+of each other, for both read and write, just read, and just write:
+C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
+C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
+C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
 Note that timeout processing is also active when you currently do not have
 any outstanding read or write requests: If you plan to keep the connection
 idle then you should disable the timout temporarily or ignore the timeout
 in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
 accomplishd by setting this option to a true value.
 The default is your opertaing system's default behaviour (most likely
 enabled), this option explicitly enables or disables it, if possible.
+=item keepalive => <boolean>
+Enables (default disable) the SO_KEEPALIVE option on the stream socket:
+normally, TCP connections have no time-out once established, so TCP
+connections, once established, can stay alive forever even when the other
+side has long gone. TCP keepalives are a cheap way to take down long-lived
+TCP connections whent he other side becomes unreachable. While the default
+is OS-dependent, TCP keepalives usually kick in after around two hours,
+and, if the other side doesn't reply, take down the TCP connection some 10
+to 15 minutes later.
+It is harmless to specify this option for file handles that do not support
+keepalives, and enabling it on connections that are potentially long-lived
+is usually a good idea.
+=item oobinline => <boolean>
+BSD majorly fucked up the implementation of TCP urgent data. The result
+is that almost no OS implements TCP according to the specs, and every OS
+implements it slightly differently.
+If you want to handle TCP urgent data, then setting this flag (the default
+is enabled) gives you the most portable way of getting urgent data, by
+putting it into the stream.
+Since BSD emulation of OOB data on top of TCP's urgent data can have
+security implications, AnyEvent::Handle sets this flag automatically
+unless explicitly specified. Note that setting this flag after
+establishing a connection I<may> be a bit too late (data loss could
+already have occured on BSD systems), but at least it will protect you
+from most attacks.
 =item read_size => <bytes>
 The default read block size (the amount of bytes this module will
 try to read during each loop iteration, which affects memory
 requirements). Default: C<8192>.
 C<undef>.
 =item tls => "accept" | "connect" | Net::SSLeay::SSL object
 When this parameter is given, it enables TLS (SSL) mode, that means
-AnyEvent will start a TLS handshake as soon as the conenction has been
+AnyEvent will start a TLS handshake as soon as the connection has been
 established and will transparently encrypt/decrypt data afterwards.
 All TLS protocol errors will be signalled as C<EPROTO>, with an
 appropriate error message.
                      delete $self->{_skip_drain_rbuf};
                      $self->_start;
                      $self->{on_connect}
                         and $self->{on_connect}($self, $host, $port, sub {
-                               delete @$self{qw(fh _tw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
+                               delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
                                $self->{_skip_drain_rbuf} = 1;
                                &$retry;
                             });
                   } else {
 }
 sub _start {
    my ($self) = @_;
+   # too many clueless people try to use udp and similar sockets
+   # with AnyEvent::Handle, do them a favour.
+   my $type = getsockopt $self->{fh}, Socket::SOL_SOCKET (), Socket::SO_TYPE ();
+   Carp::croak "AnyEvent::Handle: only stream sockets supported, anything else will NOT work!"
+      if Socket::SOCK_STREAM () != (unpack "I", $type) && defined $type;
    AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
+   $self->{_activity}  =
+   $self->{_ractivity} =
-   $self->{_activity} = AnyEvent->now;
+   $self->{_wactivity} = AE::now;
-   $self->_timeout;
+   $self->timeout   (delete $self->{timeout}  ) if $self->{timeout};
+   $self->rtimeout  (delete $self->{rtimeout} ) if $self->{rtimeout};
+   $self->wtimeout  (delete $self->{wtimeout} ) if $self->{wtimeout};
-   $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
+   $self->no_delay  (delete $self->{no_delay} ) if exists $self->{no_delay}  && $self->{no_delay};
+   $self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive} && $self->{keepalive};
+   $self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
-   $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
+   $self->starttls  (delete $self->{tls}, delete $self->{tls_ctx})
       if $self->{tls};
-   $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
+   $self->on_drain  (delete $self->{on_drain}) if $self->{on_drain};
    $self->start_read
       if $self->{on_read} || @{ $self->{_queue} };
    $self->_drain_wbuf;
 }
-#sub _shutdown {
-#   my ($self) = @_;
-#
-#   delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
-#   $self->{_eof} = 1; # tell starttls et. al to stop trying
-#
-#   &_freetls;
-#}
 sub _error {
    my ($self, $errno, $fatal, $message) = @_;
    $! = $errno;
    $message ||= "$!";
    if ($self->{on_error}) {
       $self->{on_error}($self, $fatal, $message);
       $self->destroy if $fatal;
-   } elsif ($self->{fh}) {
+   } elsif ($self->{fh} || $self->{connect}) {
       $self->destroy;
       Carp::croak "AnyEvent::Handle uncaught error: $message";
    }
 }
    $_[0]{on_eof} = $_[1];
 }
 =item $handle->on_timeout ($cb)
-Replace the current C<on_timeout> callback, or disables the callback (but
+=item $handle->on_rtimeout ($cb)
-not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
-argument and method.
-=cut
+=item $handle->on_wtimeout ($cb)
-sub on_timeout {
+Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
-   $_[0]{on_timeout} = $_[1];
+callback, or disables the callback (but not the timeout) if C<$cb> =
-}
+C<undef>. See the C<timeout> constructor argument and method.
+=cut
+# see below
 =item $handle->autocork ($boolean)
 Enables or disables the current autocork behaviour (see C<autocork>
 constructor argument). Changes will only take effect on the next write.
 sub no_delay {
    $_[0]{no_delay} = $_[1];
    eval {
       local $SIG{__DIE__};
-      setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1]
+      setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
          if $_[0]{fh};
    };
 }
+=item $handle->keepalive ($boolean)
+Enables or disables the C<keepalive> setting (see constructor argument of
+the same name for details).
+=cut
+sub keepalive {
+   $_[0]{keepalive} = $_[1];
+   eval {
+      local $SIG{__DIE__};
+      setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
+         if $_[0]{fh};
+   };
+}
+=item $handle->oobinline ($boolean)
+Enables or disables the C<oobinline> setting (see constructor argument of
+the same name for details).
+=cut
+sub oobinline {
+   $_[0]{oobinline} = $_[1];
+   eval {
+      local $SIG{__DIE__};
+      setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
+         if $_[0]{fh};
+   };
+}
+=item $handle->keepalive ($boolean)
+Enables or disables the C<keepalive> setting (see constructor argument of
+the same name for details).
+=cut
+sub keepalive {
+   $_[0]{keepalive} = $_[1];
+   eval {
+      local $SIG{__DIE__};
+      setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
+         if $_[0]{fh};
+   };
+}
 =item $handle->on_starttls ($cb)
 Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
 =cut
 Replace the current C<on_stoptls> callback (see the C<on_stoptls> constructor argument).
 =cut
-sub on_starttls {
+sub on_stoptls {
    $_[0]{on_stoptls} = $_[1];
 }
+=item $handle->rbuf_max ($max_octets)
+Configures the C<rbuf_max> setting (C<undef> disables it).
+=cut
+sub rbuf_max {
+   $_[0]{rbuf_max} = $_[1];
+}
 #############################################################################
 =item $handle->timeout ($seconds)
+=item $handle->rtimeout ($seconds)
+=item $handle->wtimeout ($seconds)
 Configures (or disables) the inactivity timeout.
-=cut
+=item $handle->timeout_reset
-sub timeout {
+=item $handle->rtimeout_reset
+=item $handle->wtimeout_reset
+Reset the activity timeout, as if data was received or sent.
+These methods are cheap to call.
+=cut
+for my $dir ("", "r", "w") {
+   my $timeout    = "${dir}timeout";
+   my $tw         = "_${dir}tw";
+   my $on_timeout = "on_${dir}timeout";
+   my $activity   = "_${dir}activity";
+   my $cb;
+   *$on_timeout = sub {
+      $_[0]{$on_timeout} = $_[1];
+   };
+   *$timeout = sub {
-   my ($self, $timeout) = @_;
+      my ($self, $new_value) = @_;
-   $self->{timeout} = $timeout;
+      $self->{$timeout} = $new_value;
-   $self->_timeout;
+      delete $self->{$tw}; &$cb;
-}
+   };
+   *{"${dir}timeout_reset"} = sub {
+      $_[0]{$activity} = AE::now;
+   };
+   # main workhorse:
-# reset the timeout watcher, as neccessary
+   # reset the timeout watcher, as neccessary
-# also check for time-outs
+   # also check for time-outs
-sub _timeout {
+   $cb = sub {
-   my ($self) = @_;
+      my ($self) = @_;
-   if ($self->{timeout} && $self->{fh}) {
+      if ($self->{$timeout} && $self->{fh}) {
-      my $NOW = AnyEvent->now;
+         my $NOW = AE::now;
-      # when would the timeout trigger?
+         # when would the timeout trigger?
-      my $after = $self->{_activity} + $self->{timeout} - $NOW;
+         my $after = $self->{$activity} + $self->{$timeout} - $NOW;
-      # now or in the past already?
+         # now or in the past already?
-      if ($after <= 0) {
+         if ($after <= 0) {
-         $self->{_activity} = $NOW;
+            $self->{$activity} = $NOW;
-         if ($self->{on_timeout}) {
+            if ($self->{$on_timeout}) {
-            $self->{on_timeout}($self);
+               $self->{$on_timeout}($self);
-         } else {
+            } else {
-            $self->_error (Errno::ETIMEDOUT);
+               $self->_error (Errno::ETIMEDOUT);
+            }
+            # callback could have changed timeout value, optimise
+            return unless $self->{$timeout};
+            # calculate new after
+            $after = $self->{$timeout};
          }
-         # callback could have changed timeout value, optimise
+         Scalar::Util::weaken $self;
-         return unless $self->{timeout};
+         return unless $self; # ->error could have destroyed $self
-         # calculate new after
+         $self->{$tw} ||= AE::timer $after, 0, sub {
-         $after = $self->{timeout};
+            delete $self->{$tw};
+            $cb->($self);
+         };
+      } else {
+         delete $self->{$tw};
       }
-      Scalar::Util::weaken $self;
-      return unless $self; # ->error could have destroyed $self
-      $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
-         delete $self->{_tw};
-         $self->_timeout;
-      });
-   } else {
-      delete $self->{_tw};
    }
 }
 #############################################################################
 =item $handle->on_drain ($cb)
 Sets the C<on_drain> callback or clears it (see the description of
 C<on_drain> in the constructor).
+This method may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 sub on_drain {
    my ($self, $cb) = @_;
 Queues the given scalar to be written. You can push as much data as you
 want (only limited by the available memory), as C<AnyEvent::Handle>
 buffers it independently of the kernel.
+This method may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 sub _drain_wbuf {
    my ($self) = @_;
          my $len = syswrite $self->{fh}, $self->{wbuf};
          if (defined $len) {
             substr $self->{wbuf}, 0, $len, "";
-            $self->{_activity} = AnyEvent->now;
+            $self->{_activity} = $self->{_wactivity} = AE::now;
             $self->{on_drain}($self)
                if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
                   && $self->{on_drain};
       # try to write data immediately
       $cb->() unless $self->{autocork};
       # if still data left in wbuf, we need to poll
-      $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
+      $self->{_ww} = AE::io $self->{fh}, 1, $cb
          if length $self->{wbuf};
    };
 }
 our %WH;
+# deprecated
 sub register_write_type($$) {
    $WH{$_[0]} = $_[1];
 }
 sub push_write {
    my $self = shift;
    if (@_ > 1) {
       my $type = shift;
+      @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
-      @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
+            or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_write")
            ->($self, @_);
    }
+   # we downgrade here to avoid hard-to-track-down bugs,
+   # and diagnose the problem earlier and better.
    if ($self->{tls}) {
-      $self->{_tls_wbuf} .= $_[0];
+      utf8::downgrade $self->{_tls_wbuf} .= $_[0];
       &_dotls ($self)    if $self->{fh};
    } else {
-      $self->{wbuf}      .= $_[0];
+      utf8::downgrade $self->{wbuf}      .= $_[0];
       $self->_drain_wbuf if $self->{fh};
    }
 }
 =item $handle->push_write (type => @args)
-Instead of formatting your data yourself, you can also let this module do
+Instead of formatting your data yourself, you can also let this module
-the job by specifying a type and type-specific arguments.
+do the job by specifying a type and type-specific arguments. You
+can also specify the (fully qualified) name of a package, in which
+case AnyEvent tries to load the package and then expects to find the
+C<anyevent_read_type> function inside (see "custom write types", below).
 Predefined types are (if you have ideas for additional types, feel free to
 drop by and tell us):
 =over 4
 Other languages could read single lines terminated by a newline and pass
 this line into their JSON decoder of choice.
 =cut
+sub json_coder() {
+   eval { require JSON::XS; JSON::XS->new->utf8 }
+      || do { require JSON; JSON->new->utf8 }
+}
 register_write_type json => sub {
    my ($self, $ref) = @_;
-   require JSON;
+   my $json = $self->{json} ||= json_coder;
-   $self->{json} ? $self->{json}->encode ($ref)
+   $json->encode ($ref)
-                 : JSON::encode_json ($ref)
 };
 =item storable => $reference
 Freezes the given reference using L<Storable> and writes it to the
 the peer.
 You can rely on the normal read queue and C<on_eof> handling
 afterwards. This is the cleanest way to close a connection.
+This method may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 sub push_shutdown {
    my ($self) = @_;
    delete $self->{low_water_mark};
    $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
 }
-=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
+=item custom write types - Package::anyevent_write_type $handle, @args
-This function (not method) lets you add your own types to C<push_write>.
+Instead of one of the predefined types, you can also specify the name of
+a package. AnyEvent will try to load the package and then expects to find
+a function named C<anyevent_write_type> inside. If it isn't found, it
+progressively tries to load the parent package until it either finds the
+function (good) or runs out of packages (bad).
-Whenever the given C<type> is used, C<push_write> will invoke the code
+Whenever the given C<type> is used, C<push_write> will the function with
-reference with the handle object and the remaining arguments.
+the handle object and the remaining arguments.
-The code reference is supposed to return a single octet string that will
+The function is supposed to return a single octet string that will be
-be appended to the write buffer.
+appended to the write buffer, so you cna mentally treat this function as a
+"arguments to on-the-wire-format" converter.
-Note that this is a function, and all types registered this way will be
+Example: implement a custom write type C<join> that joins the remaining
-global, so try to use unique names.
+arguments using the first one.
+   $handle->push_write (My::Type => " ", 1,2,3);
+   # uses the following package, which can be defined in the "My::Type" or in
+   # the "My" modules to be auto-loaded, or just about anywhere when the
+   # My::Type::anyevent_write_type is defined before invoking it.
+   package My::Type;
+   sub anyevent_write_type {
+      my ($handle, $delim, @args) = @_;
+      join $delim, @args
+   }
 =cut
 #############################################################################
 sub _drain_rbuf {
    my ($self) = @_;
    # avoid recursion
-   return if exists $self->{_skip_drain_rbuf};
+   return if $self->{_skip_drain_rbuf};
    local $self->{_skip_drain_rbuf} = 1;
-   if (
-      defined $self->{rbuf_max}
-      && $self->{rbuf_max} < length $self->{rbuf}
-   ) {
-      $self->_error (Errno::ENOSPC, 1), return;
-   }
    while () {
       # we need to use a separate tls read buffer, as we must not receive data while
       # we are draining the buffer, and this can only happen with TLS.
       $self->{rbuf} .= delete $self->{_tls_rbuf}
          : $self->_error (0, 1, "Unexpected end-of-file");
       return;
    }
+   if (
+      defined $self->{rbuf_max}
+      && $self->{rbuf_max} < length $self->{rbuf}
+   ) {
+      $self->_error (Errno::ENOSPC, 1), return;
+   }
    # may need to restart read watcher
    unless ($self->{_rw}) {
       $self->start_read
          if $self->{on_read} || @{ $self->{_queue} };
    }
 =item $handle->on_read ($cb)
 This replaces the currently set C<on_read> callback, or clears it (when
 the new callback is C<undef>). See the description of C<on_read> in the
 constructor.
+This method may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 sub on_read {
    my ($self, $cb) = @_;
 If enough data was available, then the callback must remove all data it is
 interested in (which can be none at all) and return a true value. After returning
 true, it will be removed from the queue.
+These methods may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 our %RH;
 sub register_read_type($$) {
    my $cb = pop;
    if (@_) {
       my $type = shift;
+      $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
-      $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
+             or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
             ->($self, $cb, @_);
    }
    push @{ $self->{_queue} }, $cb;
    $self->_drain_rbuf;
       $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
             ->($self, $cb, @_);
    }
    unshift @{ $self->{_queue} }, $cb;
    $self->_drain_rbuf;
 }
 =item $handle->push_read (type => @args, $cb)
 =item $handle->unshift_read (type => @args, $cb)
 Instead of providing a callback that parses the data itself you can chose
 between a number of predefined parsing formats, for chunks of data, lines
-etc.
+etc. You can also specify the (fully qualified) name of a package, in
+which case AnyEvent tries to load the package and then expects to find the
+C<anyevent_read_type> function inside (see "custom read types", below).
 Predefined types are (if you have ideas for additional types, feel free to
 drop by and tell us):
 =over 4
 =cut
 register_read_type json => sub {
    my ($self, $cb) = @_;
-   my $json = $self->{json} ||=
+   my $json = $self->{json} ||= json_coder;
-      eval { require JSON::XS; JSON::XS->new->utf8 }
-         || do { require JSON; JSON->new->utf8 };
    my $data;
    my $rbuf = \$self->{rbuf};
    sub {
    }
 };
 =back
-=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
+=item custom read types - Package::anyevent_read_type $handle, $cb, @args
-This function (not method) lets you add your own types to C<push_read>.
+Instead of one of the predefined types, you can also specify the name
+of a package. AnyEvent will try to load the package and then expects to
+find a function named C<anyevent_read_type> inside. If it isn't found, it
+progressively tries to load the parent package until it either finds the
+function (good) or runs out of packages (bad).
-Whenever the given C<type> is used, C<push_read> will invoke the code
+Whenever this type is used, C<push_read> will invoke the function with the
-reference with the handle object, the callback and the remaining
+handle object, the original callback and the remaining arguments.
-arguments.
-The code reference is supposed to return a callback (usually a closure)
+The function is supposed to return a callback (usually a closure) that
-that works as a plain read callback (see C<< ->push_read ($cb) >>).
+works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
+mentally treat the function as a "configurable read type to read callback"
+converter.
-It should invoke the passed callback when it is done reading (remember to
+It should invoke the original callback when it is done reading (remember
-pass C<$handle> as first argument as all other callbacks do that).
+to pass C<$handle> as first argument as all other callbacks do that,
+although there is no strict requirement on this).
-Note that this is a function, and all types registered this way will be
-global, so try to use unique names.
-For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
+For examples, see the source of this module (F<perldoc -m
-search for C<register_read_type>)).
+AnyEvent::Handle>, search for C<register_read_type>)).
 =item $handle->stop_read
 =item $handle->start_read
 }
 sub start_read {
    my ($self) = @_;
-   unless ($self->{_rw} || $self->{_eof}) {
+   unless ($self->{_rw} || $self->{_eof} || !$self->{fh}) {
       Scalar::Util::weaken $self;
-      $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
+      $self->{_rw} = AE::io $self->{fh}, 0, sub {
          my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
          my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
          if ($len > 0) {
-            $self->{_activity} = AnyEvent->now;
+            $self->{_activity} = $self->{_ractivity} = AE::now;
             if ($self->{tls}) {
                Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
                &_dotls ($self);
             $self->_drain_rbuf;
          } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
             return $self->_error ($!, 1);
          }
-      });
+      };
    }
 }
 our $ERROR_SYSCALL;
 our $ERROR_WANT_READ;
          && ($tmp != $ERROR_SYSCALL || $!);
    while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
       $self->{wbuf} .= $tmp;
       $self->_drain_wbuf;
+      $self->{tls} or return; # tls session might have gone away in callback
    }
    $self->{_on_starttls}
       and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK ()
       and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established");
 Due to bugs in OpenSSL, it might or might not be possible to do multiple
 handshakes on the same stream. Best do not attempt to use the stream after
 stopping TLS.
+This method may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 our %TLS_CACHE; #TODO not yet documented, should we?
 sub starttls {
    require Net::SSLeay;
    $ERROR_SYSCALL   = Net::SSLeay::ERROR_SYSCALL     ();
    $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ   ();
-   $tls = $self->{tls};
+   $tls = delete $self->{tls};
    $ctx = $self->{tls_ctx};
    local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
    if ("HASH" eq ref $ctx) {
    Net::SSLeay::CTX_set_mode ($tls, 1|2);
    $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
    $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
+   Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
    Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
    $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
       if $self->{on_starttls};
 =item $handle->stoptls
 Shuts down the SSL connection - this makes a proper EOF handshake by
 sending a close notify to the other side, but since OpenSSL doesn't
-support non-blocking shut downs, it is not guarenteed that you can re-use
+support non-blocking shut downs, it is not guaranteed that you can re-use
 the stream afterwards.
+This method may invoke callbacks (and therefore the handle might be
+destroyed after it returns).
 =cut
 sub stoptls {
    my ($self) = @_;
-   if ($self->{tls}) {
+   if ($self->{tls} && $self->{fh}) {
       Net::SSLeay::shutdown ($self->{tls});
       &_dotls;
 #      # we don't give a shit. no, we do, but we can't. no...#d#
    my ($self) = @_;
    return unless $self->{tls};
    $self->{tls_ctx}->_put_session (delete $self->{tls})
-      if ref $self->{tls};
+      if $self->{tls} > 0;
    delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
 }
 sub DESTROY {
       my $fh   = delete $self->{fh};
       my $wbuf = delete $self->{wbuf};
       my @linger;
-      push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
+      push @linger, AE::io $fh, 1, sub {
          my $len = syswrite $fh, $wbuf, length $wbuf;
          if ($len > 0) {
             substr $wbuf, 0, $len, "";
          } else {
             @linger = (); # end
          }
-      });
+      };
-      push @linger, AnyEvent->timer (after => $linger, cb => sub {
+      push @linger, AE::timer $linger, 0, sub {
          @linger = ();
-      });
+      };
    }
 }
 =item $handle->destroy
 sub AnyEvent::Handle::destroyed::AUTOLOAD {
    #nop
 }
+=item $handle->destroyed
+Returns false as long as the handle hasn't been destroyed by a call to C<<
+->destroy >>, true otherwise.
+Can be useful to decide whether the handle is still valid after some
+callback possibly destroyed the handle. For example, C<< ->push_write >>,
+C<< ->starttls >> and other methods can call user callbacks, which in turn
+can destroy the handle, so work can be avoided by checking sometimes:
+   $hdl->starttls ("accept");
+   return if $hdl->destroyed;
+   $hdl->push_write (...
+Note that the call to C<push_write> will silently be ignored if the handle
+has been destroyed, so often you can just ignore the possibility of the
+handle being destroyed.
+=cut
+sub destroyed { 0 }
+sub AnyEvent::Handle::destroyed::destroyed { 1 }
 =item AnyEvent::Handle::TLS_CTX
 This function creates and returns the AnyEvent::TLS object used by default
 for TLS mode.

Diff Legend

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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents): Revision 1.165 by root, Mon Jul 27 22:49:23 2009 UTC vs. Revision 1.196 by root, Tue Jun 8 10:04:17 2010 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.165 by root, Mon Jul 27 22:49:23 2009 UTC vs.
Revision 1.196 by root, Tue Jun 8 10:04:17 2010 UTC