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.54 by root, Tue Jun 3 09:02:46 2008 UTC vs.
Revision 1.61 by root, Fri Jun 6 10:23:50 2008 UTC

… …
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = 4.12; 19our $VERSION = 4.14;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
… …
105C<croak>. 105C<croak>.
106 106
107=item on_read => $cb->($handle) 107=item on_read => $cb->($handle)
108 108
109This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
110and no read request is in the queue. 110and no read request is in the queue (unlike read queue callbacks, this
111callback will only be called when at least one octet of data is in the
112read buffer).
111 113
112To access (and remove data from) the read buffer, use the C<< ->rbuf >> 114To access (and remove data from) the read buffer, use the C<< ->rbuf >>
113method or access the C<$handle->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
114 116
115When an EOF condition is detected then AnyEvent::Handle will first try to 117When an EOF condition is detected then AnyEvent::Handle will first try to
… …
224 if ($self->{tls}) { 226 if ($self->{tls}) {
225 require Net::SSLeay; 227 require Net::SSLeay;
226 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 228 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
227 } 229 }
228 230
229# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
230# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
231# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
232 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
233
234 $self->{_activity} = AnyEvent->now; 231 $self->{_activity} = AnyEvent->now;
235 $self->_timeout; 232 $self->_timeout;
236 233
237 $self->start_read; 234 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
238 235
239 $self 236 $self
240} 237}
241 238
242sub _shutdown { 239sub _shutdown {
… …
339 $self->{on_timeout}($self); 336 $self->{on_timeout}($self);
340 } else { 337 } else {
341 $self->_error (&Errno::ETIMEDOUT); 338 $self->_error (&Errno::ETIMEDOUT);
342 } 339 }
343 340
344 # callbakx could have changed timeout value, optimise 341 # callback could have changed timeout value, optimise
345 return unless $self->{timeout}; 342 return unless $self->{timeout};
346 343
347 # calculate new after 344 # calculate new after
348 $after = $self->{timeout}; 345 $after = $self->{timeout};
349 } 346 }
350 347
351 Scalar::Util::weaken $self; 348 Scalar::Util::weaken $self;
349 return unless $self; # ->error could have destroyed $self
352 350
353 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 351 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
354 delete $self->{_tw}; 352 delete $self->{_tw};
355 $self->_timeout; 353 $self->_timeout;
356 }); 354 });
… …
479 my ($self, $string) = @_; 477 my ($self, $string) = @_;
480 478
481 sprintf "%d:%s,", (length $string), $string 479 sprintf "%d:%s,", (length $string), $string
482}; 480};
483 481
482=item packstring => $format, $data
483
484An octet string prefixed with an encoded length. The encoding C<$format>
485uses the same format as a Perl C<pack> format, but must specify a single
486integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
487optional C<!>, C<< < >> or C<< > >> modifier).
488
489=cut
490
491register_write_type packstring => sub {
492 my ($self, $format, $string) = @_;
493
494 pack "$format/a", $string
495};
496
484=item json => $array_or_hashref 497=item json => $array_or_hashref
485 498
486Encodes the given hash or array reference into a JSON object. Unless you 499Encodes the given hash or array reference into a JSON object. Unless you
487provide your own JSON object, this means it will be encoded to JSON text 500provide your own JSON object, this means it will be encoded to JSON text
488in UTF-8. 501in UTF-8.
… …
556enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 569enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
557or not. 570or not.
558 571
559In the more complex case, you want to queue multiple callbacks. In this 572In the more complex case, you want to queue multiple callbacks. In this
560case, AnyEvent::Handle will call the first queued callback each time new 573case, AnyEvent::Handle will call the first queued callback each time new
561data arrives and removes it when it has done its job (see C<push_read>, 574data arrives (also the first time it is queued) and removes it when it has
562below). 575done its job (see C<push_read>, below).
563 576
564This way you can, for example, push three line-reads, followed by reading 577This way you can, for example, push three line-reads, followed by reading
565a chunk of data, and AnyEvent::Handle will execute them in order. 578a chunk of data, and AnyEvent::Handle will execute them in order.
566 579
567Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 580Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
… …
619=cut 632=cut
620 633
621sub _drain_rbuf { 634sub _drain_rbuf {
622 my ($self) = @_; 635 my ($self) = @_;
623 636
637 local $self->{_in_drain} = 1;
638
624 if ( 639 if (
625 defined $self->{rbuf_max} 640 defined $self->{rbuf_max}
626 && $self->{rbuf_max} < length $self->{rbuf} 641 && $self->{rbuf_max} < length $self->{rbuf}
627 ) { 642 ) {
628 return $self->_error (&Errno::ENOSPC, 1); 643 return $self->_error (&Errno::ENOSPC, 1);
629 } 644 }
630 645
631 return if $self->{in_drain}; 646 while () {
632 local $self->{in_drain} = 1;
633
634 while (my $len = length $self->{rbuf}) {
635 no strict 'refs'; 647 no strict 'refs';
648
649 my $len = length $self->{rbuf};
650
636 if (my $cb = shift @{ $self->{_queue} }) { 651 if (my $cb = shift @{ $self->{_queue} }) {
637 unless ($cb->($self)) { 652 unless ($cb->($self)) {
638 if ($self->{_eof}) { 653 if ($self->{_eof}) {
639 # no progress can be made (not enough data and no data forthcoming) 654 # no progress can be made (not enough data and no data forthcoming)
640 return $self->_error (&Errno::EPIPE, 1); 655 $self->_error (&Errno::EPIPE, 1), last;
641 } 656 }
642 657
643 unshift @{ $self->{_queue} }, $cb; 658 unshift @{ $self->{_queue} }, $cb;
644 return; 659 last;
645 } 660 }
646 } elsif ($self->{on_read}) { 661 } elsif ($self->{on_read}) {
662 last unless $len;
663
647 $self->{on_read}($self); 664 $self->{on_read}($self);
648 665
649 if ( 666 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed 667 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 668 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data 669 && $self->{on_read} # but we still have on_read
654 ) { 670 ) {
671 # no further data will arrive
655 # then no progress can be made 672 # so no progress can be made
656 return $self->_error (&Errno::EPIPE, 1); 673 $self->_error (&Errno::EPIPE, 1), last
674 if $self->{_eof};
675
676 last; # more data might arrive
657 } 677 }
658 } else { 678 } else {
659 # read side becomes idle 679 # read side becomes idle
660 delete $self->{_rw}; 680 delete $self->{_rw};
661 return; 681 last;
662 } 682 }
663 } 683 }
664 684
665 $self->{on_eof}($self) 685 $self->{on_eof}($self)
666 if $self->{_eof} && $self->{on_eof}; 686 if $self->{_eof} && $self->{on_eof};
687
688 # may need to restart read watcher
689 unless ($self->{_rw}) {
690 $self->start_read
691 if $self->{on_read} || @{ $self->{_queue} };
692 }
667} 693}
668 694
669=item $handle->on_read ($cb) 695=item $handle->on_read ($cb)
670 696
671This replaces the currently set C<on_read> callback, or clears it (when 697This replaces the currently set C<on_read> callback, or clears it (when
… …
676 702
677sub on_read { 703sub on_read {
678 my ($self, $cb) = @_; 704 my ($self, $cb) = @_;
679 705
680 $self->{on_read} = $cb; 706 $self->{on_read} = $cb;
707 $self->_drain_rbuf if $cb && !$self->{_in_drain};
681} 708}
682 709
683=item $handle->rbuf 710=item $handle->rbuf
684 711
685Returns the read buffer (as a modifiable lvalue). 712Returns the read buffer (as a modifiable lvalue).
… …
734 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 761 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
735 ->($self, $cb, @_); 762 ->($self, $cb, @_);
736 } 763 }
737 764
738 push @{ $self->{_queue} }, $cb; 765 push @{ $self->{_queue} }, $cb;
739 $self->_drain_rbuf; 766 $self->_drain_rbuf unless $self->{_in_drain};
740} 767}
741 768
742sub unshift_read { 769sub unshift_read {
743 my $self = shift; 770 my $self = shift;
744 my $cb = pop; 771 my $cb = pop;
… …
750 ->($self, $cb, @_); 777 ->($self, $cb, @_);
751 } 778 }
752 779
753 780
754 unshift @{ $self->{_queue} }, $cb; 781 unshift @{ $self->{_queue} }, $cb;
755 $self->_drain_rbuf; 782 $self->_drain_rbuf unless $self->{_in_drain};
756} 783}
757 784
758=item $handle->push_read (type => @args, $cb) 785=item $handle->push_read (type => @args, $cb)
759 786
760=item $handle->unshift_read (type => @args, $cb) 787=item $handle->unshift_read (type => @args, $cb)
… …
845 872
846sub unshift_read_line { 873sub unshift_read_line {
847 my $self = shift; 874 my $self = shift;
848 $self->unshift_read (line => @_); 875 $self->unshift_read (line => @_);
849} 876}
850
851=item netstring => $cb->($handle, $string)
852
853A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
854
855Throws an error with C<$!> set to EBADMSG on format violations.
856
857=cut
858
859register_read_type netstring => sub {
860 my ($self, $cb) = @_;
861
862 sub {
863 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
864 if ($_[0]{rbuf} =~ /[^0-9]/) {
865 $self->_error (&Errno::EBADMSG);
866 }
867 return;
868 }
869
870 my $len = $1;
871
872 $self->unshift_read (chunk => $len, sub {
873 my $string = $_[1];
874 $_[0]->unshift_read (chunk => 1, sub {
875 if ($_[1] eq ",") {
876 $cb->($_[0], $string);
877 } else {
878 $self->_error (&Errno::EBADMSG);
879 }
880 });
881 });
882
883 1
884 }
885};
886 877
887=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 878=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
888 879
889Makes a regex match against the regex object C<$accept> and returns 880Makes a regex match against the regex object C<$accept> and returns
890everything up to and including the match. 881everything up to and including the match.
… …
952 943
953 () 944 ()
954 } 945 }
955}; 946};
956 947
948=item netstring => $cb->($handle, $string)
949
950A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
951
952Throws an error with C<$!> set to EBADMSG on format violations.
953
954=cut
955
956register_read_type netstring => sub {
957 my ($self, $cb) = @_;
958
959 sub {
960 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
961 if ($_[0]{rbuf} =~ /[^0-9]/) {
962 $self->_error (&Errno::EBADMSG);
963 }
964 return;
965 }
966
967 my $len = $1;
968
969 $self->unshift_read (chunk => $len, sub {
970 my $string = $_[1];
971 $_[0]->unshift_read (chunk => 1, sub {
972 if ($_[1] eq ",") {
973 $cb->($_[0], $string);
974 } else {
975 $self->_error (&Errno::EBADMSG);
976 }
977 });
978 });
979
980 1
981 }
982};
983
984=item packstring => $format, $cb->($handle, $string)
985
986An octet string prefixed with an encoded length. The encoding C<$format>
987uses the same format as a Perl C<pack> format, but must specify a single
988integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
989optional C<!>, C<< < >> or C<< > >> modifier).
990
991DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
992
993Example: read a block of data prefixed by its length in BER-encoded
994format (very efficient).
995
996 $handle->push_read (packstring => "w", sub {
997 my ($handle, $data) = @_;
998 });
999
1000=cut
1001
1002register_read_type packstring => sub {
1003 my ($self, $cb, $format) = @_;
1004
1005 sub {
1006 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1007 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1008 or return;
1009
1010 # remove prefix
1011 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1012
1013 # read rest
1014 $_[0]->unshift_read (chunk => $len, $cb);
1015
1016 1
1017 }
1018};
1019
957=item json => $cb->($handle, $hash_or_arrayref) 1020=item json => $cb->($handle, $hash_or_arrayref)
958 1021
959Reads a JSON object or array, decodes it and passes it to the callback. 1022Reads a JSON object or array, decodes it and passes it to the callback.
960 1023
961If a C<json> object was passed to the constructor, then that will be used 1024If a C<json> object was passed to the constructor, then that will be used
… …
1023=item $handle->stop_read 1086=item $handle->stop_read
1024 1087
1025=item $handle->start_read 1088=item $handle->start_read
1026 1089
1027In rare cases you actually do not want to read anything from the 1090In rare cases you actually do not want to read anything from the
1028socket. In this case you can call C<stop_read>. Neither C<on_read> no 1091socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1029any queued callbacks will be executed then. To start reading again, call 1092any queued callbacks will be executed then. To start reading again, call
1030C<start_read>. 1093C<start_read>.
1094
1095Note that AnyEvent::Handle will automatically C<start_read> for you when
1096you change the C<on_read> callback or push/unshift a read callback, and it
1097will automatically C<stop_read> for you when neither C<on_read> is set nor
1098there are any read requests in the queue.
1031 1099
1032=cut 1100=cut
1033 1101
1034sub stop_read { 1102sub stop_read {
1035 my ($self) = @_; 1103 my ($self) = @_;
… …
1050 if ($len > 0) { 1118 if ($len > 0) {
1051 $self->{_activity} = AnyEvent->now; 1119 $self->{_activity} = AnyEvent->now;
1052 1120
1053 $self->{filter_r} 1121 $self->{filter_r}
1054 ? $self->{filter_r}($self, $rbuf) 1122 ? $self->{filter_r}($self, $rbuf)
1055 : $self->_drain_rbuf; 1123 : $self->{_in_drain} || $self->_drain_rbuf;
1056 1124
1057 } elsif (defined $len) { 1125 } elsif (defined $len) {
1058 delete $self->{_rw}; 1126 delete $self->{_rw};
1059 $self->{_eof} = 1; 1127 $self->{_eof} = 1;
1060 $self->_drain_rbuf; 1128 $self->_drain_rbuf unless $self->{_in_drain};
1061 1129
1062 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1130 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1063 return $self->_error ($!, 1); 1131 return $self->_error ($!, 1);
1064 } 1132 }
1065 }); 1133 });
… …
1067} 1135}
1068 1136
1069sub _dotls { 1137sub _dotls {
1070 my ($self) = @_; 1138 my ($self) = @_;
1071 1139
1140 my $buf;
1141
1072 if (length $self->{_tls_wbuf}) { 1142 if (length $self->{_tls_wbuf}) {
1073 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1143 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1074 substr $self->{_tls_wbuf}, 0, $len, ""; 1144 substr $self->{_tls_wbuf}, 0, $len, "";
1075 } 1145 }
1076 } 1146 }
1077 1147
1078 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1148 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1079 $self->{wbuf} .= $buf; 1149 $self->{wbuf} .= $buf;
1080 $self->_drain_wbuf; 1150 $self->_drain_wbuf;
1081 } 1151 }
1082 1152
1083 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1153 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1154 if (length $buf) {
1084 $self->{rbuf} .= $buf; 1155 $self->{rbuf} .= $buf;
1085 $self->_drain_rbuf; 1156 $self->_drain_rbuf unless $self->{_in_drain};
1157 } else {
1158 # let's treat SSL-eof as we treat normal EOF
1159 $self->{_eof} = 1;
1160 $self->_shutdown;
1161 return;
1162 }
1086 } 1163 }
1087 1164
1088 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1165 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1089 1166
1090 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1167 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {

Diff Legend

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