ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.21 by root, Sat May 24 15:03:42 2008 UTC vs.
Revision 1.30 by root, Sat May 24 23:56:26 2008 UTC

10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw/EAGAIN EINTR/;
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17This module is experimental. 17This module is experimental.
18 18
19=cut 19=cut
20 20
90 90
91The object will not be in a usable state when this callback has been 91The object will not be in a usable state when this callback has been
92called. 92called.
93 93
94On callback entrance, the value of C<$!> contains the operating system 94On callback entrance, the value of C<$!> contains the operating system
95error (or C<ENOSPC> or C<EPIPE>). 95error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>).
96 96
97While not mandatory, it is I<highly> recommended to set this callback, as 97While not mandatory, it is I<highly> recommended to set this callback, as
98you will not be notified of errors otherwise. The default simply calls 98you will not be notified of errors otherwise. The default simply calls
99die. 99die.
100 100
145 145
146When this parameter is given, it enables TLS (SSL) mode, that means it 146When this parameter is given, it enables TLS (SSL) mode, that means it
147will start making tls handshake and will transparently encrypt/decrypt 147will start making tls handshake and will transparently encrypt/decrypt
148data. 148data.
149 149
150TLS mode requires Net::SSLeay to be installed (it will be loaded
151automatically when you try to create a TLS handle).
152
150For the TLS server side, use C<accept>, and for the TLS client side of a 153For the TLS server side, use C<accept>, and for the TLS client side of a
151connection, use C<connect> mode. 154connection, use C<connect> mode.
152 155
153You can also provide your own TLS connection object, but you have 156You can also provide your own TLS connection object, but you have
154to make sure that you call either C<Net::SSLeay::set_connect_state> 157to make sure that you call either C<Net::SSLeay::set_connect_state>
155or C<Net::SSLeay::set_accept_state> on it before you pass it to 158or C<Net::SSLeay::set_accept_state> on it before you pass it to
156AnyEvent::Handle. 159AnyEvent::Handle.
157 160
161See the C<starttls> method if you need to start TLs negotiation later.
162
158=item tls_ctx => $ssl_ctx 163=item tls_ctx => $ssl_ctx
159 164
160Use the given Net::SSLeay::CTX object to create the new TLS connection 165Use the given Net::SSLeay::CTX object to create the new TLS connection
161(unless a connection object was specified directly). If this parameter is 166(unless a connection object was specified directly). If this parameter is
162missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 167missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
206 } 211 }
207 212
208 if ($self->{on_error}) { 213 if ($self->{on_error}) {
209 $self->{on_error}($self); 214 $self->{on_error}($self);
210 } else { 215 } else {
211 die "AnyEvent::Handle uncaught fatal error: $!"; 216 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
212 } 217 }
213} 218}
214 219
215=item $fh = $handle->fh 220=item $fh = $handle->fh
216 221
217This method returns the filehandle of the L<AnyEvent::Handle> object. 222This method returns the file handle of the L<AnyEvent::Handle> object.
218 223
219=cut 224=cut
220 225
221sub fh { $_[0]->{fh} } 226sub fh { $_[0]->{fh} }
222 227
282=cut 287=cut
283 288
284sub _drain_wbuf { 289sub _drain_wbuf {
285 my ($self) = @_; 290 my ($self) = @_;
286 291
287 unless ($self->{ww}) { 292 if (!$self->{ww} && length $self->{wbuf}) {
288 Scalar::Util::weaken $self; 293 Scalar::Util::weaken $self;
289 my $cb = sub { 294 my $cb = sub {
290 my $len = syswrite $self->{fh}, $self->{wbuf}; 295 my $len = syswrite $self->{fh}, $self->{wbuf};
291 296
292 if ($len > 0) { 297 if ($len >= 0) {
293 substr $self->{wbuf}, 0, $len, ""; 298 substr $self->{wbuf}, 0, $len, "";
294 299
295 $self->{on_drain}($self) 300 $self->{on_drain}($self)
296 if $self->{low_water_mark} >= length $self->{wbuf} 301 if $self->{low_water_mark} >= length $self->{wbuf}
297 && $self->{on_drain}; 302 && $self->{on_drain};
306 311
307 $cb->($self); 312 $cb->($self);
308 }; 313 };
309} 314}
310 315
316our %WH;
317
318sub register_write_type($$) {
319 $WH{$_[0]} = $_[1];
320}
321
311sub push_write { 322sub push_write {
312 my $self = shift; 323 my $self = shift;
324
325 if (@_ > 1) {
326 my $type = shift;
327
328 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
329 ->($self, @_);
330 }
313 331
314 if ($self->{filter_w}) { 332 if ($self->{filter_w}) {
315 $self->{filter_w}->($self, \$_[0]); 333 $self->{filter_w}->($self, \$_[0]);
316 } else { 334 } else {
317 $self->{wbuf} .= $_[0]; 335 $self->{wbuf} .= $_[0];
318 $self->_drain_wbuf; 336 $self->_drain_wbuf;
319 } 337 }
320} 338}
339
340=item $handle->push_write (type => @args)
341
342=item $handle->unshift_write (type => @args)
343
344Instead of formatting your data yourself, you can also let this module do
345the job by specifying a type and type-specific arguments.
346
347Predefined types are (if you have ideas for additional types, feel free to
348drop by and tell us):
349
350=over 4
351
352=item netstring => $string
353
354Formats the given value as netstring
355(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
356
357=back
358
359=cut
360
361register_write_type netstring => sub {
362 my ($self, $string) = @_;
363
364 sprintf "%d:%s,", (length $string), $string
365};
366
367=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
368
369This function (not method) lets you add your own types to C<push_write>.
370Whenever the given C<type> is used, C<push_write> will invoke the code
371reference with the handle object and the remaining arguments.
372
373The code reference is supposed to return a single octet string that will
374be appended to the write buffer.
375
376Note that this is a function, and all types registered this way will be
377global, so try to use unique names.
378
379=cut
321 380
322############################################################################# 381#############################################################################
323 382
324=back 383=back
325 384
413 local $self->{in_drain} = 1; 472 local $self->{in_drain} = 1;
414 473
415 while (my $len = length $self->{rbuf}) { 474 while (my $len = length $self->{rbuf}) {
416 no strict 'refs'; 475 no strict 'refs';
417 if (my $cb = shift @{ $self->{queue} }) { 476 if (my $cb = shift @{ $self->{queue} }) {
418 if (!$cb->($self)) { 477 unless ($cb->($self)) {
419 if ($self->{eof}) { 478 if ($self->{eof}) {
420 # no progress can be made (not enough data and no data forthcoming) 479 # no progress can be made (not enough data and no data forthcoming)
421 $! = &Errno::EPIPE; return $self->error; 480 $! = &Errno::EPIPE; return $self->error;
422 } 481 }
423 482
500interested in (which can be none at all) and return a true value. After returning 559interested in (which can be none at all) and return a true value. After returning
501true, it will be removed from the queue. 560true, it will be removed from the queue.
502 561
503=cut 562=cut
504 563
564our %RH;
565
566sub register_read_type($$) {
567 $RH{$_[0]} = $_[1];
568}
569
505sub push_read { 570sub push_read {
506 my ($self, $cb) = @_; 571 my $self = shift;
572 my $cb = pop;
573
574 if (@_) {
575 my $type = shift;
576
577 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
578 ->($self, $cb, @_);
579 }
507 580
508 push @{ $self->{queue} }, $cb; 581 push @{ $self->{queue} }, $cb;
509 $self->_drain_rbuf; 582 $self->_drain_rbuf;
510} 583}
511 584
512sub unshift_read { 585sub unshift_read {
513 my ($self, $cb) = @_; 586 my $self = shift;
587 my $cb = pop;
514 588
589 if (@_) {
590 my $type = shift;
591
592 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
593 ->($self, $cb, @_);
594 }
595
596
515 push @{ $self->{queue} }, $cb; 597 unshift @{ $self->{queue} }, $cb;
516 $self->_drain_rbuf; 598 $self->_drain_rbuf;
517} 599}
518 600
519=item $handle->push_read_chunk ($len, $cb->($self, $data)) 601=item $handle->push_read (type => @args, $cb)
520 602
521=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) 603=item $handle->unshift_read (type => @args, $cb)
522 604
523Append the given callback to the end of the queue (C<push_read_chunk>) or 605Instead of providing a callback that parses the data itself you can chose
524prepend it (C<unshift_read_chunk>). 606between a number of predefined parsing formats, for chunks of data, lines
607etc.
525 608
526The callback will be called only once C<$len> bytes have been read, and 609Predefined types are (if you have ideas for additional types, feel free to
527these C<$len> bytes will be passed to the callback. 610drop by and tell us):
528 611
529=cut 612=over 4
530 613
531sub _read_chunk($$) { 614=item chunk => $octets, $cb->($self, $data)
615
616Invoke the callback only once C<$octets> bytes have been read. Pass the
617data read to the callback. The callback will never be called with less
618data.
619
620Example: read 2 bytes.
621
622 $handle->push_read (chunk => 2, sub {
623 warn "yay ", unpack "H*", $_[1];
624 });
625
626=cut
627
628register_read_type chunk => sub {
532 my ($self, $len, $cb) = @_; 629 my ($self, $cb, $len) = @_;
533 630
534 sub { 631 sub {
535 $len <= length $_[0]{rbuf} or return; 632 $len <= length $_[0]{rbuf} or return;
536 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 633 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
537 1 634 1
538 } 635 }
539} 636};
540 637
638# compatibility with older API
541sub push_read_chunk { 639sub push_read_chunk {
542 $_[0]->push_read (&_read_chunk); 640 $_[0]->push_read (chunk => $_[1], $_[2]);
543} 641}
544
545 642
546sub unshift_read_chunk { 643sub unshift_read_chunk {
547 $_[0]->unshift_read (&_read_chunk); 644 $_[0]->unshift_read (chunk => $_[1], $_[2]);
548} 645}
549 646
550=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol)) 647=item line => [$eol, ]$cb->($self, $line, $eol)
551
552=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
553
554Append the given callback to the end of the queue (C<push_read_line>) or
555prepend it (C<unshift_read_line>).
556 648
557The callback will be called only once a full line (including the end of 649The callback will be called only once a full line (including the end of
558line marker, C<$eol>) has been read. This line (excluding the end of line 650line marker, C<$eol>) has been read. This line (excluding the end of line
559marker) will be passed to the callback as second argument (C<$line>), and 651marker) will be passed to the callback as second argument (C<$line>), and
560the end of line marker as the third argument (C<$eol>). 652the end of line marker as the third argument (C<$eol>).
571Partial lines at the end of the stream will never be returned, as they are 663Partial lines at the end of the stream will never be returned, as they are
572not marked by the end of line marker. 664not marked by the end of line marker.
573 665
574=cut 666=cut
575 667
576sub _read_line($$) { 668register_read_type line => sub {
577 my $self = shift; 669 my ($self, $cb, $eol) = @_;
578 my $cb = pop;
579 my $eol = @_ ? shift : qr|(\015?\012)|;
580 my $pos;
581 670
671 $eol = qr|(\015?\012)| if @_ < 3;
582 $eol = quotemeta $eol unless ref $eol; 672 $eol = quotemeta $eol unless ref $eol;
583 $eol = qr|^(.*?)($eol)|s; 673 $eol = qr|^(.*?)($eol)|s;
584 674
585 sub { 675 sub {
586 $_[0]{rbuf} =~ s/$eol// or return; 676 $_[0]{rbuf} =~ s/$eol// or return;
587 677
588 $cb->($_[0], $1, $2); 678 $cb->($_[0], $1, $2);
589 1 679 1
590 } 680 }
591} 681};
592 682
683# compatibility with older API
593sub push_read_line { 684sub push_read_line {
594 $_[0]->push_read (&_read_line); 685 my $self = shift;
686 $self->push_read (line => @_);
595} 687}
596 688
597sub unshift_read_line { 689sub unshift_read_line {
598 $_[0]->unshift_read (&_read_line); 690 my $self = shift;
691 $self->unshift_read (line => @_);
599} 692}
693
694=item netstring => $cb->($string)
695
696A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
697
698Throws an error with C<$!> set to EBADMSG on format violations.
699
700=cut
701
702register_read_type netstring => sub {
703 my ($self, $cb) = @_;
704
705 sub {
706 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
707 if ($_[0]{rbuf} =~ /[^0-9]/) {
708 $! = &Errno::EBADMSG;
709 $self->error;
710 }
711 return;
712 }
713
714 my $len = $1;
715
716 $self->unshift_read (chunk => $len, sub {
717 my $string = $_[1];
718 $_[0]->unshift_read (chunk => 1, sub {
719 if ($_[1] eq ",") {
720 $cb->($_[0], $string);
721 } else {
722 $! = &Errno::EBADMSG;
723 $self->error;
724 }
725 });
726 });
727
728 1
729 }
730};
731
732=back
733
734=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
735
736This function (not method) lets you add your own types to C<push_read>.
737
738Whenever the given C<type> is used, C<push_read> will invoke the code
739reference with the handle object, the callback and the remaining
740arguments.
741
742The code reference is supposed to return a callback (usually a closure)
743that works as a plain read callback (see C<< ->push_read ($cb) >>).
744
745It should invoke the passed callback when it is done reading (remember to
746pass C<$self> as first argument as all other callbacks do that).
747
748Note that this is a function, and all types registered this way will be
749global, so try to use unique names.
750
751For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
752search for C<register_read_type>)).
600 753
601=item $handle->stop_read 754=item $handle->stop_read
602 755
603=item $handle->start_read 756=item $handle->start_read
604 757
605In rare cases you actually do not want to read anything from the 758In rare cases you actually do not want to read anything from the
606socket. In this case you can call C<stop_read>. Neither C<on_read> no 759socket. In this case you can call C<stop_read>. Neither C<on_read> no
607any queued callbacks will be executed then. To start readign again, call 760any queued callbacks will be executed then. To start reading again, call
608C<start_read>. 761C<start_read>.
609 762
610=cut 763=cut
611 764
612sub stop_read { 765sub stop_read {
644 797
645sub _dotls { 798sub _dotls {
646 my ($self) = @_; 799 my ($self) = @_;
647 800
648 if (length $self->{tls_wbuf}) { 801 if (length $self->{tls_wbuf}) {
649 my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf}); 802 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) {
650 substr $self->{tls_wbuf}, 0, $len, "" if $len > 0; 803 substr $self->{tls_wbuf}, 0, $len, "";
804 }
651 } 805 }
652 806
653 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) { 807 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) {
654 $self->{wbuf} .= $buf; 808 $self->{wbuf} .= $buf;
655 $self->_drain_wbuf; 809 $self->_drain_wbuf;
656 } 810 }
657 811
658 if (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 812 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
659 $self->{rbuf} .= $buf; 813 $self->{rbuf} .= $buf;
660 $self->_drain_rbuf; 814 $self->_drain_rbuf;
661 } elsif ( 815 }
816
662 (my $err = Net::SSLeay::get_error ($self->{tls}, -1)) 817 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
818
663 != Net::SSLeay::ERROR_WANT_READ () 819 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
664 ) {
665 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 820 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
666 $self->error; 821 $self->error;
667 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 822 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
668 $! = &Errno::EIO; 823 $! = &Errno::EIO;
669 $self->error; 824 $self->error;
671 826
672 # all others are fine for our purposes 827 # all others are fine for our purposes
673 } 828 }
674} 829}
675 830
831=item $handle->starttls ($tls[, $tls_ctx])
832
833Instead of starting TLS negotiation immediately when the AnyEvent::Handle
834object is created, you can also do that at a later time by calling
835C<starttls>.
836
837The first argument is the same as the C<tls> constructor argument (either
838C<"connect">, C<"accept"> or an existing Net::SSLeay object).
839
840The second argument is the optional C<Net::SSLeay::CTX> object that is
841used when AnyEvent::Handle has to create its own TLS connection object.
842
843=cut
844
676# TODO: maybe document... 845# TODO: maybe document...
677sub starttls { 846sub starttls {
678 my ($self, $ssl, $ctx) = @_; 847 my ($self, $ssl, $ctx) = @_;
848
849 $self->stoptls;
679 850
680 if ($ssl eq "accept") { 851 if ($ssl eq "accept") {
681 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 852 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
682 Net::SSLeay::set_accept_state ($ssl); 853 Net::SSLeay::set_accept_state ($ssl);
683 } elsif ($ssl eq "connect") { 854 } elsif ($ssl eq "connect") {
689 860
690 # basically, this is deep magic (because SSL_read should have the same issues) 861 # basically, this is deep magic (because SSL_read should have the same issues)
691 # but the openssl maintainers basically said: "trust us, it just works". 862 # but the openssl maintainers basically said: "trust us, it just works".
692 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 863 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
693 # and mismaintained ssleay-module doesn't even offer them). 864 # and mismaintained ssleay-module doesn't even offer them).
865 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
694 Net::SSLeay::CTX_set_mode ($self->{tls}, 866 Net::SSLeay::CTX_set_mode ($self->{tls},
695 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 867 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
696 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 868 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
697 869
698 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 870 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
708 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]}); 880 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]});
709 &_dotls; 881 &_dotls;
710 }; 882 };
711} 883}
712 884
885=item $handle->stoptls
886
887Destroys the SSL connection, if any. Partial read or write data will be
888lost.
889
890=cut
891
892sub stoptls {
893 my ($self) = @_;
894
895 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
896 delete $self->{tls_rbio};
897 delete $self->{tls_wbio};
898 delete $self->{tls_wbuf};
899 delete $self->{filter_r};
900 delete $self->{filter_w};
901}
902
713sub DESTROY { 903sub DESTROY {
714 my $self = shift; 904 my $self = shift;
715 905
716 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 906 $self->stoptls;
717} 907}
718 908
719=item AnyEvent::Handle::TLS_CTX 909=item AnyEvent::Handle::TLS_CTX
720 910
721This function creates and returns the Net::SSLeay::CTX object used by 911This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines