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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.176 by root, Sun Aug 9 00:20:35 2009 UTC vs.
Revision 1.194 by root, Thu May 20 21:22:21 2010 UTC

 =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.92;
 =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
 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.
 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.
+   if (Socket::SOCK_STREAM != unpack "I", getsockopt $self->{fh}, Socket::SOL_SOCKET (), Socket::SO_TYPE ()) {
+      Carp::croak "AnyEvent::Handle: only stream sockets supported, anything else will NOT work!";
+   }
    AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
    $self->{_activity}  =
    $self->{_ractivity} =
    $self->{_wactivity} = AE::now;
-   $self->timeout  (delete $self->{timeout} ) if $self->{timeout};
+   $self->timeout   (delete $self->{timeout}  ) if $self->{timeout};
-   $self->rtimeout (delete $self->{rtimeout}) if $self->{rtimeout};
+   $self->rtimeout  (delete $self->{rtimeout} ) if $self->{rtimeout};
-   $self->wtimeout (delete $self->{wtimeout}) if $self->{wtimeout};
+   $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";
    }
 }
 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)
 =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) = @_;
 =item $handle->push_write ($data)
 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) = @_;
    };
 }
 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
 #############################################################################
 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} = 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;
          && ($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) {
 =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#
 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.176 by root, Sun Aug 9 00:20:35 2009 UTC vs. Revision 1.194 by root, Thu May 20 21:22:21 2010 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.176 by root, Sun Aug 9 00:20:35 2009 UTC vs.
Revision 1.194 by root, Thu May 20 21:22:21 2010 UTC