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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.52 by root, Mon Jun 2 09:10:38 2008 UTC vs.
Revision 1.64 by root, Fri Jun 6 11:01:17 2008 UTC

 AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
 =cut
-our $VERSION = 4.1;
+our $VERSION = 4.15;
 =head1 SYNOPSIS
    use AnyEvent;
    use AnyEvent::Handle;
 C<croak>.
 =item on_read => $cb->($handle)
 This sets the default read callback, which is called when data arrives
-and no read request is in the queue.
+and no read request is in the queue (unlike read queue callbacks, this
+callback will only be called when at least one octet of data is in the
+read buffer).
 To access (and remove data from) the read buffer, use the C<< ->rbuf >>
 method or access the C<$handle->{rbuf}> member directly.
 When an EOF condition is detected then AnyEvent::Handle will first try to
 Sets the amount of bytes (default: C<0>) that make up an "empty" write
 buffer: If the write reaches this size or gets even samller it is
 considered empty.
+=item linger => <seconds>
+If non-zero (default: C<3600>), then the destructor of the
+AnyEvent::Handle object will check wether there is still outstanding write
+data and will install a watcher that will write out this data. No errors
+will be reported (this mostly matches how the operating system treats
+outstanding data at socket close time).
+This will not work for partial TLS data that could not yet been
+encoded. This data will be lost.
 =item tls => "accept" | "connect" | Net::SSLeay::SSL object
 When this parameter is given, it enables TLS (SSL) mode, that means it
 will start making tls handshake and will transparently encrypt/decrypt
 data.
    if ($self->{tls}) {
       require Net::SSLeay;
       $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
 }
 sub _shutdown {
             $self->{on_timeout}($self);
          } else {
             $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
          $after = $self->{timeout};
       }
       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;
       });
    my ($self, $string) = @_;
    sprintf "%d:%s,", (length $string), $string
 };
+=item packstring => $format, $data
+An octet string prefixed with an encoded length. The encoding C<$format>
+uses the same format as a Perl C<pack> format, but must specify a single
+integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
+optional C<!>, C<< < >> or C<< > >> modifier).
+=cut
+register_write_type packstring => sub {
+   my ($self, $format, $string) = @_;
+   pack "$format/a", $string
+};
 =item json => $array_or_hashref
 Encodes the given hash or array reference into a JSON object. Unless you
 provide your own JSON object, this means it will be encoded to JSON text
 in UTF-8.
    $self->{json} ? $self->{json}->encode ($ref)
                  : JSON::encode_json ($ref)
 };
+=item storable => $reference
+Freezes the given reference using L<Storable> and writes it to the
+handle. Uses the C<nfreeze> format.
+=cut
+register_write_type storable => sub {
+   my ($self, $ref) = @_;
+   require Storable;
+   pack "w/a", Storable::nfreeze ($ref)
+};
+=back
 =item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
 This function (not method) lets you add your own types to C<push_write>.
 Whenever the given C<type> is used, C<push_write> will invoke the code
 reference with the handle object and the remaining arguments.
 The code reference is supposed to return a single octet string that will
 be appended to the write buffer.
 Note that this is a function, and all types registered this way will be
 global, so try to use unique names.
-=back
 =cut
 #############################################################################
 enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
 or not.
 In the more complex case, you want to queue multiple callbacks. In this
 case, AnyEvent::Handle will call the first queued callback each time new
