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.58 by root, Wed Jun 4 22:51:15 2008 UTC vs.
Revision 1.69 by root, Sun Jun 15 21:44:56 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.13; 19our $VERSION = 4.151;
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
… …
121 123
122This sets the callback that is called when the write buffer becomes empty 124This sets the callback that is called when the write buffer becomes empty
123(or when the callback is set and the buffer is empty already). 125(or when the callback is set and the buffer is empty already).
124 126
125To append to the write buffer, use the C<< ->push_write >> method. 127To append to the write buffer, use the C<< ->push_write >> method.
128
129This callback is useful when you don't want to put all of your write data
130into the queue at once, for example, when you want to write the contents
131of some file to the socket you might not want to read the whole file into
132memory and push it into the queue, but instead only read more data from
133the file when the write queue becomes empty.
126 134
127=item timeout => $fractional_seconds 135=item timeout => $fractional_seconds
128 136
129If non-zero, then this enables an "inactivity" timeout: whenever this many 137If non-zero, then this enables an "inactivity" timeout: whenever this many
130seconds pass without a successful read or write on the underlying file 138seconds pass without a successful read or write on the underlying file
… …
165 173
166Sets the amount of bytes (default: C<0>) that make up an "empty" write 174Sets the amount of bytes (default: C<0>) that make up an "empty" write
167buffer: If the write reaches this size or gets even samller it is 175buffer: If the write reaches this size or gets even samller it is
168considered empty. 176considered empty.
169 177
178=item linger => <seconds>
179
180If non-zero (default: C<3600>), then the destructor of the
181AnyEvent::Handle object will check wether there is still outstanding write
182data and will install a watcher that will write out this data. No errors
183will be reported (this mostly matches how the operating system treats
184outstanding data at socket close time).
185
186This will not work for partial TLS data that could not yet been
187encoded. This data will be lost.
188
170=item tls => "accept" | "connect" | Net::SSLeay::SSL object 189=item tls => "accept" | "connect" | Net::SSLeay::SSL object
171 190
172When this parameter is given, it enables TLS (SSL) mode, that means it 191When this parameter is given, it enables TLS (SSL) mode, that means it
173will start making tls handshake and will transparently encrypt/decrypt 192will start making tls handshake and will transparently encrypt/decrypt
174data. 193data.
… …
228 247
229 $self->{_activity} = AnyEvent->now; 248 $self->{_activity} = AnyEvent->now;
230 $self->_timeout; 249 $self->_timeout;
231 250
232 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 251 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
233 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 252
253 $self->start_read
254 if $self->{on_read};
234 255
235 $self 256 $self
236} 257}
237 258
238sub _shutdown { 259sub _shutdown {
… …
476 my ($self, $string) = @_; 497 my ($self, $string) = @_;
477 498
478 sprintf "%d:%s,", (length $string), $string 499 sprintf "%d:%s,", (length $string), $string
479}; 500};
480 501
502=item packstring => $format, $data
503
504An octet string prefixed with an encoded length. The encoding C<$format>
505uses the same format as a Perl C<pack> format, but must specify a single
506integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
507optional C<!>, C<< < >> or C<< > >> modifier).
508
509=cut
510
511register_write_type packstring => sub {
512 my ($self, $format, $string) = @_;
513
514 pack "$format/a*", $string
515};
516
481=item json => $array_or_hashref 517=item json => $array_or_hashref
482 518
483Encodes the given hash or array reference into a JSON object. Unless you 519Encodes the given hash or array reference into a JSON object. Unless you
484provide your own JSON object, this means it will be encoded to JSON text 520provide your own JSON object, this means it will be encoded to JSON text
485in UTF-8. 521in UTF-8.
… …
517 553
518 $self->{json} ? $self->{json}->encode ($ref) 554 $self->{json} ? $self->{json}->encode ($ref)
519 : JSON::encode_json ($ref) 555 : JSON::encode_json ($ref)
520}; 556};
521 557
558=item storable => $reference
559
560Freezes the given reference using L<Storable> and writes it to the
561handle. Uses the C<nfreeze> format.
562
563=cut
564
565register_write_type storable => sub {
566 my ($self, $ref) = @_;
567
568 require Storable;
569
570 pack "w/a*", Storable::nfreeze ($ref)
571};
572
522=back 573=back
523 574
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 575=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525 576
526This function (not method) lets you add your own types to C<push_write>. 577This function (not method) lets you add your own types to C<push_write>.
… …
548ways, the "simple" way, using only C<on_read> and the "complex" way, using 599ways, the "simple" way, using only C<on_read> and the "complex" way, using
549a queue. 600a queue.
550 601
551In the simple case, you just install an C<on_read> callback and whenever 602In the simple case, you just install an C<on_read> callback and whenever
552new data arrives, it will be called. You can then remove some data (if 603new data arrives, it will be called. You can then remove some data (if
553enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 604enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
554or not. 605leave the data there if you want to accumulate more (e.g. when only a
606partial message has been received so far).
555 607
556In the more complex case, you want to queue multiple callbacks. In this 608In the more complex case, you want to queue multiple callbacks. In this
557case, AnyEvent::Handle will call the first queued callback each time new 609case, AnyEvent::Handle will call the first queued callback each time new
558data arrives and removes it when it has done its job (see C<push_read>, 610data arrives (also the first time it is queued) and removes it when it has
559below). 611done its job (see C<push_read>, below).
560 612
561This way you can, for example, push three line-reads, followed by reading 613This way you can, for example, push three line-reads, followed by reading
562a chunk of data, and AnyEvent::Handle will execute them in order. 614a chunk of data, and AnyEvent::Handle will execute them in order.
563 615
564Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 616Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
… …
577 # handle xml 629 # handle xml
578 }); 630 });
579 }); 631 });
580 }); 632 });
581 633
582Example 2: Implement a client for a protocol that replies either with 634Example 2: Implement a client for a protocol that replies either with "OK"
583"OK" and another line or "ERROR" for one request, and 64 bytes for the 635and another line or "ERROR" for the first request that is sent, and 64
584second request. Due tot he availability of a full queue, we can just 636bytes for the second request. Due to the availability of a queue, we can
585pipeline sending both requests and manipulate the queue as necessary in 637just pipeline sending both requests and manipulate the queue as necessary
586the callbacks: 638in the callbacks.
587 639
588 # request one 640When the first callback is called and sees an "OK" response, it will
641C<unshift> another line-read. This line-read will be queued I<before> the
64264-byte chunk callback.
643
644 # request one, returns either "OK + extra line" or "ERROR"
589 $handle->push_write ("request 1\015\012"); 645 $handle->push_write ("request 1\015\012");
590 646
591 # we expect "ERROR" or "OK" as response, so push a line read 647 # we expect "ERROR" or "OK" as response, so push a line read
592 $handle->push_read (line => sub { 648 $handle->push_read (line => sub {
593 # if we got an "OK", we have to _prepend_ another line, 649 # if we got an "OK", we have to _prepend_ another line,
… …
600 ... 656 ...
601 }); 657 });
602 } 658 }
603 }); 659 });
604 660
605 # request two 661 # request two, simply returns 64 octets
606 $handle->push_write ("request 2\015\012"); 662 $handle->push_write ("request 2\015\012");
607 663
608 # simply read 64 bytes, always 664 # simply read 64 bytes, always
609 $handle->push_read (chunk => 64, sub { 665 $handle->push_read (chunk => 64, sub {
610 my $response = $_[1]; 666 my $response = $_[1];
… …
616=cut 672=cut
617 673
618sub _drain_rbuf { 674sub _drain_rbuf {
619 my ($self) = @_; 675 my ($self) = @_;
620 676
677 local $self->{_in_drain} = 1;
678
621 if ( 679 if (
622 defined $self->{rbuf_max} 680 defined $self->{rbuf_max}
623 && $self->{rbuf_max} < length $self->{rbuf} 681 && $self->{rbuf_max} < length $self->{rbuf}
624 ) { 682 ) {
625 return $self->_error (&Errno::ENOSPC, 1); 683 return $self->_error (&Errno::ENOSPC, 1);
626 } 684 }
627 685
628 return if $self->{in_drain}; 686 while () {
629 local $self->{in_drain} = 1;
630
631 while (my $len = length $self->{rbuf}) {
632 no strict 'refs'; 687 no strict 'refs';
688
689 my $len = length $self->{rbuf};
690
633 if (my $cb = shift @{ $self->{_queue} }) { 691 if (my $cb = shift @{ $self->{_queue} }) {
634 unless ($cb->($self)) { 692 unless ($cb->($self)) {
635 if ($self->{_eof}) { 693 if ($self->{_eof}) {
636 # no progress can be made (not enough data and no data forthcoming) 694 # no progress can be made (not enough data and no data forthcoming)
637 return $self->_error (&Errno::EPIPE, 1); 695 $self->_error (&Errno::EPIPE, 1), last;
638 } 696 }
639 697
640 unshift @{ $self->{_queue} }, $cb; 698 unshift @{ $self->{_queue} }, $cb;
641 last; 699 last;
642 } 700 }
643 } elsif ($self->{on_read}) { 701 } elsif ($self->{on_read}) {
702 last unless $len;
703
644 $self->{on_read}($self); 704 $self->{on_read}($self);
645 705
646 if ( 706 if (
647 $len == length $self->{rbuf} # if no data has been consumed 707 $len == length $self->{rbuf} # if no data has been consumed
648 && !@{ $self->{_queue} } # and the queue is still empty 708 && !@{ $self->{_queue} } # and the queue is still empty
649 && $self->{on_read} # but we still have on_read 709 && $self->{on_read} # but we still have on_read
650 ) { 710 ) {
651 # no further data will arrive 711 # no further data will arrive
652 # so no progress can be made 712 # so no progress can be made
653 return $self->_error (&Errno::EPIPE, 1) 713 $self->_error (&Errno::EPIPE, 1), last
654 if $self->{_eof}; 714 if $self->{_eof};
655 715
656 last; # more data might arrive 716 last; # more data might arrive
657 } 717 }
658 } else { 718 } else {
… …
682 742
683sub on_read { 743sub on_read {
684 my ($self, $cb) = @_; 744 my ($self, $cb) = @_;
685 745
686 $self->{on_read} = $cb; 746 $self->{on_read} = $cb;
687 $self->_drain_rbuf if $cb; 747 $self->_drain_rbuf if $cb && !$self->{_in_drain};
688} 748}
689 749
690=item $handle->rbuf 750=item $handle->rbuf
691 751
692Returns the read buffer (as a modifiable lvalue). 752Returns the read buffer (as a modifiable lvalue).
… …
741 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 801 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
742 ->($self, $cb, @_); 802 ->($self, $cb, @_);
743 } 803 }
744 804
745 push @{ $self->{_queue} }, $cb; 805 push @{ $self->{_queue} }, $cb;
746 $self->_drain_rbuf; 806 $self->_drain_rbuf unless $self->{_in_drain};
747} 807}
748 808
749sub unshift_read { 809sub unshift_read {
750 my $self = shift; 810 my $self = shift;
751 my $cb = pop; 811 my $cb = pop;
… …
757 ->($self, $cb, @_); 817 ->($self, $cb, @_);
758 } 818 }
759 819
760 820
761 unshift @{ $self->{_queue} }, $cb; 821 unshift @{ $self->{_queue} }, $cb;
762 $self->_drain_rbuf; 822 $self->_drain_rbuf unless $self->{_in_drain};
763} 823}
764 824
765=item $handle->push_read (type => @args, $cb) 825=item $handle->push_read (type => @args, $cb)
766 826
767=item $handle->unshift_read (type => @args, $cb) 827=item $handle->unshift_read (type => @args, $cb)
… …
852 912
853sub unshift_read_line { 913sub unshift_read_line {
854 my $self = shift; 914 my $self = shift;
855 $self->unshift_read (line => @_); 915 $self->unshift_read (line => @_);
856} 916}
857
858=item netstring => $cb->($handle, $string)
859
860A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
861
862Throws an error with C<$!> set to EBADMSG on format violations.
863
864=cut
865
866register_read_type netstring => sub {
867 my ($self, $cb) = @_;
868
869 sub {
870 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
871 if ($_[0]{rbuf} =~ /[^0-9]/) {
872 $self->_error (&Errno::EBADMSG);
873 }
874 return;
875 }
876
877 my $len = $1;
878
879 $self->unshift_read (chunk => $len, sub {
880 my $string = $_[1];
881 $_[0]->unshift_read (chunk => 1, sub {
882 if ($_[1] eq ",") {
883 $cb->($_[0], $string);
884 } else {
885 $self->_error (&Errno::EBADMSG);
886 }
887 });
888 });
889
890 1
891 }
892};
893 917
894=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 918=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
895 919
896Makes a regex match against the regex object C<$accept> and returns 920Makes a regex match against the regex object C<$accept> and returns
897everything up to and including the match. 921everything up to and including the match.
… …
959 983
960 () 984 ()
961 } 985 }
962}; 986};
963 987
988=item netstring => $cb->($handle, $string)
989
990A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
991
992Throws an error with C<$!> set to EBADMSG on format violations.
993
994=cut
995
996register_read_type netstring => sub {
997 my ($self, $cb) = @_;
998
999 sub {
1000 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1001 if ($_[0]{rbuf} =~ /[^0-9]/) {
1002 $self->_error (&Errno::EBADMSG);
1003 }
1004 return;
1005 }
1006
1007 my $len = $1;
1008
1009 $self->unshift_read (chunk => $len, sub {
1010 my $string = $_[1];
1011 $_[0]->unshift_read (chunk => 1, sub {
1012 if ($_[1] eq ",") {
1013 $cb->($_[0], $string);
1014 } else {
1015 $self->_error (&Errno::EBADMSG);
1016 }
1017 });
1018 });
1019
1020 1
1021 }
1022};
1023
1024=item packstring => $format, $cb->($handle, $string)
1025
1026An octet string prefixed with an encoded length. The encoding C<$format>
1027uses the same format as a Perl C<pack> format, but must specify a single
1028integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1029optional C<!>, C<< < >> or C<< > >> modifier).
1030
1031DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1032
1033Example: read a block of data prefixed by its length in BER-encoded
1034format (very efficient).
1035
1036 $handle->push_read (packstring => "w", sub {
1037 my ($handle, $data) = @_;
1038 });
1039
1040=cut
1041
1042register_read_type packstring => sub {
1043 my ($self, $cb, $format) = @_;
1044
1045 sub {
1046 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1047 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1048 or return;
1049
1050 # remove prefix
1051 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1052
1053 # read rest
1054 $_[0]->unshift_read (chunk => $len, $cb);
1055
1056 1
1057 }
1058};
1059
964=item json => $cb->($handle, $hash_or_arrayref) 1060=item json => $cb->($handle, $hash_or_arrayref)
965 1061
966Reads a JSON object or array, decodes it and passes it to the callback. 1062Reads a JSON object or array, decodes it and passes it to the callback.
967 1063
968If a C<json> object was passed to the constructor, then that will be used 1064If a C<json> object was passed to the constructor, then that will be used
… …
978the C<json> write type description, above, for an actual example. 1074the C<json> write type description, above, for an actual example.
979 1075
980=cut 1076=cut
981 1077
982register_read_type json => sub { 1078register_read_type json => sub {
983 my ($self, $cb, $accept, $reject, $skip) = @_; 1079 my ($self, $cb) = @_;
984 1080
985 require JSON; 1081 require JSON;
986 1082
987 my $data; 1083 my $data;
988 my $rbuf = \$self->{rbuf}; 1084 my $rbuf = \$self->{rbuf};
… …
1003 () 1099 ()
1004 } 1100 }
1005 } 1101 }
1006}; 1102};
1007 1103
1104=item storable => $cb->($handle, $ref)
1105
1106Deserialises a L<Storable> frozen representation as written by the
1107C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1108data).
1109
1110Raises C<EBADMSG> error if the data could not be decoded.
1111
1112=cut
1113
1114register_read_type storable => sub {
1115 my ($self, $cb) = @_;
1116
1117 require Storable;
1118
1119 sub {
1120 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1121 defined (my $len = eval { unpack "w", $_[0]->{rbuf} })
1122 or return;
1123
1124 # remove prefix
1125 substr $_[0]->{rbuf}, 0, (length pack "w", $len), "";
1126
1127 # read rest
1128 $_[0]->unshift_read (chunk => $len, sub {
1129 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1130 $cb->($_[0], $ref);
1131 } else {
1132 $self->_error (&Errno::EBADMSG);
1133 }
1134 });
1135 }
1136};
1137
1008=back 1138=back
1009 1139
1010=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1140=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1011 1141
1012This function (not method) lets you add your own types to C<push_read>. 1142This function (not method) lets you add your own types to C<push_read>.
… …
1062 if ($len > 0) { 1192 if ($len > 0) {
1063 $self->{_activity} = AnyEvent->now; 1193 $self->{_activity} = AnyEvent->now;
1064 1194
1065 $self->{filter_r} 1195 $self->{filter_r}
1066 ? $self->{filter_r}($self, $rbuf) 1196 ? $self->{filter_r}($self, $rbuf)
1067 : $self->_drain_rbuf; 1197 : $self->{_in_drain} || $self->_drain_rbuf;
1068 1198
1069 } elsif (defined $len) { 1199 } elsif (defined $len) {
1070 delete $self->{_rw}; 1200 delete $self->{_rw};
1071 $self->{_eof} = 1; 1201 $self->{_eof} = 1;
1072 $self->_drain_rbuf; 1202 $self->_drain_rbuf unless $self->{_in_drain};
1073 1203
1074 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1204 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1075 return $self->_error ($!, 1); 1205 return $self->_error ($!, 1);
1076 } 1206 }
1077 }); 1207 });
… …
1095 } 1225 }
1096 1226
1097 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1227 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1098 if (length $buf) { 1228 if (length $buf) {
1099 $self->{rbuf} .= $buf; 1229 $self->{rbuf} .= $buf;
1100 $self->_drain_rbuf; 1230 $self->_drain_rbuf unless $self->{_in_drain};
1101 } else { 1231 } else {
1102 # let's treat SSL-eof as we treat normal EOF 1232 # let's treat SSL-eof as we treat normal EOF
1103 $self->{_eof} = 1; 1233 $self->{_eof} = 1;
1104 $self->_shutdown; 1234 $self->_shutdown;
1105 return; 1235 return;
… …
1197 1327
1198sub DESTROY { 1328sub DESTROY {
1199 my $self = shift; 1329 my $self = shift;
1200 1330
1201 $self->stoptls; 1331 $self->stoptls;
1332
1333 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1334
1335 if ($linger && length $self->{wbuf}) {
1336 my $fh = delete $self->{fh};
1337 my $wbuf = delete $self->{wbuf};
1338
1339 my @linger;
1340
1341 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1342 my $len = syswrite $fh, $wbuf, length $wbuf;
1343
1344 if ($len > 0) {
1345 substr $wbuf, 0, $len, "";
1346 } else {
1347 @linger = (); # end
1348 }
1349 });
1350 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1351 @linger = ();
1352 });
1353 }
1202} 1354}
1203 1355
1204=item AnyEvent::Handle::TLS_CTX 1356=item AnyEvent::Handle::TLS_CTX
1205 1357
1206This function creates and returns the Net::SSLeay::CTX object used by 1358This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

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