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.59 by root, Thu Jun 5 16:53:11 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];
634 690
635 if (my $cb = shift @{ $self->{_queue} }) { 691 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 692 unless ($cb->($self)) {
637 if ($self->{_eof}) { 693 if ($self->{_eof}) {
638 # 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)
639 return $self->_error (&Errno::EPIPE, 1); 695 $self->_error (&Errno::EPIPE, 1), last;
640 } 696 }
641 697
642 unshift @{ $self->{_queue} }, $cb; 698 unshift @{ $self->{_queue} }, $cb;
643 last; 699 last;
644 } 700 }
645 } elsif ($self->{on_read}) { 701 } elsif ($self->{on_read}) {
702 last unless $len;
703
646 $self->{on_read}($self); 704 $self->{on_read}($self);
647 705
648 if ( 706 if (
649 $len == length $self->{rbuf} # if no data has been consumed 707 $len == length $self->{rbuf} # if no data has been consumed
650 && !@{ $self->{_queue} } # and the queue is still empty 708 && !@{ $self->{_queue} } # and the queue is still empty
651 && $self->{on_read} # but we still have on_read 709 && $self->{on_read} # but we still have on_read
652 ) { 710 ) {
653 # no further data will arrive 711 # no further data will arrive
654 # so no progress can be made 712 # so no progress can be made
655 return $self->_error (&Errno::EPIPE, 1) 713 $self->_error (&Errno::EPIPE, 1), last
656 if $self->{_eof}; 714 if $self->{_eof};
657 715
658 last; # more data might arrive 716 last; # more data might arrive
659 } 717 }
660 } else { 718 } else {
854 912
855sub unshift_read_line { 913sub unshift_read_line {
856 my $self = shift; 914 my $self = shift;
857 $self->unshift_read (line => @_); 915 $self->unshift_read (line => @_);
858} 916}
859
860=item netstring => $cb->($handle, $string)
861
862A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
863
864Throws an error with C<$!> set to EBADMSG on format violations.
865
866=cut
867
868register_read_type netstring => sub {
869 my ($self, $cb) = @_;
870
871 sub {
872 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
873 if ($_[0]{rbuf} =~ /[^0-9]/) {
874 $self->_error (&Errno::EBADMSG);
875 }
876 return;
877 }
878
879 my $len = $1;
880
881 $self->unshift_read (chunk => $len, sub {
882 my $string = $_[1];
883 $_[0]->unshift_read (chunk => 1, sub {
884 if ($_[1] eq ",") {
885 $cb->($_[0], $string);
886 } else {
887 $self->_error (&Errno::EBADMSG);
888 }
889 });
890 });
891
892 1
893 }
894};
895 917
896=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 918=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
897 919
898Makes a regex match against the regex object C<$accept> and returns 920Makes a regex match against the regex object C<$accept> and returns
899everything up to and including the match. 921everything up to and including the match.
961 983
962 () 984 ()
963 } 985 }
964}; 986};
965 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
966=item json => $cb->($handle, $hash_or_arrayref) 1060=item json => $cb->($handle, $hash_or_arrayref)
967 1061
968Reads 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.
969 1063
970If 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
980the C<json> write type description, above, for an actual example. 1074the C<json> write type description, above, for an actual example.
981 1075
982=cut 1076=cut
983 1077
984register_read_type json => sub { 1078register_read_type json => sub {
985 my ($self, $cb, $accept, $reject, $skip) = @_; 1079 my ($self, $cb) = @_;
986 1080
987 require JSON; 1081 require JSON;
988 1082
989 my $data; 1083 my $data;
990 my $rbuf = \$self->{rbuf}; 1084 my $rbuf = \$self->{rbuf};
1002 1 1096 1
1003 } else { 1097 } else {
1004 $self->{rbuf} = ""; 1098 $self->{rbuf} = "";
1005 () 1099 ()
1006 } 1100 }
1101 }
1102};
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 });
1007 } 1135 }
1008}; 1136};
1009 1137
1010=back 1138=back
1011 1139
1199 1327
1200sub DESTROY { 1328sub DESTROY {
1201 my $self = shift; 1329 my $self = shift;
1202 1330
1203 $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 }
1204} 1354}
1205 1355
1206=item AnyEvent::Handle::TLS_CTX 1356=item AnyEvent::Handle::TLS_CTX
1207 1357
1208This 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