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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.21 by root, Sat May 24 15:03:42 2008 UTC vs.
Revision 1.34 by root, Mon May 26 03:27:52 2008 UTC

 no warnings;
 use strict;
 use AnyEvent ();
-use AnyEvent::Util ();
+use AnyEvent::Util qw(WSAWOULDBLOCK);
 use Scalar::Util ();
 use Carp ();
 use Fcntl ();
 use Errno qw/EAGAIN EINTR/;
 =head1 NAME
-AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent
+AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
-This module is experimental.
 =cut
 our $VERSION = '0.04';
    use AnyEvent;
    use AnyEvent::Handle;
    my $cv = AnyEvent->condvar;
-   my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
+   my $handle =
-   #TODO
-   # or use the constructor to pass the callback:
-   my $ae_fh2 =
       AnyEvent::Handle->new (
          fh => \*STDIN,
          on_eof => sub {
             $cv->broadcast;
          },
-         #TODO
       );
-   $cv->wait;
+   # send some request line
+   $handle->push_write ("getinfo\015\012");
+   # read the response line
+   $handle->push_read (line => sub {
+      my ($handle, $line) = @_;
+      warn "read line <$line>\n";
+      $cv->send;
+   });
+   $cv->recv;
 =head1 DESCRIPTION
 This module is a helper module to make it easier to do event-based I/O on
 filehandles. For utility functions for doing non-blocking connects and accepts
 The object will not be in a usable state when this callback has been
 called.
 On callback entrance, the value of C<$!> contains the operating system
-error (or C<ENOSPC> or C<EPIPE>).
+error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>).
 While not mandatory, it is I<highly> recommended to set this callback, as
 you will not be notified of errors otherwise. The default simply calls
 die.
 When this parameter is given, it enables TLS (SSL) mode, that means it
 will start making tls handshake and will transparently encrypt/decrypt
 data.
+TLS mode requires Net::SSLeay to be installed (it will be loaded
+automatically when you try to create a TLS handle).
 For the TLS server side, use C<accept>, and for the TLS client side of a
 connection, use C<connect> mode.
 You can also provide your own TLS connection object, but you have
 to make sure that you call either C<Net::SSLeay::set_connect_state>
 or C<Net::SSLeay::set_accept_state> on it before you pass it to
 AnyEvent::Handle.
+See the C<starttls> method if you need to start TLs negotiation later.
 =item tls_ctx => $ssl_ctx
 Use the given Net::SSLeay::CTX object to create the new TLS connection
 (unless a connection object was specified directly). If this parameter is
 missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
    }
    if ($self->{on_error}) {
       $self->{on_error}($self);
    } else {
-      die "AnyEvent::Handle uncaught fatal error: $!";
+      Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
    }
 }
 =item $fh = $handle->fh