-data arrives and removes it when it has done its job (see C<push_read>,
+data arrives (also the first time it is queued) and removes it when it has
-below).
+done its job (see C<push_read>, below).
 This way you can, for example, push three line-reads, followed by reading
 a chunk of data, and AnyEvent::Handle will execute them in order.
 Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
 =cut
 sub _drain_rbuf {
    my ($self) = @_;
+   local $self->{_in_drain} = 1;
    if (
       defined $self->{rbuf_max}
       && $self->{rbuf_max} < length $self->{rbuf}
    ) {
       return $self->_error (&Errno::ENOSPC, 1);
    }
-   return if $self->{in_drain};
+   while () {
-   local $self->{in_drain} = 1;
-   while (my $len = length $self->{rbuf}) {
       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)
-               return $self->_error (&Errno::EPIPE, 1);
+               $self->_error (&Errno::EPIPE, 1), last;
             }
             unshift @{ $self->{_queue} }, $cb;
-            return;
+            last;
          }
       } elsif ($self->{on_read}) {
+         last unless $len;
          $self->{on_read}($self);
          if (
-            $self->{_eof}                   # if no further data will arrive
-            && $len == length $self->{rbuf} # and no data has been consumed
+            $len == length $self->{rbuf} # if no data has been consumed
-            && !@{ $self->{_queue} }        # and the queue is still empty
+            && !@{ $self->{_queue} }     # and the queue is still empty
-            && $self->{on_read}             # and we still want to read data
+            && $self->{on_read}          # but we still have on_read
          ) {
+            # no further data will arrive
-            # then no progress can be made
+            # so no progress can be made
-            return $self->_error (&Errno::EPIPE, 1);
+            $self->_error (&Errno::EPIPE, 1), last
+               if $self->{_eof};
+            last; # more data might arrive
          }
       } else {
          # read side becomes idle
          delete $self->{_rw};
-         return;
+         last;
       }
    }
    $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} };
+   }
 }
 =item $handle->on_read ($cb)
 This replaces the currently set C<on_read> callback, or clears it (when
 sub on_read {
    my ($self, $cb) = @_;
    $self->{on_read} = $cb;
+   $self->_drain_rbuf if $cb && !$self->{_in_drain};
 }
 =item $handle->rbuf
 Returns the read buffer (as a modifiable lvalue).
       $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
             ->($self, $cb, @_);
    }
    push @{ $self->{_queue} }, $cb;
-   $self->_drain_rbuf;
+   $self->_drain_rbuf unless $self->{_in_drain};
 }
 sub unshift_read {
    my $self = shift;
    my $cb = pop;
             ->($self, $cb, @_);
    }
    unshift @{ $self->{_queue} }, $cb;
-   $self->_drain_rbuf;
+   $self->_drain_rbuf unless $self->{_in_drain};
 }
 =item $handle->push_read (type => @args, $cb)
 =item $handle->unshift_read (type => @args, $cb)
 sub unshift_read_line {
    my $self = shift;
    $self->unshift_read (line => @_);
 }
-=item netstring => $cb->($handle, $string)
-A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
-Throws an error with C<$!> set to EBADMSG on format violations.
-=cut
-register_read_type netstring => sub {
-   my ($self, $cb) = @_;
-   sub {
-      unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
-         if ($_[0]{rbuf} =~ /[^0-9]/) {
-            $self->_error (&Errno::EBADMSG);
-         }
-         return;
-      }
-      my $len = $1;
-      $self->unshift_read (chunk => $len, sub {
-         my $string = $_[1];
-         $_[0]->unshift_read (chunk => 1, sub {
-            if ($_[1] eq ",") {
-               $cb->($_[0], $string);
-            } else {
-               $self->_error (&Errno::EBADMSG);
-            }
-         });
-      });
-      1
-   }
-};
 =item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
 Makes a regex match against the regex object C<$accept> and returns
 everything up to and including the match.
       ()
    }
 };
