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.34 by root, Mon May 26 03:27:52 2008 UTC vs.
Revision 1.38 by root, Mon May 26 21:28:33 2008 UTC

… …
93called. 93called.
94 94
95On callback entrance, the value of C<$!> contains the operating system 95On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>). 96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>).
97 97
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100
98While not mandatory, it is I<highly> recommended to set this callback, as 101While not mandatory, it is I<highly> recommended to set this callback, as
99you will not be notified of errors otherwise. The default simply calls 102you will not be notified of errors otherwise. The default simply calls
100die. 103die.
101 104
102=item on_read => $cb->($self) 105=item on_read => $cb->($self)
… …
165 168
166Use the given Net::SSLeay::CTX object to create the new TLS connection 169Use the given Net::SSLeay::CTX object to create the new TLS connection
167(unless a connection object was specified directly). If this parameter is 170(unless a connection object was specified directly). If this parameter is
168missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 171missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
169 172
173=item filter_r => $cb
174
175=item filter_w => $cb
176
177These exist, but are undocumented at this time.
178
170=back 179=back
171 180
172=cut 181=cut
173 182
174sub new { 183sub new {
… …
196} 205}
197 206
198sub _shutdown { 207sub _shutdown {
199 my ($self) = @_; 208 my ($self) = @_;
200 209
201 delete $self->{rw}; 210 delete $self->{_rw};
202 delete $self->{ww}; 211 delete $self->{_ww};
203 delete $self->{fh}; 212 delete $self->{fh};
204} 213}
205 214
206sub error { 215sub error {
207 my ($self) = @_; 216 my ($self) = @_;
… …
209 { 218 {
210 local $!; 219 local $!;
211 $self->_shutdown; 220 $self->_shutdown;
212 } 221 }
213 222
214 if ($self->{on_error}) {
215 $self->{on_error}($self); 223 $self->{on_error}($self)
216 } else { 224 if $self->{on_error};
225
217 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 226 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
218 }
219} 227}
220 228
221=item $fh = $handle->fh 229=item $fh = $handle->fh
222 230
223This method returns the file handle of the L<AnyEvent::Handle> object. 231This method returns the file handle of the L<AnyEvent::Handle> object.
224 232
225=cut 233=cut
226 234
227sub fh { $_[0]->{fh} } 235sub fh { $_[0]{fh} }
228 236
229=item $handle->on_error ($cb) 237=item $handle->on_error ($cb)
230 238
231Replace the current C<on_error> callback (see the C<on_error> constructor argument). 239Replace the current C<on_error> callback (see the C<on_error> constructor argument).
232 240
… …
288=cut 296=cut
289 297
290sub _drain_wbuf { 298sub _drain_wbuf {
291 my ($self) = @_; 299 my ($self) = @_;
292 300
293 if (!$self->{ww} && length $self->{wbuf}) { 301 if (!$self->{_ww} && length $self->{wbuf}) {
302
294 Scalar::Util::weaken $self; 303 Scalar::Util::weaken $self;
304
295 my $cb = sub { 305 my $cb = sub {
296 my $len = syswrite $self->{fh}, $self->{wbuf}; 306 my $len = syswrite $self->{fh}, $self->{wbuf};
297 307
298 if ($len >= 0) { 308 if ($len >= 0) {
299 substr $self->{wbuf}, 0, $len, ""; 309 substr $self->{wbuf}, 0, $len, "";
300 310
301 $self->{on_drain}($self) 311 $self->{on_drain}($self)
302 if $self->{low_water_mark} >= length $self->{wbuf} 312 if $self->{low_water_mark} >= length $self->{wbuf}
303 && $self->{on_drain}; 313 && $self->{on_drain};
304 314
305 delete $self->{ww} unless length $self->{wbuf}; 315 delete $self->{_ww} unless length $self->{wbuf};
306 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 316 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
307 $self->error; 317 $self->error;
308 } 318 }
309 }; 319 };
310 320
321 # try to write data immediately
322 $cb->();
323
324 # if still data left in wbuf, we need to poll
311 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 325 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
312 326 if length $self->{wbuf};
313 $cb->($self);
314 }; 327 };
315} 328}
316 329
317our %WH; 330our %WH;
318 331
… …
464 477
465 if ( 478 if (
466 defined $self->{rbuf_max} 479 defined $self->{rbuf_max}
467 && $self->{rbuf_max} < length $self->{rbuf} 480 && $self->{rbuf_max} < length $self->{rbuf}
468 ) { 481 ) {
469 $! = &Errno::ENOSPC; return $self->error; 482 $! = &Errno::ENOSPC;
483 $self->error;
470 } 484 }
471 485
472 return if $self->{in_drain}; 486 return if $self->{in_drain};
473 local $self->{in_drain} = 1; 487 local $self->{in_drain} = 1;
474 488
475 while (my $len = length $self->{rbuf}) { 489 while (my $len = length $self->{rbuf}) {
476 no strict 'refs'; 490 no strict 'refs';
477 if (my $cb = shift @{ $self->{queue} }) { 491 if (my $cb = shift @{ $self->{_queue} }) {
478 unless ($cb->($self)) { 492 unless ($cb->($self)) {
479 if ($self->{eof}) { 493 if ($self->{_eof}) {
480 # no progress can be made (not enough data and no data forthcoming) 494 # no progress can be made (not enough data and no data forthcoming)
481 $! = &Errno::EPIPE; return $self->error; 495 $! = &Errno::EPIPE;
496 $self->error;
482 } 497 }
483 498
484 unshift @{ $self->{queue} }, $cb; 499 unshift @{ $self->{_queue} }, $cb;
485 return; 500 return;
486 } 501 }
487 } elsif ($self->{on_read}) { 502 } elsif ($self->{on_read}) {
488 $self->{on_read}($self); 503 $self->{on_read}($self);
489 504
490 if ( 505 if (
491 $self->{eof} # if no further data will arrive 506 $self->{_eof} # if no further data will arrive
492 && $len == length $self->{rbuf} # and no data has been consumed 507 && $len == length $self->{rbuf} # and no data has been consumed
493 && !@{ $self->{queue} } # and the queue is still empty 508 && !@{ $self->{_queue} } # and the queue is still empty
494 && $self->{on_read} # and we still want to read data 509 && $self->{on_read} # and we still want to read data
495 ) { 510 ) {
496 # then no progress can be made 511 # then no progress can be made
497 $! = &Errno::EPIPE; return $self->error; 512 $! = &Errno::EPIPE;
513 $self->error;
498 } 514 }
499 } else { 515 } else {
500 # read side becomes idle 516 # read side becomes idle
501 delete $self->{rw}; 517 delete $self->{_rw};
502 return; 518 return;
503 } 519 }
504 } 520 }
505 521
506 if ($self->{eof}) { 522 if ($self->{_eof}) {
507 $self->_shutdown; 523 $self->_shutdown;
508 $self->{on_eof}($self) 524 $self->{on_eof}($self)
509 if $self->{on_eof}; 525 if $self->{on_eof};
510 } 526 }
511} 527}
… …
577 593
578 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 594 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
579 ->($self, $cb, @_); 595 ->($self, $cb, @_);
580 } 596 }
581 597
582 push @{ $self->{queue} }, $cb; 598 push @{ $self->{_queue} }, $cb;
583 $self->_drain_rbuf; 599 $self->_drain_rbuf;
584} 600}
585 601
586sub unshift_read { 602sub unshift_read {
587 my $self = shift; 603 my $self = shift;
… …
593 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 609 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
594 ->($self, $cb, @_); 610 ->($self, $cb, @_);
595 } 611 }
596 612
597 613
598 unshift @{ $self->{queue} }, $cb; 614 unshift @{ $self->{_queue} }, $cb;
599 $self->_drain_rbuf; 615 $self->_drain_rbuf;
600} 616}
601 617
602=item $handle->push_read (type => @args, $cb) 618=item $handle->push_read (type => @args, $cb)
603 619
… …
728 744
729 1 745 1
730 } 746 }
731}; 747};
732 748
749=item regex => $accept[, $reject[, $skip], $cb->($data)
750
751Makes a regex match against the regex object C<$accept> and returns
752everything up to and including the match.
753
754Example: read a single line terminated by '\n'.
755
756 $handle->push_read (regex => qr<\n>, sub { ... });
757
758If C<$reject> is given and not undef, then it determines when the data is
759to be rejected: it is matched against the data when the C<$accept> regex
760does not match and generates an C<EBADMSG> error when it matches. This is
761useful to quickly reject wrong data (to avoid waiting for a timeout or a
762receive buffer overflow).
763
764Example: expect a single decimal number followed by whitespace, reject
765anything else (not the use of an anchor).
766
767 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
768
769If C<$skip> is given and not C<undef>, then it will be matched against
770the receive buffer when neither C<$accept> nor C<$reject> match,
771and everything preceding and including the match will be accepted
772unconditionally. This is useful to skip large amounts of data that you
773know cannot be matched, so that the C<$accept> or C<$reject> regex do not
774have to start matching from the beginning. This is purely an optimisation
775and is usually worth only when you expect more than a few kilobytes.
776
777Example: expect a http header, which ends at C<\015\012\015\012>. Since we
778expect the header to be very large (it isn't in practise, but...), we use
779a skip regex to skip initial portions. The skip regex is tricky in that
780it only accepts something not ending in either \015 or \012, as these are
781required for the accept regex.
782
783 $handle->push_read (regex =>
784 qr<\015\012\015\012>,
785 undef, # no reject
786 qr<^.*[^\015\012]>,
787 sub { ... });
788
789=cut
790
791register_read_type regex => sub {
792 my ($self, $cb, $accept, $reject, $skip) = @_;
793
794 my $data;
795 my $rbuf = \$self->{rbuf};
796
797 sub {
798 # accept
799 if ($$rbuf =~ $accept) {
800 $data .= substr $$rbuf, 0, $+[0], "";
801 $cb->($self, $data);
802 return 1;
803 }
804
805 # reject
806 if ($reject && $$rbuf =~ $reject) {
807 $! = &Errno::EBADMSG;
808 $self->error;
809 }
810
811 # skip
812 if ($skip && $$rbuf =~ $skip) {
813 $data .= substr $$rbuf, 0, $+[0], "";
814 }
815
816 ()
817 }
818};
819
733=back 820=back
734 821
735=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args) 822=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
736 823
737This function (not method) lets you add your own types to C<push_read>. 824This function (not method) lets you add your own types to C<push_read>.
… …
764=cut 851=cut
765 852
766sub stop_read { 853sub stop_read {
767 my ($self) = @_; 854 my ($self) = @_;
768 855
769 delete $self->{rw}; 856 delete $self->{_rw};
770} 857}
771 858
772sub start_read { 859sub start_read {
773 my ($self) = @_; 860 my ($self) = @_;
774 861
775 unless ($self->{rw} || $self->{eof}) { 862 unless ($self->{_rw} || $self->{_eof}) {
776 Scalar::Util::weaken $self; 863 Scalar::Util::weaken $self;
777 864
778 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 865 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
779 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 866 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
780 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 867 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
781 868
782 if ($len > 0) { 869 if ($len > 0) {
783 $self->{filter_r} 870 $self->{filter_r}
784 ? $self->{filter_r}->($self, $rbuf) 871 ? $self->{filter_r}->($self, $rbuf)
785 : $self->_drain_rbuf; 872 : $self->_drain_rbuf;
786 873
787 } elsif (defined $len) { 874 } elsif (defined $len) {
788 delete $self->{rw}; 875 delete $self->{_rw};
789 $self->{eof} = 1; 876 $self->{_eof} = 1;
790 $self->_drain_rbuf; 877 $self->_drain_rbuf;
791 878
792 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 879 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
793 return $self->error; 880 return $self->error;
794 } 881 }
… …
797} 884}
798 885
799sub _dotls { 886sub _dotls {
800 my ($self) = @_; 887 my ($self) = @_;
801 888
802 if (length $self->{tls_wbuf}) { 889 if (length $self->{_tls_wbuf}) {
803 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) { 890 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
804 substr $self->{tls_wbuf}, 0, $len, ""; 891 substr $self->{_tls_wbuf}, 0, $len, "";
805 } 892 }
806 } 893 }
807 894
808 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) { 895 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
809 $self->{wbuf} .= $buf; 896 $self->{wbuf} .= $buf;
810 $self->_drain_wbuf; 897 $self->_drain_wbuf;
811 } 898 }
812 899
813 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 900 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
… …
838The first argument is the same as the C<tls> constructor argument (either 925The first argument is the same as the C<tls> constructor argument (either
839C<"connect">, C<"accept"> or an existing Net::SSLeay object). 926C<"connect">, C<"accept"> or an existing Net::SSLeay object).
840 927
841The second argument is the optional C<Net::SSLeay::CTX> object that is 928The second argument is the optional C<Net::SSLeay::CTX> object that is
842used when AnyEvent::Handle has to create its own TLS connection object. 929used when AnyEvent::Handle has to create its own TLS connection object.
930
931The TLS connection object will end up in C<< $handle->{tls} >> after this
932call and can be used or changed to your liking. Note that the handshake
933might have already started when this function returns.
843 934
844=cut 935=cut
845 936
846# TODO: maybe document... 937# TODO: maybe document...
847sub starttls { 938sub starttls {
… …
866 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 957 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
867 Net::SSLeay::CTX_set_mode ($self->{tls}, 958 Net::SSLeay::CTX_set_mode ($self->{tls},
868 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 959 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
869 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 960 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
870 961
871 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 962 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
872 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 963 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
873 964
874 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 965 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
875 966
876 $self->{filter_w} = sub { 967 $self->{filter_w} = sub {
877 $_[0]{tls_wbuf} .= ${$_[1]}; 968 $_[0]{_tls_wbuf} .= ${$_[1]};
878 &_dotls; 969 &_dotls;
879 }; 970 };
880 $self->{filter_r} = sub { 971 $self->{filter_r} = sub {
881 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]}); 972 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
882 &_dotls; 973 &_dotls;
883 }; 974 };
884} 975}
885 976
886=item $handle->stoptls 977=item $handle->stoptls
… …
892 983
893sub stoptls { 984sub stoptls {
894 my ($self) = @_; 985 my ($self) = @_;
895 986
896 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 987 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
988
897 delete $self->{tls_rbio}; 989 delete $self->{_rbio};
898 delete $self->{tls_wbio}; 990 delete $self->{_wbio};
899 delete $self->{tls_wbuf}; 991 delete $self->{_tls_wbuf};
900 delete $self->{filter_r}; 992 delete $self->{filter_r};
901 delete $self->{filter_w}; 993 delete $self->{filter_w};
902} 994}
903 995
904sub DESTROY { 996sub DESTROY {
… …
942 } 1034 }
943} 1035}
944 1036
945=back 1037=back
946 1038
1039=head1 SUBCLASSING AnyEvent::Handle
1040
1041In many cases, you might want to subclass AnyEvent::Handle.
1042
1043To make this easier, a given version of AnyEvent::Handle uses these
1044conventions:
1045
1046=over 4
1047
1048=item * all constructor arguments become object members.
1049
1050At least initially, when you pass a C<tls>-argument to the constructor it
1051will end up in C<< $handle->{tls} >>. Those members might be changes or
1052mutated later on (for example C<tls> will hold the TLS connection object).
1053
1054=item * other object member names are prefixed with an C<_>.
1055
1056All object members not explicitly documented (internal use) are prefixed
1057with an underscore character, so the remaining non-C<_>-namespace is free
1058for use for subclasses.
1059
1060=item * all members not documented here and not prefixed with an underscore
1061are free to use in subclasses.
1062
1063Of course, new versions of AnyEvent::Handle may introduce more "public"
1064member variables, but thats just life, at least it is documented.
1065
1066=back
1067
947=head1 AUTHOR 1068=head1 AUTHOR
948 1069
949Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 1070Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
950 1071
951=cut 1072=cut

Diff Legend

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