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.16 by root, Fri May 23 05:16:57 2008 UTC vs.
Revision 1.28 by root, Sat May 24 22:27:11 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
… …
83waiting for data. 83waiting for data.
84 84
85=item on_error => $cb->($self) 85=item on_error => $cb->($self)
86 86
87This is the fatal error callback, that is called when, well, a fatal error 87This is the fatal error callback, that is called when, well, a fatal error
88ocurs, such as not being able to resolve the hostname, failure to connect 88occurs, such as not being able to resolve the hostname, failure to connect
89or a read error. 89or a read error.
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
… …
102 102
103This sets the default read callback, which is called when data arrives 103This sets the default read callback, which is called when data arrives
104and no read request is in the queue. 104and no read request is in the queue.
105 105
106To access (and remove data from) the read buffer, use the C<< ->rbuf >> 106To access (and remove data from) the read buffer, use the C<< ->rbuf >>
107method or acces sthe C<$self->{rbuf}> member directly. 107method or access the C<$self->{rbuf}> member directly.
108 108
109When an EOF condition is detected then AnyEvent::Handle will first try to 109When an EOF condition is detected then AnyEvent::Handle will first try to
110feed all the remaining data to the queued callbacks and C<on_read> before 110feed all the remaining data to the queued callbacks and C<on_read> before
111calling the C<on_eof> callback. If no progress can be made, then a fatal 111calling the C<on_eof> callback. If no progress can be made, then a fatal
112error will be raised (with C<$!> set to C<EPIPE>). 112error will be raised (with C<$!> set to C<EPIPE>).
… …
139 139
140Sets the amount of bytes (default: C<0>) that make up an "empty" write 140Sets the amount of bytes (default: C<0>) that make up an "empty" write
141buffer: If the write reaches this size or gets even samller it is 141buffer: If the write reaches this size or gets even samller it is
142considered empty. 142considered empty.
143 143
144=item tls => "accept" | "connect" | Net::SSLeay::SSL object
145
146When this parameter is given, it enables TLS (SSL) mode, that means it
147will start making tls handshake and will transparently encrypt/decrypt
148data.
149
150TLS mode requires Net::SSLeay to be installed (it will be loaded
151automatically when you try to create a TLS handle).
152
153For the TLS server side, use C<accept>, and for the TLS client side of a
154connection, use C<connect> mode.
155
156You can also provide your own TLS connection object, but you have
157to make sure that you call either C<Net::SSLeay::set_connect_state>
158or C<Net::SSLeay::set_accept_state> on it before you pass it to
159AnyEvent::Handle.
160
161See the C<starttls> method if you need to start TLs negotiation later.
162
163=item tls_ctx => $ssl_ctx
164
165Use the given Net::SSLeay::CTX object to create the new TLS connection
166(unless a connection object was specified directly). If this parameter is
167missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
168
144=back 169=back
145 170
146=cut 171=cut
172
173our (%RH, %WH);
174
175sub register_read_type($$) {
176 $RH{$_[0]} = $_[1];
177}
178
179sub register_write_type($$) {
180 $WH{$_[0]} = $_[1];
181}
147 182
148sub new { 183sub new {
149 my $class = shift; 184 my $class = shift;
150 185
151 my $self = bless { @_ }, $class; 186 my $self = bless { @_ }, $class;
152 187
153 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 188 $self->{fh} or Carp::croak "mandatory argument fh is missing";
154 189
155 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 190 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
191
192 if ($self->{tls}) {
193 require Net::SSLeay;
194 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
195 }
156 196
157 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 197 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
158 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 198 $self->on_error (delete $self->{on_error}) if $self->{on_error};
159 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 199 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
160 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 200 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
… …
187 } 227 }
188} 228}
189 229
190=item $fh = $handle->fh 230=item $fh = $handle->fh
191 231
192This method returns the filehandle of the L<AnyEvent::Handle> object. 232This method returns the file handle of the L<AnyEvent::Handle> object.
193 233
194=cut 234=cut
195 235
196sub fh { $_[0]->{fh} } 236sub fh { $_[0]->{fh} }
197 237
… …
225for reading. 265for reading.
226 266
227The write queue is very simple: you can add data to its end, and 267The write queue is very simple: you can add data to its end, and
228AnyEvent::Handle will automatically try to get rid of it for you. 268AnyEvent::Handle will automatically try to get rid of it for you.
229 269
230When data could be writtena nd the write buffer is shorter then the low 270When data could be written and the write buffer is shorter then the low
231water mark, the C<on_drain> callback will be invoked. 271water mark, the C<on_drain> callback will be invoked.
232 272
233=over 4 273=over 4
234 274
235=item $handle->on_drain ($cb) 275=item $handle->on_drain ($cb)
… …
254want (only limited by the available memory), as C<AnyEvent::Handle> 294want (only limited by the available memory), as C<AnyEvent::Handle>
255buffers it independently of the kernel. 295buffers it independently of the kernel.
256 296
257=cut 297=cut
258 298
259sub push_write { 299sub _drain_wbuf {
260 my ($self, $data) = @_; 300 my ($self) = @_;
261
262 $self->{wbuf} .= $data;
263 301
264 unless ($self->{ww}) { 302 unless ($self->{ww}) {
265 Scalar::Util::weaken $self; 303 Scalar::Util::weaken $self;
266 my $cb = sub { 304 my $cb = sub {
267 my $len = syswrite $self->{fh}, $self->{wbuf}; 305 my $len = syswrite $self->{fh}, $self->{wbuf};
268 306
269 if ($len > 0) { 307 if ($len > 0) {
270 substr $self->{wbuf}, 0, $len, ""; 308 substr $self->{wbuf}, 0, $len, "";
271
272 309
273 $self->{on_drain}($self) 310 $self->{on_drain}($self)
274 if $self->{low_water_mark} >= length $self->{wbuf} 311 if $self->{low_water_mark} >= length $self->{wbuf}
275 && $self->{on_drain}; 312 && $self->{on_drain};
276 313
… …
282 319
283 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 320 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
284 321
285 $cb->($self); 322 $cb->($self);
286 }; 323 };
324}
325
326sub push_write {
327 my $self = shift;
328
329 if ($self->{filter_w}) {
330 $self->{filter_w}->($self, \$_[0]);
331 } else {
332 $self->{wbuf} .= $_[0];
333 $self->_drain_wbuf;
334 }
287} 335}
288 336
289############################################################################# 337#############################################################################
290 338
291=back 339=back
… …
366 414
367=cut 415=cut
368 416
369sub _drain_rbuf { 417sub _drain_rbuf {
370 my ($self) = @_; 418 my ($self) = @_;
419
420 if (
421 defined $self->{rbuf_max}
422 && $self->{rbuf_max} < length $self->{rbuf}
423 ) {
424 $! = &Errno::ENOSPC; return $self->error;
425 }
371 426
372 return if $self->{in_drain}; 427 return if $self->{in_drain};
373 local $self->{in_drain} = 1; 428 local $self->{in_drain} = 1;
374 429
375 while (my $len = length $self->{rbuf}) { 430 while (my $len = length $self->{rbuf}) {
… …
448Append the given callback to the end of the queue (C<push_read>) or 503Append the given callback to the end of the queue (C<push_read>) or
449prepend it (C<unshift_read>). 504prepend it (C<unshift_read>).
450 505
451The callback is called each time some additional read data arrives. 506The callback is called each time some additional read data arrives.
452 507
453It must check wether enough data is in the read buffer already. 508It must check whether enough data is in the read buffer already.
454 509
455If not enough data is available, it must return the empty list or a false 510If not enough data is available, it must return the empty list or a false
456value, in which case it will be called repeatedly until enough data is 511value, in which case it will be called repeatedly until enough data is
457available (or an error condition is detected). 512available (or an error condition is detected).
458 513
… …
461true, it will be removed from the queue. 516true, it will be removed from the queue.
462 517
463=cut 518=cut
464 519
465sub push_read { 520sub push_read {
466 my ($self, $cb) = @_; 521 my $self = shift;
522 my $cb = pop;
523
524 if (@_) {
525 my $type = shift;
526
527 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
528 ->($self, $cb, @_);
529 }
467 530
468 push @{ $self->{queue} }, $cb; 531 push @{ $self->{queue} }, $cb;
469 $self->_drain_rbuf; 532 $self->_drain_rbuf;
470} 533}
471 534
472sub unshift_read { 535sub unshift_read {
473 my ($self, $cb) = @_; 536 my $self = shift;
537 my $cb = pop;
474 538
539 if (@_) {
540 my $type = shift;
541
542 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
543 ->($self, $cb, @_);
544 }
545
546
475 push @{ $self->{queue} }, $cb; 547 unshift @{ $self->{queue} }, $cb;
476 $self->_drain_rbuf; 548 $self->_drain_rbuf;
477} 549}
478 550
479=item $handle->push_read_chunk ($len, $cb->($self, $data)) 551=item $handle->push_read (type => @args, $cb)
480 552
481=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) 553=item $handle->unshift_read (type => @args, $cb)
482 554
483Append the given callback to the end of the queue (C<push_read_chunk>) or 555Instead of providing a callback that parses the data itself you can chose
484prepend it (C<unshift_read_chunk>). 556between a number of predefined parsing formats, for chunks of data, lines
557etc.
485 558
486The callback will be called only once C<$len> bytes have been read, and 559The types currently supported are:
487these C<$len> bytes will be passed to the callback.
488 560
489=cut 561=over 4
490 562
491sub _read_chunk($$) { 563=item chunk => $octets, $cb->($self, $data)
564
565Invoke the callback only once C<$octets> bytes have been read. Pass the
566data read to the callback. The callback will never be called with less
567data.
568
569Example: read 2 bytes.
570
571 $handle->push_read (chunk => 2, sub {
572 warn "yay ", unpack "H*", $_[1];
573 });
574
575=cut
576
577register_read_type chunk => sub {
492 my ($self, $len, $cb) = @_; 578 my ($self, $cb, $len) = @_;
493 579
494 sub { 580 sub {
495 $len <= length $_[0]{rbuf} or return; 581 $len <= length $_[0]{rbuf} or return;
496 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 582 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
497 1 583 1
498 } 584 }
499} 585};
500 586
587# compatibility with older API
501sub push_read_chunk { 588sub push_read_chunk {
502 $_[0]->push_read (&_read_chunk); 589 $_[0]->push_read (chunk => $_[1], $_[2]);
503} 590}
504
505 591
506sub unshift_read_chunk { 592sub unshift_read_chunk {
507 $_[0]->unshift_read (&_read_chunk); 593 $_[0]->unshift_read (chunk => $_[1], $_[2]);
508} 594}
509 595
510=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol)) 596=item line => [$eol, ]$cb->($self, $line, $eol)
511
512=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
513
514Append the given callback to the end of the queue (C<push_read_line>) or
515prepend it (C<unshift_read_line>).
516 597
517The callback will be called only once a full line (including the end of 598The callback will be called only once a full line (including the end of
518line marker, C<$eol>) has been read. This line (excluding the end of line 599line marker, C<$eol>) has been read. This line (excluding the end of line
519marker) will be passed to the callback as second argument (C<$line>), and 600marker) will be passed to the callback as second argument (C<$line>), and
520the end of line marker as the third argument (C<$eol>). 601the end of line marker as the third argument (C<$eol>).
… …
531Partial lines at the end of the stream will never be returned, as they are 612Partial lines at the end of the stream will never be returned, as they are
532not marked by the end of line marker. 613not marked by the end of line marker.
533 614
534=cut 615=cut
535 616
536sub _read_line($$) { 617register_read_type line => sub {
537 my $self = shift; 618 my ($self, $cb, $eol) = @_;
538 my $cb = pop;
539 my $eol = @_ ? shift : qr|(\015?\012)|;
540 my $pos;
541 619
620 $eol = qr|(\015?\012)| if @_ < 3;
542 $eol = quotemeta $eol unless ref $eol; 621 $eol = quotemeta $eol unless ref $eol;
543 $eol = qr|^(.*?)($eol)|s; 622 $eol = qr|^(.*?)($eol)|s;
544 623
545 sub { 624 sub {
546 $_[0]{rbuf} =~ s/$eol// or return; 625 $_[0]{rbuf} =~ s/$eol// or return;
547 626
548 $cb->($_[0], $1, $2); 627 $cb->($_[0], $1, $2);
549 1 628 1
550 } 629 }
551} 630};
552 631
632# compatibility with older API
553sub push_read_line { 633sub push_read_line {
554 $_[0]->push_read (&_read_line); 634 my $self = shift;
635 $self->push_read (line => @_);
555} 636}
556 637
557sub unshift_read_line { 638sub unshift_read_line {
558 $_[0]->unshift_read (&_read_line); 639 my $self = shift;
640 $self->unshift_read (line => @_);
559} 641}
642
643=back
560 644
561=item $handle->stop_read 645=item $handle->stop_read
562 646
563=item $handle->start_read 647=item $handle->start_read
564 648
565In rare cases you actually do not want to read anything form the 649In rare cases you actually do not want to read anything from the
566socket. In this case you can call C<stop_read>. Neither C<on_read> no 650socket. In this case you can call C<stop_read>. Neither C<on_read> no
567any queued callbacks will be executed then. To start readign again, call 651any queued callbacks will be executed then. To start reading again, call
568C<start_read>. 652C<start_read>.
569 653
570=cut 654=cut
571 655
572sub stop_read { 656sub stop_read {
… …
580 664
581 unless ($self->{rw} || $self->{eof}) { 665 unless ($self->{rw} || $self->{eof}) {
582 Scalar::Util::weaken $self; 666 Scalar::Util::weaken $self;
583 667
584 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 668 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
669 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
585 my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} || 8192, length $self->{rbuf}; 670 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
586 671
587 if ($len > 0) { 672 if ($len > 0) {
588 if (defined $self->{rbuf_max}) { 673 $self->{filter_r}
589 if ($self->{rbuf_max} < length $self->{rbuf}) { 674 ? $self->{filter_r}->($self, $rbuf)
590 $! = &Errno::ENOSPC; return $self->error; 675 : $self->_drain_rbuf;
591 }
592 }
593 676
594 } elsif (defined $len) { 677 } elsif (defined $len) {
678 delete $self->{rw};
595 $self->{eof} = 1; 679 $self->{eof} = 1;
596 delete $self->{rw}; 680 $self->_drain_rbuf;
597 681
598 } elsif ($! != EAGAIN && $! != EINTR) { 682 } elsif ($! != EAGAIN && $! != EINTR) {
599 return $self->error; 683 return $self->error;
600 } 684 }
601
602 $self->_drain_rbuf;
603 }); 685 });
604 } 686 }
605} 687}
606 688
689sub _dotls {
690 my ($self) = @_;
691
692 if (length $self->{tls_wbuf}) {
693 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) {
694 substr $self->{tls_wbuf}, 0, $len, "";
695 }
696 }
697
698 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) {
699 $self->{wbuf} .= $buf;
700 $self->_drain_wbuf;
701 }
702
703 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
704 $self->{rbuf} .= $buf;
705 $self->_drain_rbuf;
706 }
707
708 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
709
710 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
711 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
712 $self->error;
713 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
714 $! = &Errno::EIO;
715 $self->error;
716 }
717
718 # all others are fine for our purposes
719 }
720}
721
722=item $handle->starttls ($tls[, $tls_ctx])
723
724Instead of starting TLS negotiation immediately when the AnyEvent::Handle
725object is created, you can also do that at a later time by calling
726C<starttls>.
727
728The first argument is the same as the C<tls> constructor argument (either
729C<"connect">, C<"accept"> or an existing Net::SSLeay object).
730
731The second argument is the optional C<Net::SSLeay::CTX> object that is
732used when AnyEvent::Handle has to create its own TLS connection object.
733
734=cut
735
736# TODO: maybe document...
737sub starttls {
738 my ($self, $ssl, $ctx) = @_;
739
740 $self->stoptls;
741
742 if ($ssl eq "accept") {
743 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
744 Net::SSLeay::set_accept_state ($ssl);
745 } elsif ($ssl eq "connect") {
746 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
747 Net::SSLeay::set_connect_state ($ssl);
748 }
749
750 $self->{tls} = $ssl;
751
752 # basically, this is deep magic (because SSL_read should have the same issues)
753 # but the openssl maintainers basically said: "trust us, it just works".
754 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
755 # and mismaintained ssleay-module doesn't even offer them).
756 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
757 Net::SSLeay::CTX_set_mode ($self->{tls},
758 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
759 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
760
761 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
762 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
763
764 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});
765
766 $self->{filter_w} = sub {
767 $_[0]{tls_wbuf} .= ${$_[1]};
768 &_dotls;
769 };
770 $self->{filter_r} = sub {
771 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]});
772 &_dotls;
773 };
774}
775
776=item $handle->stoptls
777
778Destroys the SSL connection, if any. Partial read or write data will be
779lost.
780
781=cut
782
783sub stoptls {
784 my ($self) = @_;
785
786 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
787 delete $self->{tls_rbio};
788 delete $self->{tls_wbio};
789 delete $self->{tls_wbuf};
790 delete $self->{filter_r};
791 delete $self->{filter_w};
792}
793
794sub DESTROY {
795 my $self = shift;
796
797 $self->stoptls;
798}
799
800=item AnyEvent::Handle::TLS_CTX
801
802This function creates and returns the Net::SSLeay::CTX object used by
803default for TLS mode.
804
805The context is created like this:
806
807 Net::SSLeay::load_error_strings;
808 Net::SSLeay::SSLeay_add_ssl_algorithms;
809 Net::SSLeay::randomize;
810
811 my $CTX = Net::SSLeay::CTX_new;
812
813 Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
814
815=cut
816
817our $TLS_CTX;
818
819sub TLS_CTX() {
820 $TLS_CTX || do {
821 require Net::SSLeay;
822
823 Net::SSLeay::load_error_strings ();
824 Net::SSLeay::SSLeay_add_ssl_algorithms ();
825 Net::SSLeay::randomize ();
826
827 $TLS_CTX = Net::SSLeay::CTX_new ();
828
829 Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
830
831 $TLS_CTX
832 }
833}
834
607=back 835=back
608 836
609=head1 AUTHOR 837=head1 AUTHOR
610 838
611Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 839Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.

Diff Legend

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