+=item netstring => $cb->($handle, $string)
+A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
+Throws an error with C<$!> set to EBADMSG on format violations.
+=cut
+register_read_type netstring => sub {
+   my ($self, $cb) = @_;
+   sub {
+      unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
+         if ($_[0]{rbuf} =~ /[^0-9]/) {
+            $self->_error (&Errno::EBADMSG);
+         }
+         return;
+      }
+      my $len = $1;
+      $self->unshift_read (chunk => $len, sub {
+         my $string = $_[1];
+         $_[0]->unshift_read (chunk => 1, sub {
+            if ($_[1] eq ",") {
+               $cb->($_[0], $string);
+            } else {
+               $self->_error (&Errno::EBADMSG);
+            }
+         });
+      });
+      1
+   }
+};
+=item packstring => $format, $cb->($handle, $string)
+An octet string prefixed with an encoded length. The encoding C<$format>
+uses the same format as a Perl C<pack> format, but must specify a single
+integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
+optional C<!>, C<< < >> or C<< > >> modifier).
+DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
+Example: read a block of data prefixed by its length in BER-encoded
+format (very efficient).
+   $handle->push_read (packstring => "w", sub {
+      my ($handle, $data) = @_;
+   });
+=cut
+register_read_type packstring => sub {
+   my ($self, $cb, $format) = @_;
+   sub {
+      # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
+      defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
+         or return;
+      # remove prefix
+      substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
+      # read rest
+      $_[0]->unshift_read (chunk => $len, $cb);
+      1
+   }
+};
 =item json => $cb->($handle, $hash_or_arrayref)
 Reads a JSON object or array, decodes it and passes it to the callback.
 If a C<json> object was passed to the constructor, then that will be used
 the C<json> write type description, above, for an actual example.
 =cut
 register_read_type json => sub {
-   my ($self, $cb, $accept, $reject, $skip) = @_;
+   my ($self, $cb) = @_;
    require JSON;
    my $data;
    my $rbuf = \$self->{rbuf};
          ()
       }
    }
 };
+=item storable => $cb->($handle, $ref)
+Deserialises a L<Storable> frozen representation as written by the
+C<storable> write type (BER-encoded length prefix followed by nfreeze'd
+data).
+Raises C<EBADMSG> error if the data could not be decoded.
+=cut
+register_read_type storable => sub {
+   my ($self, $cb) = @_;
+   require Storable;
+   sub {
+      # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
+      defined (my $len = eval { unpack "w", $_[0]->{rbuf} })
+         or return;
+      # remove prefix
+      substr $_[0]->{rbuf}, 0, (length pack "w", $len), "";
+      # read rest
+      $_[0]->unshift_read (chunk => $len, sub {
+         if (my $ref = eval { Storable::thaw ($_[1]) }) {
+            $cb->($_[0], $ref);
+         } else {
+            $self->_error (&Errno::EBADMSG);
+         }
+      });
+   }
+};
 =back
 =item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
 This function (not method) lets you add your own types to C<push_read>.
 =item $handle->stop_read
 =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<stop_read>. Neither C<on_read> no
+socket. In this case you can call C<stop_read>. Neither C<on_read> nor
 any queued callbacks will be executed then. To start reading again, call
 C<start_read>.
+Note that AnyEvent::Handle will automatically C<start_read> for you when
+you change the C<on_read> callback or push/unshift a read callback, and it
+will automatically C<stop_read> for you when neither C<on_read> is set nor
+there are any read requests in the queue.
 =cut
 sub stop_read {
    my ($self) = @_;
          if ($len > 0) {
             $self->{_activity} = AnyEvent->now;
             $self->{filter_r}
                ? $self->{filter_r}($self, $rbuf)
-               : $self->_drain_rbuf;
+               : $self->{_in_drain} || $self->_drain_rbuf;
          } elsif (defined $len) {
             delete $self->{_rw};
             $self->{_eof} = 1;
-            $self->_drain_rbuf;
+            $self->_drain_rbuf unless $self->{_in_drain};
          } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
             return $self->_error ($!, 1);
          }
       });
 }
 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}))) {
+   while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
+      if (length $buf) {
-      $self->{rbuf} .= $buf;
+         $self->{rbuf} .= $buf;
-      $self->_drain_rbuf;
+         $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 ()) {
 sub DESTROY {
    my $self = shift;
    $self->stoptls;
+   my $linger = exists $self->{linger} ? $self->{linger} : 3600;
+   if ($linger && length $self->{wbuf}) {
+      my $fh   = delete $self->{fh};
+      my $wbuf = delete $self->{wbuf};
+      my @linger;
+      push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => 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 {
+         @linger = ();
+      });
+   }
 }
 =item AnyEvent::Handle::TLS_CTX
 This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents): Revision 1.52 by root, Mon Jun 2 09:10:38 2008 UTC vs. Revision 1.64 by root, Fri Jun 6 11:01:17 2008 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.52 by root, Mon Jun 2 09:10:38 2008 UTC vs.
Revision 1.64 by root, Fri Jun 6 11:01:17 2008 UTC