--- AnyEvent/lib/AnyEvent/Handle.pm 2008/05/15 09:03:43 1.12 +++ AnyEvent/lib/AnyEvent/Handle.pm 2008/05/26 03:27:52 1.34 @@ -4,7 +4,7 @@ use strict; use AnyEvent (); -use AnyEvent::Util (); +use AnyEvent::Util qw(WSAWOULDBLOCK); use Scalar::Util (); use Carp (); use Fcntl (); @@ -12,11 +12,11 @@ =head1 NAME -AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent +AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent =cut -our $VERSION = '0.02'; +our $VERSION = '0.04'; =head1 SYNOPSIS @@ -25,28 +25,31 @@ my $cv = AnyEvent->condvar; - my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN); - - #TODO - - # or use the constructor to pass the callback: - - my $ae_fh2 = + my $handle = 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 (and sockets, see L for an easy way to make -non-blocking resolves and connects). +filehandles. For utility functions for doing non-blocking connects and accepts +on sockets see L. In the following, when the documentation refers to of "bytes" then this means characters. As sysread and syswrite are used for all I/O, their @@ -72,21 +75,25 @@ NOTE: The filehandle will be set to non-blocking (using AnyEvent::Util::fh_nonblocking). -=item on_eof => $cb->($self) [MANDATORY] +=item on_eof => $cb->($self) Set the callback to be called on EOF. +While not mandatory, it is highly recommended to set an eof callback, +otherwise you might end up with a closed socket while you are still +waiting for data. + =item on_error => $cb->($self) This is the fatal error callback, that is called when, well, a fatal error -ocurs, such as not being able to resolve the hostname, failure to connect +occurs, such as not being able to resolve the hostname, failure to connect or a read error. 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 or C). +error (or C, C or C). While not mandatory, it is I recommended to set this callback, as you will not be notified of errors otherwise. The default simply calls @@ -98,7 +105,7 @@ and no read request is in the queue. To access (and remove data from) the read buffer, use the C<< ->rbuf >> -method or acces sthe C<$self->{rbuf}> member directly. +method or access the C<$self->{rbuf}> member directly. When an EOF condition is detected then AnyEvent::Handle will first try to feed all the remaining data to the queued callbacks and C before @@ -135,6 +142,31 @@ buffer: If the write reaches this size or gets even samller it is considered empty. +=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. + +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, and for the TLS client side of a +connection, use C mode. + +You can also provide your own TLS connection object, but you have +to make sure that you call either C +or C on it before you pass it to +AnyEvent::Handle. + +See the C 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. + =back =cut @@ -148,8 +180,12 @@ AnyEvent::Util::fh_nonblocking $self->{fh}, 1; - $self->on_eof ((delete $self->{on_eof} ) or Carp::croak "mandatory argument on_eof is missing"); + 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}; $self->on_error (delete $self->{on_error}) if $self->{on_error}; $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; $self->on_read (delete $self->{on_read} ) if $self->{on_read}; @@ -178,13 +214,13 @@ 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 object. +This method returns the file handle of the L object. =cut @@ -222,7 +258,7 @@ The write queue is very simple: you can add data to its end, and AnyEvent::Handle will automatically try to get rid of it for you. -When data could be writtena nd the write buffer is shorter then the low +When data could be written and the write buffer is shorter then the low water mark, the C callback will be invoked. =over 4 @@ -251,26 +287,23 @@ =cut -sub push_write { - my ($self, $data) = @_; - - $self->{wbuf} .= $data; +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; } }; @@ -281,6 +314,71 @@ }; } +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. +Whenever the given C is used, C 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 @@ -364,13 +462,20 @@ sub _drain_rbuf { my ($self) = @_; + if ( + defined $self->{rbuf_max} + && $self->{rbuf_max} < length $self->{rbuf} + ) { + $! = &Errno::ENOSPC; return $self->error; + } + return if $self->{in_drain}; 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; @@ -400,7 +505,8 @@ if ($self->{eof}) { $self->_shutdown; - $self->{on_eof}($self); + $self->{on_eof}($self) + if $self->{on_eof}; } } @@ -444,7 +550,7 @@ The callback is called each time some additional read data arrives. -It must check wether enough data is in the read buffer already. +It must check whether enough data is in the read buffer already. If not enough data is available, it must return the empty list or a false value, in which case it will be called repeatedly until enough data is @@ -456,57 +562,90 @@ =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; - push @{ $self->{queue} }, $cb; + if (@_) { + my $type = shift; + + $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_chunk ($len, $cb->($self, $data)) +=item $handle->push_read (type => @args, $cb) + +=item $handle->unshift_read (type => @args, $cb) -=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) +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. -Append the given callback to the end of the queue (C) or -prepend it (C). +Predefined types are (if you have ideas for additional types, feel free to +drop by and tell us): -The callback will be called only once C<$len> bytes have been read, and -these C<$len> bytes will be passed to the callback. +=over 4 + +=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 -sub _read_chunk($$) { - my ($self, $len, $cb) = @_; +register_read_type chunk => sub { + 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 $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol)) - -Append the given callback to the end of the queue (C) or -prepend it (C). +=item line => [$eol, ]$cb->($self, $line, $eol) 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 @@ -527,14 +666,12 @@ =cut -sub _read_line($$) { - my $self = shift; - my $cb = pop; - my $eol = @_ ? shift : qr|(\015?\012)|; - my $pos; +register_read_type line => sub { + my ($self, $cb, $eol) = @_; - $eol = qr|(\Q$eol\E)| unless ref $eol; - $eol = qr|^(.*?)($eol)|; + $eol = qr|(\015?\012)| if @_ < 3; + $eol = quotemeta $eol unless ref $eol; + $eol = qr|^(.*?)($eol)|s; sub { $_[0]{rbuf} =~ s/$eol// or return; @@ -542,23 +679,86 @@ $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. + +Whenever the given C is used, C 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, +search for C)). + =item $handle->stop_read =item $handle->start_read -In rare cases you actually do not want to read anything form the +In rare cases you actually do not want to read anything from the socket. In this case you can call C. Neither C 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. =cut @@ -576,28 +776,172 @@ Scalar::Util::weaken $self; $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { - my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} || 8192, length $self->{rbuf}; + my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; + my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; if ($len > 0) { - if (defined $self->{rbuf_max}) { - if ($self->{rbuf_max} < length $self->{rbuf}) { - $! = &Errno::ENOSPC; return $self->error; - } - } + $self->{filter_r} + ? $self->{filter_r}->($self, $rbuf) + : $self->_drain_rbuf; } elsif (defined $len) { - $self->{eof} = 1; delete $self->{rw}; + $self->{eof} = 1; + $self->_drain_rbuf; - } elsif ($! != EAGAIN && $! != EINTR) { + } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { return $self->error; } - - $self->_drain_rbuf; }); } } +sub _dotls { + my ($self) = @_; + + 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->{tls_wbio}))) { + $self->{wbuf} .= $buf; + $self->_drain_wbuf; + } + + while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { + $self->{rbuf} .= $buf; + $self->_drain_rbuf; + } + + my $err = Net::SSLeay::get_error ($self->{tls}, -1); + + 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. + +The first argument is the same as the C constructor argument (either +C<"connect">, C<"accept"> or an existing Net::SSLeay object). + +The second argument is the optional C 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") { + $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); + Net::SSLeay::set_connect_state ($ssl); + } + + $self->{tls} = $ssl; + + # 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 { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) + | (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}); + + $self->{filter_w} = sub { + $_[0]{tls_wbuf} .= ${$_[1]}; + &_dotls; + }; + $self->{filter_r} = sub { + 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; + + $self->stoptls; +} + +=item AnyEvent::Handle::TLS_CTX + +This function creates and returns the Net::SSLeay::CTX object used by +default for TLS mode. + +The context is created like this: + + Net::SSLeay::load_error_strings; + Net::SSLeay::SSLeay_add_ssl_algorithms; + Net::SSLeay::randomize; + + my $CTX = Net::SSLeay::CTX_new; + + Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL + +=cut + +our $TLS_CTX; + +sub TLS_CTX() { + $TLS_CTX || do { + require Net::SSLeay; + + Net::SSLeay::load_error_strings (); + Net::SSLeay::SSLeay_add_ssl_algorithms (); + Net::SSLeay::randomize (); + + $TLS_CTX = Net::SSLeay::CTX_new (); + + Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ()); + + $TLS_CTX + } +} + =back =head1 AUTHOR