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.31 by root, Sun May 25 00:08:49 2008 UTC vs.
Revision 1.39 by root, Tue May 27 04:59:51 2008 UTC

… …
2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util (); 7use AnyEvent::Util qw(WSAWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw/EAGAIN EINTR/;
12 12
… …
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) { 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
… …
362register_write_type netstring => sub { 375register_write_type netstring => sub {
363 my ($self, $string) = @_; 376 my ($self, $string) = @_;
364 377
365 sprintf "%d:%s,", (length $string), $string 378 sprintf "%d:%s,", (length $string), $string
366}; 379};
380
381=item json => $array_or_hashref
367 382
368=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args) 383=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
369 384
370This function (not method) lets you add your own types to C<push_write>. 385This function (not method) lets you add your own types to C<push_write>.
371Whenever the given C<type> is used, C<push_write> will invoke the code 386Whenever the given C<type> is used, C<push_write> will invoke the code
… …
464 479
465 if ( 480 if (
466 defined $self->{rbuf_max} 481 defined $self->{rbuf_max}
467 && $self->{rbuf_max} < length $self->{rbuf} 482 && $self->{rbuf_max} < length $self->{rbuf}
468 ) { 483 ) {
469 $! = &Errno::ENOSPC; return $self->error; 484 $! = &Errno::ENOSPC;
485 $self->error;
470 } 486 }
471 487
472 return if $self->{in_drain}; 488 return if $self->{in_drain};
473 local $self->{in_drain} = 1; 489 local $self->{in_drain} = 1;
474 490
475 while (my $len = length $self->{rbuf}) { 491 while (my $len = length $self->{rbuf}) {
476 no strict 'refs'; 492 no strict 'refs';
477 if (my $cb = shift @{ $self->{queue} }) { 493 if (my $cb = shift @{ $self->{_queue} }) {
478 unless ($cb->($self)) { 494 unless ($cb->($self)) {
479 if ($self->{eof}) { 495 if ($self->{_eof}) {
480 # no progress can be made (not enough data and no data forthcoming) 496 # no progress can be made (not enough data and no data forthcoming)
481 $! = &Errno::EPIPE; return $self->error; 497 $! = &Errno::EPIPE;
498 $self->error;
482 } 499 }
483 500
484 unshift @{ $self->{queue} }, $cb; 501 unshift @{ $self->{_queue} }, $cb;
485 return; 502 return;
486 } 503 }
487 } elsif ($self->{on_read}) { 504 } elsif ($self->{on_read}) {
488 $self->{on_read}($self); 505 $self->{on_read}($self);
489 506
490 if ( 507 if (
491 $self->{eof} # if no further data will arrive 508 $self->{_eof} # if no further data will arrive
492 && $len == length $self->{rbuf} # and no data has been consumed 509 && $len == length $self->{rbuf} # and no data has been consumed
493 && !@{ $self->{queue} } # and the queue is still empty 510 && !@{ $self->{_queue} } # and the queue is still empty
494 && $self->{on_read} # and we still want to read data 511 && $self->{on_read} # and we still want to read data
495 ) { 512 ) {
496 # then no progress can be made 513 # then no progress can be made
497 $! = &Errno::EPIPE; return $self->error; 514 $! = &Errno::EPIPE;
515 $self->error;
498 } 516 }
499 } else { 517 } else {
500 # read side becomes idle 518 # read side becomes idle
501 delete $self->{rw}; 519 delete $self->{_rw};
502 return; 520 return;
503 } 521 }
504 } 522 }
505 523
506 if ($self->{eof}) { 524 if ($self->{_eof}) {
507 $self->_shutdown; 525 $self->_shutdown;
508 $self->{on_eof}($self) 526 $self->{on_eof}($self)
509 if $self->{on_eof}; 527 if $self->{on_eof};
510 } 528 }
511} 529}
… …
577 595
578 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 596 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
579 ->($self, $cb, @_); 597 ->($self, $cb, @_);
580 } 598 }
581 599
582 push @{ $self->{queue} }, $cb; 600 push @{ $self->{_queue} }, $cb;
583 $self->_drain_rbuf; 601 $self->_drain_rbuf;
584} 602}
585 603
586sub unshift_read { 604sub unshift_read {
587 my $self = shift; 605 my $self = shift;
… …
593 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 611 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
594 ->($self, $cb, @_); 612 ->($self, $cb, @_);
595 } 613 }
596 614
597 615
598 unshift @{ $self->{queue} }, $cb; 616 unshift @{ $self->{_queue} }, $cb;
599 $self->_drain_rbuf; 617 $self->_drain_rbuf;
600} 618}
601 619
602=item $handle->push_read (type => @args, $cb) 620=item $handle->push_read (type => @args, $cb)
603 621
… …
728 746
729 1 747 1
730 } 748 }
731}; 749};
732 750
751=item regex => $accept[, $reject[, $skip], $cb->($data)
752
753Makes a regex match against the regex object C<$accept> and returns
754everything up to and including the match.
755
756Example: read a single line terminated by '\n'.
757
758 $handle->push_read (regex => qr<\n>, sub { ... });
759
760If C<$reject> is given and not undef, then it determines when the data is
761to be rejected: it is matched against the data when the C<$accept> regex
762does not match and generates an C<EBADMSG> error when it matches. This is
763useful to quickly reject wrong data (to avoid waiting for a timeout or a
764receive buffer overflow).
765
766Example: expect a single decimal number followed by whitespace, reject
767anything else (not the use of an anchor).
768
769 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
770
771If C<$skip> is given and not C<undef>, then it will be matched against
772the receive buffer when neither C<$accept> nor C<$reject> match,
773and everything preceding and including the match will be accepted
774unconditionally. This is useful to skip large amounts of data that you
775know cannot be matched, so that the C<$accept> or C<$reject> regex do not
776have to start matching from the beginning. This is purely an optimisation
777and is usually worth only when you expect more than a few kilobytes.
778
779Example: expect a http header, which ends at C<\015\012\015\012>. Since we
780expect the header to be very large (it isn't in practise, but...), we use
781a skip regex to skip initial portions. The skip regex is tricky in that
782it only accepts something not ending in either \015 or \012, as these are
783required for the accept regex.
784
785 $handle->push_read (regex =>
786 qr<\015\012\015\012>,
787 undef, # no reject
788 qr<^.*[^\015\012]>,
789 sub { ... });
790
791=cut
792
793register_read_type regex => sub {
794 my ($self, $cb, $accept, $reject, $skip) = @_;
795
796 my $data;
797 my $rbuf = \$self->{rbuf};
798
799 sub {
800 # accept
801 if ($$rbuf =~ $accept) {
802 $data .= substr $$rbuf, 0, $+[0], "";
803 $cb->($self, $data);
804 return 1;
805 }
806
807 # reject
808 if ($reject && $$rbuf =~ $reject) {
809 $! = &Errno::EBADMSG;
810 $self->error;
811 }
812
813 # skip
814 if ($skip && $$rbuf =~ $skip) {
815 $data .= substr $$rbuf, 0, $+[0], "";
816 }
817
818 ()
819 }
820};
821
733=back 822=back
734 823
735=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args) 824=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
736 825
737This function (not method) lets you add your own types to C<push_read>. 826This function (not method) lets you add your own types to C<push_read>.
… …
764=cut 853=cut
765 854
766sub stop_read { 855sub stop_read {
767 my ($self) = @_; 856 my ($self) = @_;
768 857
769 delete $self->{rw}; 858 delete $self->{_rw};
770} 859}
771 860
772sub start_read { 861sub start_read {
773 my ($self) = @_; 862 my ($self) = @_;
774 863
775 unless ($self->{rw} || $self->{eof}) { 864 unless ($self->{_rw} || $self->{_eof}) {
776 Scalar::Util::weaken $self; 865 Scalar::Util::weaken $self;
777 866
778 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 867 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
779 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 868 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
780 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 869 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
781 870
782 if ($len > 0) { 871 if ($len > 0) {
783 $self->{filter_r} 872 $self->{filter_r}
784 ? $self->{filter_r}->($self, $rbuf) 873 ? $self->{filter_r}->($self, $rbuf)
785 : $self->_drain_rbuf; 874 : $self->_drain_rbuf;
786 875
787 } elsif (defined $len) { 876 } elsif (defined $len) {
788 delete $self->{rw}; 877 delete $self->{_rw};
789 $self->{eof} = 1; 878 $self->{_eof} = 1;
790 $self->_drain_rbuf; 879 $self->_drain_rbuf;
791 880
792 } elsif ($! != EAGAIN && $! != EINTR) { 881 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
793 return $self->error; 882 return $self->error;
794 } 883 }
795 }); 884 });
796 } 885 }
797} 886}
798 887
799sub _dotls { 888sub _dotls {
800 my ($self) = @_; 889 my ($self) = @_;
801 890
802 if (length $self->{tls_wbuf}) { 891 if (length $self->{_tls_wbuf}) {
803 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) { 892 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
804 substr $self->{tls_wbuf}, 0, $len, ""; 893 substr $self->{_tls_wbuf}, 0, $len, "";
805 } 894 }
806 } 895 }
807 896
808 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) { 897 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
809 $self->{wbuf} .= $buf; 898 $self->{wbuf} .= $buf;
810 $self->_drain_wbuf; 899 $self->_drain_wbuf;
811 } 900 }
812 901
813 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 902 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
… …
839C<"connect">, C<"accept"> or an existing Net::SSLeay object). 928C<"connect">, C<"accept"> or an existing Net::SSLeay object).
840 929
841The second argument is the optional C<Net::SSLeay::CTX> object that is 930The second argument is the optional C<Net::SSLeay::CTX> object that is
842used when AnyEvent::Handle has to create its own TLS connection object. 931used when AnyEvent::Handle has to create its own TLS connection object.
843 932
933The TLS connection object will end up in C<< $handle->{tls} >> after this
934call and can be used or changed to your liking. Note that the handshake
935might have already started when this function returns.
936
844=cut 937=cut
845 938
846# TODO: maybe document... 939# TODO: maybe document...
847sub starttls { 940sub starttls {
848 my ($self, $ssl, $ctx) = @_; 941 my ($self, $ssl, $ctx) = @_;
… …
863 # but the openssl maintainers basically said: "trust us, it just works". 956 # but the openssl maintainers basically said: "trust us, it just works".
864 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 957 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
865 # and mismaintained ssleay-module doesn't even offer them). 958 # and mismaintained ssleay-module doesn't even offer them).
866 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 959 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
867 Net::SSLeay::CTX_set_mode ($self->{tls}, 960 Net::SSLeay::CTX_set_mode ($self->{tls},
868 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 961 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
869 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 962 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
870 963
871 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 964 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
872 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 965 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
873 966
874 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 967 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
875 968
876 $self->{filter_w} = sub { 969 $self->{filter_w} = sub {
877 $_[0]{tls_wbuf} .= ${$_[1]}; 970 $_[0]{_tls_wbuf} .= ${$_[1]};
878 &_dotls; 971 &_dotls;
879 }; 972 };
880 $self->{filter_r} = sub { 973 $self->{filter_r} = sub {
881 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]}); 974 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
882 &_dotls; 975 &_dotls;
883 }; 976 };
884} 977}
885 978
886=item $handle->stoptls 979=item $handle->stoptls
… …
892 985
893sub stoptls { 986sub stoptls {
894 my ($self) = @_; 987 my ($self) = @_;
895 988
896 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 989 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
990
897 delete $self->{tls_rbio}; 991 delete $self->{_rbio};
898 delete $self->{tls_wbio}; 992 delete $self->{_wbio};
899 delete $self->{tls_wbuf}; 993 delete $self->{_tls_wbuf};
900 delete $self->{filter_r}; 994 delete $self->{filter_r};
901 delete $self->{filter_w}; 995 delete $self->{filter_w};
902} 996}
903 997
904sub DESTROY { 998sub DESTROY {
… …
942 } 1036 }
943} 1037}
944 1038
945=back 1039=back
946 1040
1041=head1 SUBCLASSING AnyEvent::Handle
1042
1043In many cases, you might want to subclass AnyEvent::Handle.
1044
1045To make this easier, a given version of AnyEvent::Handle uses these
1046conventions:
1047
1048=over 4
1049
1050=item * all constructor arguments become object members.
1051
1052At least initially, when you pass a C<tls>-argument to the constructor it
1053will end up in C<< $handle->{tls} >>. Those members might be changes or
1054mutated later on (for example C<tls> will hold the TLS connection object).
1055
1056=item * other object member names are prefixed with an C<_>.
1057
1058All object members not explicitly documented (internal use) are prefixed
1059with an underscore character, so the remaining non-C<_>-namespace is free
1060for use for subclasses.
1061
1062=item * all members not documented here and not prefixed with an underscore
1063are free to use in subclasses.
1064
1065Of course, new versions of AnyEvent::Handle may introduce more "public"
1066member variables, but thats just life, at least it is documented.
1067
1068=back
1069
947=head1 AUTHOR 1070=head1 AUTHOR
948 1071
949Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 1072Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
950 1073
951=cut 1074=cut

Diff Legend

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