-This method returns the filehandle of the L<AnyEvent::Handle> object.
+This method returns the file handle of the L<AnyEvent::Handle> object.
 =cut
 sub fh { $_[0]->{fh} }
 =cut
 sub _drain_wbuf {
    my ($self) = @_;
-   unless ($self->{ww}) {
+   if (!$self->{ww} && length $self->{wbuf}) {
       Scalar::Util::weaken $self;
       my $cb = sub {
          my $len = syswrite $self->{fh}, $self->{wbuf};
-         if ($len > 0) {
+         if ($len >= 0) {
             substr $self->{wbuf}, 0, $len, "";
             $self->{on_drain}($self)
                if $self->{low_water_mark} >= length $self->{wbuf}
                   && $self->{on_drain};
             delete $self->{ww} unless length $self->{wbuf};
-         } elsif ($! != EAGAIN && $! != EINTR) {
+         } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
             $self->error;
          }
       };
       $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
       $cb->($self);
    };
 }
+our %WH;
+sub register_write_type($$) {
+   $WH{$_[0]} = $_[1];
+}
 sub push_write {
    my $self = shift;
+   if (@_ > 1) {
+      my $type = shift;
+      @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
+           ->($self, @_);
+   }
    if ($self->{filter_w}) {
       $self->{filter_w}->($self, \$_[0]);
    } else {
       $self->{wbuf} .= $_[0];
       $self->_drain_wbuf;
    }
 }
+=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.
+Predefined types are (if you have ideas for additional types, feel free to
+drop by and tell us):
+=over 4
+=item netstring => $string
+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 {
+   my ($self, $string) = @_;
+   sprintf "%d:%s,", (length $string), $string
+};
+=item AnyEvent::Handle::register_write_type type => $coderef->($self, @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.
+=cut
 #############################################################################
 =back
    local $self->{in_drain} = 1;
    while (my $len = length $self->{rbuf}) {
       no strict 'refs';
       if (my $cb = shift @{ $self->{queue} }) {
-         if (!$cb->($self)) {
+         unless ($cb->($self)) {
             if ($self->{eof}) {
                # no progress can be made (not enough data and no data forthcoming)
                $! = &Errno::EPIPE; return $self->error;
             }
 interested in (which can be none at all) and return a true value. After returning
 true, it will be removed from the queue.
 =cut
+our %RH;
+sub register_read_type($$) {
+   $RH{$_[0]} = $_[1];
+}
 sub push_read {
-   my ($self, $cb) = @_;
+   my $self = shift;
+   my $cb = pop;
+   if (@_) {
+      my $type = shift;
+      $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
+            ->($self, $cb, @_);
+   }
    push @{ $self->{queue} }, $cb;
    $self->_drain_rbuf;
 }
 sub unshift_read {
-   my ($self, $cb) = @_;
+   my $self = shift;
+   my $cb = pop;
+   if (@_) {
+      my $type = shift;
+      $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
+            ->($self, $cb, @_);
+   }
-   push @{ $self->{queue} }, $cb;
+   unshift @{ $self->{queue} }, $cb;
    $self->_drain_rbuf;
 }
-=item $handle->push_read_chunk ($len, $cb->($self, $data))
+=item $handle->push_read (type => @args, $cb)
-=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
+=item $handle->unshift_read (type => @args, $cb)
-Append the given callback to the end of the queue (C<push_read_chunk>) or
+Instead of providing a callback that parses the data itself you can chose
-prepend it (C<unshift_read_chunk>).
+between a number of predefined parsing formats, for chunks of data, lines
+etc.
-The callback will be called only once C<$len> bytes have been read, and
+Predefined types are (if you have ideas for additional types, feel free to
-these C<$len> bytes will be passed to the callback.
+drop by and tell us):
-=cut
+=over 4
-sub _read_chunk($$) {
+=item chunk => $octets, $cb->($self, $data)
+Invoke the callback only once C<$octets> bytes have been read. Pass the
+data read to the callback. The callback will never be called with less
+data.
+Example: read 2 bytes.
+   $handle->push_read (chunk => 2, sub {
+      warn "yay ", unpack "H*", $_[1];
+   });
+=cut
+register_read_type chunk => sub {
-   my ($self, $len, $cb) = @_;
+   my ($self, $cb, $len) = @_;
    sub {
       $len <= length $_[0]{rbuf} or return;
       $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
       1
    }
-}
+};
+# compatibility with older API
 sub push_read_chunk {
-   $_[0]->push_read (&_read_chunk);
+   $_[0]->push_read (chunk => $_[1], $_[2]);
 }
 sub unshift_read_chunk {
-   $_[0]->unshift_read (&_read_chunk);
+   $_[0]->unshift_read (chunk => $_[1], $_[2]);
 }
-=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
+=item line => [$eol, ]$cb->($self, $line, $eol)
-=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
-Append the given callback to the end of the queue (C<push_read_line>) or
-prepend it (C<unshift_read_line>).
 The callback will be called only once a full line (including the end of
 line marker, C<$eol>) has been read. This line (excluding the end of line
 marker) will be passed to the callback as second argument (C<$line>), and
 the end of line marker as the third argument (C<$eol>).
 Partial lines at the end of the stream will never be returned, as they are
 not marked by the end of line marker.
 =cut
-sub _read_line($$) {
+register_read_type line => sub {
-   my $self = shift;
+   my ($self, $cb, $eol) = @_;
-   my $cb = pop;
-   my $eol = @_ ? shift : qr|(\015?\012)|;
-   my $pos;
+   $eol = qr|(\015?\012)| if @_ < 3;
    $eol = quotemeta $eol unless ref $eol;
    $eol = qr|^(.*?)($eol)|s;
    sub {
       $_[0]{rbuf} =~ s/$eol// or return;
       $cb->($_[0], $1, $2);
       1
    }
-}
+};
+# compatibility with older API
 sub push_read_line {
-   $_[0]->push_read (&_read_line);
+   my $self = shift;
+   $self->push_read (line => @_);
 }
 sub unshift_read_line {
-   $_[0]->unshift_read (&_read_line);
+   my $self = shift;
+   $self->unshift_read (line => @_);
 }
+=item netstring => $cb->($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]/) {
+            $! = &Errno::EBADMSG;
+            $self->error;
+         }
+         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 {
+               $! = &Errno::EBADMSG;
+               $self->error;
+            }
+         });
+      });
+      1
+   }
+};
+=back
+=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
+This function (not method) lets you add your own types to C<push_read>.
+Whenever the given C<type> is used, C<push_read> will invoke the code
+reference with the handle object, the callback and the remaining
+arguments.
+The code reference is supposed to return a callback (usually a closure)
+that works as a plain read callback (see C<< ->push_read ($cb) >>).
+It should invoke the passed callback when it is done reading (remember to
+pass C<$self> as first argument as all other callbacks do that).
+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>,
+search for C<register_read_type>)).
 =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
