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.80 by root, Sun Jul 27 08:43:32 2008 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict qw(subs vars);
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
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.22;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
75NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
76AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
77 77
78=item on_eof => $cb->($handle) 78=item on_eof => $cb->($handle)
79 79
80Set the callback to be called when an end-of-file condition is detcted, 80Set the callback to be called when an end-of-file condition is detected,
81i.e. in the case of a socket, when the other side has closed the 81i.e. in the case of a socket, when the other side has closed the
82connection cleanly. 82connection cleanly.
83 83
84While not mandatory, it is highly recommended to set an eof callback, 84While not mandatory, it is I<highly> recommended to set an eof callback,
85otherwise you might end up with a closed socket while you are still 85otherwise you might end up with a closed socket while you are still
86waiting for data. 86waiting for data.
87
88If an EOF condition has been detected but no C<on_eof> callback has been
89set, then a fatal error will be raised with C<$!> set to <0>.
87 90
88=item on_error => $cb->($handle, $fatal) 91=item on_error => $cb->($handle, $fatal)
89 92
90This is the error callback, which is called when, well, some error 93This is the error callback, which is called when, well, some error
91occured, such as not being able to resolve the hostname, failure to 94occured, such as not being able to resolve the hostname, failure to
105C<croak>. 108C<croak>.
106 109
107=item on_read => $cb->($handle) 110=item on_read => $cb->($handle)
108 111
109This sets the default read callback, which is called when data arrives 112This sets the default read callback, which is called when data arrives
110and no read request is in the queue. 113and no read request is in the queue (unlike read queue callbacks, this
114callback will only be called when at least one octet of data is in the
115read buffer).
111 116
112To access (and remove data from) the read buffer, use the C<< ->rbuf >> 117To access (and remove data from) the read buffer, use the C<< ->rbuf >>
113method or access the C<$handle->{rbuf}> member directly. 118method or access the C<$handle->{rbuf}> member directly.
114 119
115When an EOF condition is detected then AnyEvent::Handle will first try to 120When an EOF condition is detected then AnyEvent::Handle will first try to
121 126
122This sets the callback that is called when the write buffer becomes empty 127This 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). 128(or when the callback is set and the buffer is empty already).
124 129
125To append to the write buffer, use the C<< ->push_write >> method. 130To append to the write buffer, use the C<< ->push_write >> method.
131
132This callback is useful when you don't want to put all of your write data
133into the queue at once, for example, when you want to write the contents
134of some file to the socket you might not want to read the whole file into
135memory and push it into the queue, but instead only read more data from
136the file when the write queue becomes empty.
126 137
127=item timeout => $fractional_seconds 138=item timeout => $fractional_seconds
128 139
129If non-zero, then this enables an "inactivity" timeout: whenever this many 140If non-zero, then this enables an "inactivity" timeout: whenever this many
130seconds pass without a successful read or write on the underlying file 141seconds pass without a successful read or write on the underlying file
154be configured to accept only so-and-so much data that it cannot act on 165be configured to accept only so-and-so much data that it cannot act on
155(for example, when expecting a line, an attacker could send an unlimited 166(for example, when expecting a line, an attacker could send an unlimited
156amount of data without a callback ever being called as long as the line 167amount of data without a callback ever being called as long as the line
157isn't finished). 168isn't finished).
158 169
170=item autocork => <boolean>
171
172When disabled (the default), then C<push_write> will try to immediately
173write the data to the handle if possible. This avoids having to register
174a write watcher and wait for the next event loop iteration, but can be
175inefficient if you write multiple small chunks (this disadvantage is
176usually avoided by your kernel's nagle algorithm, see C<low_delay>).
177
178When enabled, then writes will always be queued till the next event loop
179iteration. This is efficient when you do many small writes per iteration,
180but less efficient when you do a single write only.
181
182=item no_delay => <boolean>
183
184When doing small writes on sockets, your operating system kernel might
185wait a bit for more data before actually sending it out. This is called
186the Nagle algorithm, and usually it is beneficial.
187
188In some situations you want as low a delay as possible, which cna be
189accomplishd by setting this option to true.
190
191The default is your opertaing system's default behaviour, this option
192explicitly enables or disables it, if possible.
193
159=item read_size => <bytes> 194=item read_size => <bytes>
160 195
161The default read block size (the amount of bytes this module will try to read 196The default read block size (the amount of bytes this module will try to read
162during each (loop iteration). Default: C<8192>. 197during each (loop iteration). Default: C<8192>.
163 198
164=item low_water_mark => <bytes> 199=item low_water_mark => <bytes>
165 200
166Sets the amount of bytes (default: C<0>) that make up an "empty" write 201Sets 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 202buffer: If the write reaches this size or gets even samller it is
168considered empty. 203considered empty.
204
205=item linger => <seconds>
206
207If non-zero (default: C<3600>), then the destructor of the
208AnyEvent::Handle object will check wether there is still outstanding write
209data and will install a watcher that will write out this data. No errors
210will be reported (this mostly matches how the operating system treats
211outstanding data at socket close time).
212
213This will not work for partial TLS data that could not yet been
214encoded. This data will be lost.
169 215
170=item tls => "accept" | "connect" | Net::SSLeay::SSL object 216=item tls => "accept" | "connect" | Net::SSLeay::SSL object
171 217
172When this parameter is given, it enables TLS (SSL) mode, that means it 218When this parameter is given, it enables TLS (SSL) mode, that means it
173will start making tls handshake and will transparently encrypt/decrypt 219will start making tls handshake and will transparently encrypt/decrypt
182You can also provide your own TLS connection object, but you have 228You can also provide your own TLS connection object, but you have
183to make sure that you call either C<Net::SSLeay::set_connect_state> 229to make sure that you call either C<Net::SSLeay::set_connect_state>
184or C<Net::SSLeay::set_accept_state> on it before you pass it to 230or C<Net::SSLeay::set_accept_state> on it before you pass it to
185AnyEvent::Handle. 231AnyEvent::Handle.
186 232
187See the C<starttls> method if you need to start TLs negotiation later. 233See the C<starttls> method if you need to start TLS negotiation later.
188 234
189=item tls_ctx => $ssl_ctx 235=item tls_ctx => $ssl_ctx
190 236
191Use the given Net::SSLeay::CTX object to create the new TLS connection 237Use the given Net::SSLeay::CTX object to create the new TLS connection
192(unless a connection object was specified directly). If this parameter is 238(unless a connection object was specified directly). If this parameter is
227 } 273 }
228 274
229 $self->{_activity} = AnyEvent->now; 275 $self->{_activity} = AnyEvent->now;
230 $self->_timeout; 276 $self->_timeout;
231 277
232 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 278 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
233 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 279 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
280
281 $self->start_read
282 if $self->{on_read};
234 283
235 $self 284 $self
236} 285}
237 286
238sub _shutdown { 287sub _shutdown {
297 346
298=cut 347=cut
299 348
300sub on_timeout { 349sub on_timeout {
301 $_[0]{on_timeout} = $_[1]; 350 $_[0]{on_timeout} = $_[1];
351}
352
353=item $handle->autocork ($boolean)
354
355Enables or disables the current autocork behaviour (see C<autocork>
356constructor argument).
357
358=cut
359
360=item $handle->no_delay ($boolean)
361
362Enables or disables the C<no_delay> setting (see constructor argument of
363the same name for details).
364
365=cut
366
367sub no_delay {
368 $_[0]{no_delay} = $_[1];
369
370 eval {
371 local $SIG{__DIE__};
372 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
373 };
302} 374}
303 375
304############################################################################# 376#############################################################################
305 377
306=item $handle->timeout ($seconds) 378=item $handle->timeout ($seconds)
421 $self->_error ($!, 1); 493 $self->_error ($!, 1);
422 } 494 }
423 }; 495 };
424 496
425 # try to write data immediately 497 # try to write data immediately
426 $cb->(); 498 $cb->() unless $self->{autocork};
427 499
428 # if still data left in wbuf, we need to poll 500 # if still data left in wbuf, we need to poll
429 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 501 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
430 if length $self->{wbuf}; 502 if length $self->{wbuf};
431 }; 503 };
476 my ($self, $string) = @_; 548 my ($self, $string) = @_;
477 549
478 sprintf "%d:%s,", (length $string), $string 550 sprintf "%d:%s,", (length $string), $string
479}; 551};
480 552
553=item packstring => $format, $data
554
555An octet string prefixed with an encoded length. The encoding C<$format>
556uses the same format as a Perl C<pack> format, but must specify a single
557integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
558optional C<!>, C<< < >> or C<< > >> modifier).
559
560=cut
561
562register_write_type packstring => sub {
563 my ($self, $format, $string) = @_;
564
565 pack "$format/a*", $string
566};
567
481=item json => $array_or_hashref 568=item json => $array_or_hashref
482 569
483Encodes the given hash or array reference into a JSON object. Unless you 570Encodes 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 571provide your own JSON object, this means it will be encoded to JSON text
485in UTF-8. 572in UTF-8.
517 604
518 $self->{json} ? $self->{json}->encode ($ref) 605 $self->{json} ? $self->{json}->encode ($ref)
519 : JSON::encode_json ($ref) 606 : JSON::encode_json ($ref)
520}; 607};
521 608
609=item storable => $reference
610
611Freezes the given reference using L<Storable> and writes it to the
612handle. Uses the C<nfreeze> format.
613
614=cut
615
616register_write_type storable => sub {
617 my ($self, $ref) = @_;
618
619 require Storable;
620
621 pack "w/a*", Storable::nfreeze ($ref)
622};
623
522=back 624=back
523 625
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 626=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525 627
526This function (not method) lets you add your own types to C<push_write>. 628This 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 650ways, the "simple" way, using only C<on_read> and the "complex" way, using
549a queue. 651a queue.
550 652
551In the simple case, you just install an C<on_read> callback and whenever 653In 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 654new 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 655enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
554or not. 656leave the data there if you want to accumulate more (e.g. when only a
657partial message has been received so far).
555 658
556In the more complex case, you want to queue multiple callbacks. In this 659In the more complex case, you want to queue multiple callbacks. In this
557case, AnyEvent::Handle will call the first queued callback each time new 660case, 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>, 661data arrives (also the first time it is queued) and removes it when it has
559below). 662done its job (see C<push_read>, below).
560 663
561This way you can, for example, push three line-reads, followed by reading 664This way you can, for example, push three line-reads, followed by reading
562a chunk of data, and AnyEvent::Handle will execute them in order. 665a chunk of data, and AnyEvent::Handle will execute them in order.
563 666
564Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 667Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
577 # handle xml 680 # handle xml
578 }); 681 });
579 }); 682 });
580 }); 683 });
581 684
582Example 2: Implement a client for a protocol that replies either with 685Example 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 686and 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 687bytes for the second request. Due to the availability of a queue, we can
585pipeline sending both requests and manipulate the queue as necessary in 688just pipeline sending both requests and manipulate the queue as necessary
586the callbacks: 689in the callbacks.
587 690
588 # request one 691When the first callback is called and sees an "OK" response, it will
692C<unshift> another line-read. This line-read will be queued I<before> the
69364-byte chunk callback.
694
695 # request one, returns either "OK + extra line" or "ERROR"
589 $handle->push_write ("request 1\015\012"); 696 $handle->push_write ("request 1\015\012");
590 697
591 # we expect "ERROR" or "OK" as response, so push a line read 698 # we expect "ERROR" or "OK" as response, so push a line read
592 $handle->push_read (line => sub { 699 $handle->push_read (line => sub {
593 # if we got an "OK", we have to _prepend_ another line, 700 # if we got an "OK", we have to _prepend_ another line,
600 ... 707 ...
601 }); 708 });
602 } 709 }
603 }); 710 });
604 711
605 # request two 712 # request two, simply returns 64 octets
606 $handle->push_write ("request 2\015\012"); 713 $handle->push_write ("request 2\015\012");
607 714
608 # simply read 64 bytes, always 715 # simply read 64 bytes, always
609 $handle->push_read (chunk => 64, sub { 716 $handle->push_read (chunk => 64, sub {
610 my $response = $_[1]; 717 my $response = $_[1];
626 ) { 733 ) {
627 return $self->_error (&Errno::ENOSPC, 1); 734 return $self->_error (&Errno::ENOSPC, 1);
628 } 735 }
629 736
630 while () { 737 while () {
631 no strict 'refs';
632
633 my $len = length $self->{rbuf}; 738 my $len = length $self->{rbuf};
634 739
635 if (my $cb = shift @{ $self->{_queue} }) { 740 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 741 unless ($cb->($self)) {
637 if ($self->{_eof}) { 742 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming) 743 # no progress can be made (not enough data and no data forthcoming)
639 return $self->_error (&Errno::EPIPE, 1); 744 $self->_error (&Errno::EPIPE, 1), last;
640 } 745 }
641 746
642 unshift @{ $self->{_queue} }, $cb; 747 unshift @{ $self->{_queue} }, $cb;
643 last; 748 last;
644 } 749 }
645 } elsif ($self->{on_read}) { 750 } elsif ($self->{on_read}) {
751 last unless $len;
752
646 $self->{on_read}($self); 753 $self->{on_read}($self);
647 754
648 if ( 755 if (
649 $len == length $self->{rbuf} # if no data has been consumed 756 $len == length $self->{rbuf} # if no data has been consumed
650 && !@{ $self->{_queue} } # and the queue is still empty 757 && !@{ $self->{_queue} } # and the queue is still empty
651 && $self->{on_read} # but we still have on_read 758 && $self->{on_read} # but we still have on_read
652 ) { 759 ) {
653 # no further data will arrive 760 # no further data will arrive
654 # so no progress can be made 761 # so no progress can be made
655 return $self->_error (&Errno::EPIPE, 1) 762 $self->_error (&Errno::EPIPE, 1), last
656 if $self->{_eof}; 763 if $self->{_eof};
657 764
658 last; # more data might arrive 765 last; # more data might arrive
659 } 766 }
660 } else { 767 } else {
662 delete $self->{_rw}; 769 delete $self->{_rw};
663 last; 770 last;
664 } 771 }
665 } 772 }
666 773
774 if ($self->{_eof}) {
775 if ($self->{on_eof}) {
667 $self->{on_eof}($self) 776 $self->{on_eof}($self)
668 if $self->{_eof} && $self->{on_eof}; 777 } else {
778 $self->_error (0, 1);
779 }
780 }
669 781
670 # may need to restart read watcher 782 # may need to restart read watcher
671 unless ($self->{_rw}) { 783 unless ($self->{_rw}) {
672 $self->start_read 784 $self->start_read
673 if $self->{on_read} || @{ $self->{_queue} }; 785 if $self->{on_read} || @{ $self->{_queue} };
799 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 911 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
800 1 912 1
801 } 913 }
802}; 914};
803 915
804# compatibility with older API
805sub push_read_chunk {
806 $_[0]->push_read (chunk => $_[1], $_[2]);
807}
808
809sub unshift_read_chunk {
810 $_[0]->unshift_read (chunk => $_[1], $_[2]);
811}
812
813=item line => [$eol, ]$cb->($handle, $line, $eol) 916=item line => [$eol, ]$cb->($handle, $line, $eol)
814 917
815The callback will be called only once a full line (including the end of 918The callback will be called only once a full line (including the end of
816line marker, C<$eol>) has been read. This line (excluding the end of line 919line marker, C<$eol>) has been read. This line (excluding the end of line
817marker) will be passed to the callback as second argument (C<$line>), and 920marker) will be passed to the callback as second argument (C<$line>), and
832=cut 935=cut
833 936
834register_read_type line => sub { 937register_read_type line => sub {
835 my ($self, $cb, $eol) = @_; 938 my ($self, $cb, $eol) = @_;
836 939
837 $eol = qr|(\015?\012)| if @_ < 3; 940 if (@_ < 3) {
941 # this is more than twice as fast as the generic code below
942 sub {
943 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
944
945 $cb->($_[0], $1, $2);
946 1
947 }
948 } else {
838 $eol = quotemeta $eol unless ref $eol; 949 $eol = quotemeta $eol unless ref $eol;
839 $eol = qr|^(.*?)($eol)|s; 950 $eol = qr|^(.*?)($eol)|s;
951
952 sub {
953 $_[0]{rbuf} =~ s/$eol// or return;
954
955 $cb->($_[0], $1, $2);
956 1
957 }
958 }
959};
960
961=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
962
963Makes a regex match against the regex object C<$accept> and returns
964everything up to and including the match.
965
966Example: read a single line terminated by '\n'.
967
968 $handle->push_read (regex => qr<\n>, sub { ... });
969
970If C<$reject> is given and not undef, then it determines when the data is
971to be rejected: it is matched against the data when the C<$accept> regex
972does not match and generates an C<EBADMSG> error when it matches. This is
973useful to quickly reject wrong data (to avoid waiting for a timeout or a
974receive buffer overflow).
975
976Example: expect a single decimal number followed by whitespace, reject
977anything else (not the use of an anchor).
978
979 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
980
981If C<$skip> is given and not C<undef>, then it will be matched against
982the receive buffer when neither C<$accept> nor C<$reject> match,
983and everything preceding and including the match will be accepted
984unconditionally. This is useful to skip large amounts of data that you
985know cannot be matched, so that the C<$accept> or C<$reject> regex do not
986have to start matching from the beginning. This is purely an optimisation
987and is usually worth only when you expect more than a few kilobytes.
988
989Example: expect a http header, which ends at C<\015\012\015\012>. Since we
990expect the header to be very large (it isn't in practise, but...), we use
991a skip regex to skip initial portions. The skip regex is tricky in that
992it only accepts something not ending in either \015 or \012, as these are
993required for the accept regex.
994
995 $handle->push_read (regex =>
996 qr<\015\012\015\012>,
997 undef, # no reject
998 qr<^.*[^\015\012]>,
999 sub { ... });
1000
1001=cut
1002
1003register_read_type regex => sub {
1004 my ($self, $cb, $accept, $reject, $skip) = @_;
1005
1006 my $data;
1007 my $rbuf = \$self->{rbuf};
840 1008
841 sub { 1009 sub {
842 $_[0]{rbuf} =~ s/$eol// or return; 1010 # accept
843 1011 if ($$rbuf =~ $accept) {
844 $cb->($_[0], $1, $2); 1012 $data .= substr $$rbuf, 0, $+[0], "";
1013 $cb->($self, $data);
1014 return 1;
1015 }
845 1 1016
1017 # reject
1018 if ($reject && $$rbuf =~ $reject) {
1019 $self->_error (&Errno::EBADMSG);
1020 }
1021
1022 # skip
1023 if ($skip && $$rbuf =~ $skip) {
1024 $data .= substr $$rbuf, 0, $+[0], "";
1025 }
1026
1027 ()
846 } 1028 }
847}; 1029};
848
849# compatibility with older API
850sub push_read_line {
851 my $self = shift;
852 $self->push_read (line => @_);
853}
854
855sub unshift_read_line {
856 my $self = shift;
857 $self->unshift_read (line => @_);
858}
859 1030
860=item netstring => $cb->($handle, $string) 1031=item netstring => $cb->($handle, $string)
861 1032
862A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement). 1033A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
863 1034
891 1062
892 1 1063 1
893 } 1064 }
894}; 1065};
895 1066
896=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 1067=item packstring => $format, $cb->($handle, $string)
897 1068
898Makes a regex match against the regex object C<$accept> and returns 1069An octet string prefixed with an encoded length. The encoding C<$format>
899everything up to and including the match. 1070uses the same format as a Perl C<pack> format, but must specify a single
1071integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1072optional C<!>, C<< < >> or C<< > >> modifier).
900 1073
901Example: read a single line terminated by '\n'. 1074DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
902 1075
903 $handle->push_read (regex => qr<\n>, sub { ... }); 1076Example: read a block of data prefixed by its length in BER-encoded
1077format (very efficient).
904 1078
905If C<$reject> is given and not undef, then it determines when the data is 1079 $handle->push_read (packstring => "w", sub {
906to be rejected: it is matched against the data when the C<$accept> regex 1080 my ($handle, $data) = @_;
907does not match and generates an C<EBADMSG> error when it matches. This is 1081 });
908useful to quickly reject wrong data (to avoid waiting for a timeout or a
909receive buffer overflow).
910 1082
911Example: expect a single decimal number followed by whitespace, reject
912anything else (not the use of an anchor).
913
914 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
915
916If C<$skip> is given and not C<undef>, then it will be matched against
917the receive buffer when neither C<$accept> nor C<$reject> match,
918and everything preceding and including the match will be accepted
919unconditionally. This is useful to skip large amounts of data that you
920know cannot be matched, so that the C<$accept> or C<$reject> regex do not
921have to start matching from the beginning. This is purely an optimisation
922and is usually worth only when you expect more than a few kilobytes.
923
924Example: expect a http header, which ends at C<\015\012\015\012>. Since we
925expect the header to be very large (it isn't in practise, but...), we use
926a skip regex to skip initial portions. The skip regex is tricky in that
927it only accepts something not ending in either \015 or \012, as these are
928required for the accept regex.
929
930 $handle->push_read (regex =>
931 qr<\015\012\015\012>,
932 undef, # no reject
933 qr<^.*[^\015\012]>,
934 sub { ... });
935
936=cut 1083=cut
937 1084
938register_read_type regex => sub { 1085register_read_type packstring => sub {
939 my ($self, $cb, $accept, $reject, $skip) = @_; 1086 my ($self, $cb, $format) = @_;
940
941 my $data;
942 my $rbuf = \$self->{rbuf};
943 1087
944 sub { 1088 sub {
945 # accept 1089 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
946 if ($$rbuf =~ $accept) { 1090 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
947 $data .= substr $$rbuf, 0, $+[0], "";
948 $cb->($self, $data);
949 return 1; 1091 or return;
950 } 1092
1093 $format = length pack $format, $len;
1094
1095 # bypass unshift if we already have the remaining chunk
1096 if ($format + $len <= length $_[0]{rbuf}) {
1097 my $data = substr $_[0]{rbuf}, $format, $len;
1098 substr $_[0]{rbuf}, 0, $format + $len, "";
1099 $cb->($_[0], $data);
1100 } else {
1101 # remove prefix
1102 substr $_[0]{rbuf}, 0, $format, "";
1103
1104 # read remaining chunk
1105 $_[0]->unshift_read (chunk => $len, $cb);
951 1106 }
952 # reject 1107
953 if ($reject && $$rbuf =~ $reject) {
954 $self->_error (&Errno::EBADMSG);
955 } 1108 1
956
957 # skip
958 if ($skip && $$rbuf =~ $skip) {
959 $data .= substr $$rbuf, 0, $+[0], "";
960 }
961
962 ()
963 } 1109 }
964}; 1110};
965 1111
966=item json => $cb->($handle, $hash_or_arrayref) 1112=item json => $cb->($handle, $hash_or_arrayref)
967 1113
980the C<json> write type description, above, for an actual example. 1126the C<json> write type description, above, for an actual example.
981 1127
982=cut 1128=cut
983 1129
984register_read_type json => sub { 1130register_read_type json => sub {
985 my ($self, $cb, $accept, $reject, $skip) = @_; 1131 my ($self, $cb) = @_;
986 1132
987 require JSON; 1133 require JSON;
988 1134
989 my $data; 1135 my $data;
990 my $rbuf = \$self->{rbuf}; 1136 my $rbuf = \$self->{rbuf};
1002 1 1148 1
1003 } else { 1149 } else {
1004 $self->{rbuf} = ""; 1150 $self->{rbuf} = "";
1005 () 1151 ()
1006 } 1152 }
1153 }
1154};
1155
1156=item storable => $cb->($handle, $ref)
1157
1158Deserialises a L<Storable> frozen representation as written by the
1159C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1160data).
1161
1162Raises C<EBADMSG> error if the data could not be decoded.
1163
1164=cut
1165
1166register_read_type storable => sub {
1167 my ($self, $cb) = @_;
1168
1169 require Storable;
1170
1171 sub {
1172 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1173 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1174 or return;
1175
1176 my $format = length pack "w", $len;
1177
1178 # bypass unshift if we already have the remaining chunk
1179 if ($format + $len <= length $_[0]{rbuf}) {
1180 my $data = substr $_[0]{rbuf}, $format, $len;
1181 substr $_[0]{rbuf}, 0, $format + $len, "";
1182 $cb->($_[0], Storable::thaw ($data));
1183 } else {
1184 # remove prefix
1185 substr $_[0]{rbuf}, 0, $format, "";
1186
1187 # read remaining chunk
1188 $_[0]->unshift_read (chunk => $len, sub {
1189 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1190 $cb->($_[0], $ref);
1191 } else {
1192 $self->_error (&Errno::EBADMSG);
1193 }
1194 });
1195 }
1196
1197 1
1007 } 1198 }
1008}; 1199};
1009 1200
1010=back 1201=back
1011 1202
1199 1390
1200sub DESTROY { 1391sub DESTROY {
1201 my $self = shift; 1392 my $self = shift;
1202 1393
1203 $self->stoptls; 1394 $self->stoptls;
1395
1396 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1397
1398 if ($linger && length $self->{wbuf}) {
1399 my $fh = delete $self->{fh};
1400 my $wbuf = delete $self->{wbuf};
1401
1402 my @linger;
1403
1404 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1405 my $len = syswrite $fh, $wbuf, length $wbuf;
1406
1407 if ($len > 0) {
1408 substr $wbuf, 0, $len, "";
1409 } else {
1410 @linger = (); # end
1411 }
1412 });
1413 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1414 @linger = ();
1415 });
1416 }
1204} 1417}
1205 1418
1206=item AnyEvent::Handle::TLS_CTX 1419=item AnyEvent::Handle::TLS_CTX
1207 1420
1208This function creates and returns the Net::SSLeay::CTX object used by 1421This function creates and returns the Net::SSLeay::CTX object used by
1250=over 4 1463=over 4
1251 1464
1252=item * all constructor arguments become object members. 1465=item * all constructor arguments become object members.
1253 1466
1254At least initially, when you pass a C<tls>-argument to the constructor it 1467At least initially, when you pass a C<tls>-argument to the constructor it
1255will end up in C<< $handle->{tls} >>. Those members might be changes or 1468will end up in C<< $handle->{tls} >>. Those members might be changed or
1256mutated later on (for example C<tls> will hold the TLS connection object). 1469mutated later on (for example C<tls> will hold the TLS connection object).
1257 1470
1258=item * other object member names are prefixed with an C<_>. 1471=item * other object member names are prefixed with an C<_>.
1259 1472
1260All object members not explicitly documented (internal use) are prefixed 1473All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines