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.15 by root, Sat May 17 21:34:15 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
… …
72The filehandle this L<AnyEvent::Handle> object will operate on. 72The filehandle this L<AnyEvent::Handle> object will operate on.
73 73
74NOTE: The filehandle will be set to non-blocking (using 74NOTE: The filehandle will be set to non-blocking (using
75AnyEvent::Util::fh_nonblocking). 75AnyEvent::Util::fh_nonblocking).
76 76
77=item on_eof => $cb->($self) [MANDATORY] 77=item on_eof => $cb->($self)
78 78
79Set the callback to be called on EOF. 79Set the callback to be called on EOF.
80 80
81While not mandatory, it is highly recommended to set an eof callback,
82otherwise you might end up with a closed socket while you are still
83waiting for data.
84
81=item on_error => $cb->($self) 85=item on_error => $cb->($self)
82 86
83This 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
84ocurs, 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
85or a read error. 89or a read error.
86 90
87The 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
88called. 92called.
89 93
… …
98 102
99This sets the default read callback, which is called when data arrives 103This sets the default read callback, which is called when data arrives
100and no read request is in the queue. 104and no read request is in the queue.
101 105
102To 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 >>
103method or acces sthe C<$self->{rbuf}> member directly. 107method or access the C<$self->{rbuf}> member directly.
104 108
105When 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
106feed 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
107calling 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
108error will be raised (with C<$!> set to C<EPIPE>). 112error will be raised (with C<$!> set to C<EPIPE>).
… …
135 139
136Sets 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
137buffer: 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
138considered empty. 142considered empty.
139 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
140=back 169=back
141 170
142=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}
143 182
144sub new { 183sub new {
145 my $class = shift; 184 my $class = shift;
146 185
147 my $self = bless { @_ }, $class; 186 my $self = bless { @_ }, $class;
148 187
149 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 188 $self->{fh} or Carp::croak "mandatory argument fh is missing";
150 189
151 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 190 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
152 191
153 $self->on_eof ((delete $self->{on_eof} ) or Carp::croak "mandatory argument on_eof is missing"); 192 if ($self->{tls}) {
193 require Net::SSLeay;
194 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
195 }
154 196
197 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
155 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 198 $self->on_error (delete $self->{on_error}) if $self->{on_error};
156 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 199 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
157 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 200 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
158 201
159 $self->start_read; 202 $self->start_read;
… …
184 } 227 }
185} 228}
186 229
187=item $fh = $handle->fh 230=item $fh = $handle->fh
188 231
189This method returns the filehandle of the L<AnyEvent::Handle> object. 232This method returns the file handle of the L<AnyEvent::Handle> object.
190 233
191=cut 234=cut
192 235
193sub fh { $_[0]->{fh} } 236sub fh { $_[0]->{fh} }
194 237
… …
222for reading. 265for reading.
223 266
224The 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
225AnyEvent::Handle will automatically try to get rid of it for you. 268AnyEvent::Handle will automatically try to get rid of it for you.
226 269
227When 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
228water mark, the C<on_drain> callback will be invoked. 271water mark, the C<on_drain> callback will be invoked.
229 272
230=over 4 273=over 4
231 274
232=item $handle->on_drain ($cb) 275=item $handle->on_drain ($cb)
… …
251want (only limited by the available memory), as C<AnyEvent::Handle> 294want (only limited by the available memory), as C<AnyEvent::Handle>
252buffers it independently of the kernel. 295buffers it independently of the kernel.
253 296
254=cut 297=cut
255 298
256sub push_write { 299sub _drain_wbuf {
257 my ($self, $data) = @_; 300 my ($self) = @_;
258
259 $self->{wbuf} .= $data;
260 301
261 unless ($self->{ww}) { 302 unless ($self->{ww}) {
262 Scalar::Util::weaken $self; 303 Scalar::Util::weaken $self;
263 my $cb = sub { 304 my $cb = sub {
264 my $len = syswrite $self->{fh}, $self->{wbuf}; 305 my $len = syswrite $self->{fh}, $self->{wbuf};
265 306
266 if ($len > 0) { 307 if ($len > 0) {
267 substr $self->{wbuf}, 0, $len, ""; 308 substr $self->{wbuf}, 0, $len, "";
268
269 309
270 $self->{on_drain}($self) 310 $self->{on_drain}($self)
271 if $self->{low_water_mark} >= length $self->{wbuf} 311 if $self->{low_water_mark} >= length $self->{wbuf}
272 && $self->{on_drain}; 312 && $self->{on_drain};
273 313
… …
279 319
280 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 320 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
281 321
282 $cb->($self); 322 $cb->($self);
283 }; 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 }
284} 335}
285 336
286############################################################################# 337#############################################################################
287 338
288=back 339=back
… …
363 414
364=cut 415=cut
365 416
366sub _drain_rbuf { 417sub _drain_rbuf {
367 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 }
368 426
369 return if $self->{in_drain}; 427 return if $self->{in_drain};
370 local $self->{in_drain} = 1; 428 local $self->{in_drain} = 1;
371 429
372 while (my $len = length $self->{rbuf}) { 430 while (my $len = length $self->{rbuf}) {
… …
400 } 458 }
401 } 459 }
402 460
403 if ($self->{eof}) { 461 if ($self->{eof}) {
404 $self->_shutdown; 462 $self->_shutdown;
405 $self->{on_eof}($self); 463 $self->{on_eof}($self)
464 if $self->{on_eof};
406 } 465 }
407} 466}
408 467
409=item $handle->on_read ($cb) 468=item $handle->on_read ($cb)
410 469
… …
444Append 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
445prepend it (C<unshift_read>). 504prepend it (C<unshift_read>).
446 505
447The callback is called each time some additional read data arrives. 506The callback is called each time some additional read data arrives.
448 507
449It must check wether enough data is in the read buffer already. 508It must check whether enough data is in the read buffer already.
450 509
451If 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
452value, 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
453available (or an error condition is detected). 512available (or an error condition is detected).
454 513
… …
457true, it will be removed from the queue. 516true, it will be removed from the queue.
458 517
459=cut 518=cut
460 519
461sub push_read { 520sub push_read {
462 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 }
463 530
464 push @{ $self->{queue} }, $cb; 531 push @{ $self->{queue} }, $cb;
465 $self->_drain_rbuf; 532 $self->_drain_rbuf;
466} 533}
467 534
468sub unshift_read { 535sub unshift_read {
469 my ($self, $cb) = @_; 536 my $self = shift;
537 my $cb = pop;
470 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
471 push @{ $self->{queue} }, $cb; 547 unshift @{ $self->{queue} }, $cb;
472 $self->_drain_rbuf; 548 $self->_drain_rbuf;
473} 549}
474 550
475=item $handle->push_read_chunk ($len, $cb->($self, $data)) 551=item $handle->push_read (type => @args, $cb)
476 552
477=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) 553=item $handle->unshift_read (type => @args, $cb)
478 554
479Append 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
480prepend it (C<unshift_read_chunk>). 556between a number of predefined parsing formats, for chunks of data, lines
557etc.
481 558
482The callback will be called only once C<$len> bytes have been read, and 559The types currently supported are:
483these C<$len> bytes will be passed to the callback.
484 560
485=cut 561=over 4
486 562
487sub _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 {
488 my ($self, $len, $cb) = @_; 578 my ($self, $cb, $len) = @_;
489 579
490 sub { 580 sub {
491 $len <= length $_[0]{rbuf} or return; 581 $len <= length $_[0]{rbuf} or return;
492 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 582 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
493 1 583 1
494 } 584 }
495} 585};
496 586
587# compatibility with older API
497sub push_read_chunk { 588sub push_read_chunk {
498 $_[0]->push_read (&_read_chunk); 589 $_[0]->push_read (chunk => $_[1], $_[2]);
499} 590}
500
501 591
502sub unshift_read_chunk { 592sub unshift_read_chunk {
503 $_[0]->unshift_read (&_read_chunk); 593 $_[0]->unshift_read (chunk => $_[1], $_[2]);
504} 594}
505 595
506=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol)) 596=item line => [$eol, ]$cb->($self, $line, $eol)
507
508=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
509
510Append the given callback to the end of the queue (C<push_read_line>) or
511prepend it (C<unshift_read_line>).
512 597
513The 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
514line 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
515marker) 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
516the end of line marker as the third argument (C<$eol>). 601the end of line marker as the third argument (C<$eol>).
… …
527Partial 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
528not marked by the end of line marker. 613not marked by the end of line marker.
529 614
530=cut 615=cut
531 616
532sub _read_line($$) { 617register_read_type line => sub {
533 my $self = shift; 618 my ($self, $cb, $eol) = @_;
534 my $cb = pop;
535 my $eol = @_ ? shift : qr|(\015?\012)|;
536 my $pos;
537 619
620 $eol = qr|(\015?\012)| if @_ < 3;
538 $eol = quotemeta $eol unless ref $eol; 621 $eol = quotemeta $eol unless ref $eol;
539 $eol = qr|^(.*?)($eol)|s; 622 $eol = qr|^(.*?)($eol)|s;
540 623
541 sub { 624 sub {
542 $_[0]{rbuf} =~ s/$eol// or return; 625 $_[0]{rbuf} =~ s/$eol// or return;
543 626
544 $cb->($_[0], $1, $2); 627 $cb->($_[0], $1, $2);
545 1 628 1
546 } 629 }
547} 630};
548 631
632# compatibility with older API
549sub push_read_line { 633sub push_read_line {
550 $_[0]->push_read (&_read_line); 634 my $self = shift;
635 $self->push_read (line => @_);
551} 636}
552 637
553sub unshift_read_line { 638sub unshift_read_line {
554 $_[0]->unshift_read (&_read_line); 639 my $self = shift;
640 $self->unshift_read (line => @_);
555} 641}
642
643=back
556 644
557=item $handle->stop_read 645=item $handle->stop_read
558 646
559=item $handle->start_read 647=item $handle->start_read
560 648
561In 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
562socket. 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
563any queued callbacks will be executed then. To start readign again, call 651any queued callbacks will be executed then. To start reading again, call
564C<start_read>. 652C<start_read>.
565 653
566=cut 654=cut
567 655
568sub stop_read { 656sub stop_read {
… …
576 664
577 unless ($self->{rw} || $self->{eof}) { 665 unless ($self->{rw} || $self->{eof}) {
578 Scalar::Util::weaken $self; 666 Scalar::Util::weaken $self;
579 667
580 $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};
581 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;
582 671
583 if ($len > 0) { 672 if ($len > 0) {
584 if (defined $self->{rbuf_max}) { 673 $self->{filter_r}
585 if ($self->{rbuf_max} < length $self->{rbuf}) { 674 ? $self->{filter_r}->($self, $rbuf)
586 $! = &Errno::ENOSPC; return $self->error; 675 : $self->_drain_rbuf;
587 }
588 }
589 676
590 } elsif (defined $len) { 677 } elsif (defined $len) {
678 delete $self->{rw};
591 $self->{eof} = 1; 679 $self->{eof} = 1;
592 delete $self->{rw}; 680 $self->_drain_rbuf;
593 681
594 } elsif ($! != EAGAIN && $! != EINTR) { 682 } elsif ($! != EAGAIN && $! != EINTR) {
595 return $self->error; 683 return $self->error;
596 } 684 }
597
598 $self->_drain_rbuf;
599 }); 685 });
600 } 686 }
601} 687}
602 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
603=back 835=back
604 836
605=head1 AUTHOR 837=head1 AUTHOR
606 838
607Robin 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