-any queued callbacks will be executed then. To start readign again, call
+any queued callbacks will be executed then. To start reading again, call
 C<start_read>.
 =cut
 sub stop_read {
          } elsif (defined $len) {
             delete $self->{rw};
             $self->{eof} = 1;
             $self->_drain_rbuf;
-         } elsif ($! != EAGAIN && $! != EINTR) {
+         } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
             return $self->error;
          }
       });
    }
 }
 sub _dotls {
    my ($self) = @_;
    if (length $self->{tls_wbuf}) {
-      my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf});
+      while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) {
-      substr $self->{tls_wbuf}, 0, $len, "" if $len > 0;
+         substr $self->{tls_wbuf}, 0, $len, "";
+      }
    }
    if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) {
       $self->{wbuf} .= $buf;
       $self->_drain_wbuf;
    }
-   if (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
+   while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
       $self->{rbuf} .= $buf;
       $self->_drain_rbuf;
-   } elsif (
+   }
-      (my $err = Net::SSLeay::get_error ($self->{tls}, -1))
+   my $err = Net::SSLeay::get_error ($self->{tls}, -1);
-      != Net::SSLeay::ERROR_WANT_READ ()
+   if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
-   ) {
       if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
          $self->error;
       } elsif ($err == Net::SSLeay::ERROR_SSL ())  {
          $! = &Errno::EIO;
          $self->error;
       # all others are fine for our purposes
    }
 }
+=item $handle->starttls ($tls[, $tls_ctx])
+Instead of starting TLS negotiation immediately when the AnyEvent::Handle
+object is created, you can also do that at a later time by calling
+C<starttls>.
+The first argument is the same as the C<tls> constructor argument (either
+C<"connect">, C<"accept"> or an existing Net::SSLeay object).
+The second argument is the optional C<Net::SSLeay::CTX> object that is
+used when AnyEvent::Handle has to create its own TLS connection object.
+=cut
 # TODO: maybe document...
 sub starttls {
    my ($self, $ssl, $ctx) = @_;
+   $self->stoptls;
    if ($ssl eq "accept") {
       $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
       Net::SSLeay::set_accept_state ($ssl);
    } elsif ($ssl eq "connect") {
    # basically, this is deep magic (because SSL_read should have the same issues)
    # but the openssl maintainers basically said: "trust us, it just works".
    # (unfortunately, we have to hardcode constants because the abysmally misdesigned
    # and mismaintained ssleay-module doesn't even offer them).
+   # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
    Net::SSLeay::CTX_set_mode ($self->{tls},
-      (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
+      (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
-      | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
+      | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
    $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
    $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
    Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});
       Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]});
       &_dotls;
    };
 }
+=item $handle->stoptls
+Destroys the SSL connection, if any. Partial read or write data will be
+lost.
+=cut
+sub stoptls {
+   my ($self) = @_;
+   Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
+   delete $self->{tls_rbio};
+   delete $self->{tls_wbio};
+   delete $self->{tls_wbuf};
+   delete $self->{filter_r};
+   delete $self->{filter_w};
+}
 sub DESTROY {
    my $self = shift;
-   Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
+   $self->stoptls;
 }
 =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.21 by root, Sat May 24 15:03:42 2008 UTC vs. Revision 1.34 by root, Mon May 26 03:27:52 2008 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.21 by root, Sat May 24 15:03:42 2008 UTC vs.
Revision 1.34 by root, Mon May 26 03:27:52 2008 UTC