[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.190 by root, Thu Nov 26 10:30:24 2009 UTC

 =head1 NAME
 AnyEvent::Handle - non-blocking I/O on file 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
 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...
 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 {
    $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)
    };
 }
 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
    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
 #############################################################################
    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
    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) {

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.190 by root, Thu Nov 26 10:30:24 2009 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.190 by root, Thu Nov 26 10:30:24 2009 UTC