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.57 by root, Wed Jun 4 11:45:21 2008 UTC vs.
Revision 1.79 by root, Sun Jul 27 08:37:56 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 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
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
154be configured to accept only so-and-so much data that it cannot act on 162be 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 163(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 164amount of data without a callback ever being called as long as the line
157isn't finished). 165isn't finished).
158 166
167=item autocork => <boolean>
168
169When disabled (the default), then C<push_write> will try to immediately
170write the data to the handle if possible. This avoids having to register
171a write watcher and wait for the next event loop iteration, but can be
172inefficient if you write multiple small chunks (this disadvantage is
173usually avoided by your kernel's nagle algorithm, see C<low_delay>).
174
175When enabled, then writes will always be queued till the next event loop
176iteration. This is efficient when you do many small writes per iteration,
177but less efficient when you do a single write only.
178
179=item no_delay => <boolean>
180
181When doing small writes on sockets, your operating system kernel might
182wait a bit for more data before actually sending it out. This is called
183the Nagle algorithm, and usually it is beneficial.
184
185In some situations you want as low a delay as possible, which cna be
186accomplishd by setting this option to true.
187
188The default is your opertaing system's default behaviour, this option
189explicitly enables or disables it, if possible.
190
159=item read_size => <bytes> 191=item read_size => <bytes>
160 192
161The default read block size (the amount of bytes this module will try to read 193The default read block size (the amount of bytes this module will try to read
162during each (loop iteration). Default: C<8192>. 194during each (loop iteration). Default: C<8192>.
163 195
164=item low_water_mark => <bytes> 196=item low_water_mark => <bytes>
165 197
166Sets the amount of bytes (default: C<0>) that make up an "empty" write 198Sets 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 199buffer: If the write reaches this size or gets even samller it is
168considered empty. 200considered empty.
201
202=item linger => <seconds>
203
204If non-zero (default: C<3600>), then the destructor of the
205AnyEvent::Handle object will check wether there is still outstanding write
206data and will install a watcher that will write out this data. No errors
207will be reported (this mostly matches how the operating system treats
208outstanding data at socket close time).
209
210This will not work for partial TLS data that could not yet been
211encoded. This data will be lost.
169 212
170=item tls => "accept" | "connect" | Net::SSLeay::SSL object 213=item tls => "accept" | "connect" | Net::SSLeay::SSL object
171 214
172When this parameter is given, it enables TLS (SSL) mode, that means it 215When this parameter is given, it enables TLS (SSL) mode, that means it
173will start making tls handshake and will transparently encrypt/decrypt 216will start making tls handshake and will transparently encrypt/decrypt
182You can also provide your own TLS connection object, but you have 225You can also provide your own TLS connection object, but you have
183to make sure that you call either C<Net::SSLeay::set_connect_state> 226to 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 227or C<Net::SSLeay::set_accept_state> on it before you pass it to
185AnyEvent::Handle. 228AnyEvent::Handle.
186 229
187See the C<starttls> method if you need to start TLs negotiation later. 230See the C<starttls> method if you need to start TLS negotiation later.
188 231
189=item tls_ctx => $ssl_ctx 232=item tls_ctx => $ssl_ctx
190 233
191Use the given Net::SSLeay::CTX object to create the new TLS connection 234Use the given Net::SSLeay::CTX object to create the new TLS connection
192(unless a connection object was specified directly). If this parameter is 235(unless a connection object was specified directly). If this parameter is
224 if ($self->{tls}) { 267 if ($self->{tls}) {
225 require Net::SSLeay; 268 require Net::SSLeay;
226 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 269 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
227 } 270 }
228 271
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; 272 $self->{_activity} = AnyEvent->now;
235 $self->_timeout; 273 $self->_timeout;
236 274
275 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
276 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
277
237 $self->start_read; 278 $self->start_read
279 if $self->{on_read};
238 280
239 $self 281 $self
240} 282}
241 283
242sub _shutdown { 284sub _shutdown {
301 343
302=cut 344=cut
303 345
304sub on_timeout { 346sub on_timeout {
305 $_[0]{on_timeout} = $_[1]; 347 $_[0]{on_timeout} = $_[1];
348}
349
350=item $handle->autocork ($boolean)
351
352Enables or disables the current autocork behaviour (see C<autocork>
353constructor argument).
354
355=cut
356
357=item $handle->no_delay ($boolean)
358
359Enables or disables the C<no_delay> setting (see constructor argument of
360the same name for details).
361
362=cut
363
364sub no_delay {
365 $_[0]{no_delay} = $_[1];
366
367 eval {
368 local $SIG{__DIE__};
369 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
370 };
306} 371}
307 372
308############################################################################# 373#############################################################################
309 374
310=item $handle->timeout ($seconds) 375=item $handle->timeout ($seconds)
425 $self->_error ($!, 1); 490 $self->_error ($!, 1);
426 } 491 }
427 }; 492 };
428 493
429 # try to write data immediately 494 # try to write data immediately
430 $cb->(); 495 $cb->() unless $self->{autocork};
431 496
432 # if still data left in wbuf, we need to poll 497 # if still data left in wbuf, we need to poll
433 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 498 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
434 if length $self->{wbuf}; 499 if length $self->{wbuf};
435 }; 500 };
480 my ($self, $string) = @_; 545 my ($self, $string) = @_;
481 546
482 sprintf "%d:%s,", (length $string), $string 547 sprintf "%d:%s,", (length $string), $string
483}; 548};
484 549
550=item packstring => $format, $data
551
552An octet string prefixed with an encoded length. The encoding C<$format>
553uses the same format as a Perl C<pack> format, but must specify a single
554integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
555optional C<!>, C<< < >> or C<< > >> modifier).
556
557=cut
558
559register_write_type packstring => sub {
560 my ($self, $format, $string) = @_;
561
562 pack "$format/a*", $string
563};
564
485=item json => $array_or_hashref 565=item json => $array_or_hashref
486 566
487Encodes the given hash or array reference into a JSON object. Unless you 567Encodes the given hash or array reference into a JSON object. Unless you
488provide your own JSON object, this means it will be encoded to JSON text 568provide your own JSON object, this means it will be encoded to JSON text
489in UTF-8. 569in UTF-8.
521 601
522 $self->{json} ? $self->{json}->encode ($ref) 602 $self->{json} ? $self->{json}->encode ($ref)
523 : JSON::encode_json ($ref) 603 : JSON::encode_json ($ref)
524}; 604};
525 605
606=item storable => $reference
607
608Freezes the given reference using L<Storable> and writes it to the
609handle. Uses the C<nfreeze> format.
610
611=cut
612
613register_write_type storable => sub {
614 my ($self, $ref) = @_;
615
616 require Storable;
617
618 pack "w/a*", Storable::nfreeze ($ref)
619};
620
526=back 621=back
527 622
528=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 623=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
529 624
530This function (not method) lets you add your own types to C<push_write>. 625This function (not method) lets you add your own types to C<push_write>.
552ways, the "simple" way, using only C<on_read> and the "complex" way, using 647ways, the "simple" way, using only C<on_read> and the "complex" way, using
553a queue. 648a queue.
554 649
555In the simple case, you just install an C<on_read> callback and whenever 650In the simple case, you just install an C<on_read> callback and whenever
556new data arrives, it will be called. You can then remove some data (if 651new data arrives, it will be called. You can then remove some data (if
557enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 652enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
558or not. 653leave the data there if you want to accumulate more (e.g. when only a
654partial message has been received so far).
559 655
560In the more complex case, you want to queue multiple callbacks. In this 656In the more complex case, you want to queue multiple callbacks. In this
561case, AnyEvent::Handle will call the first queued callback each time new 657case, AnyEvent::Handle will call the first queued callback each time new
562data arrives and removes it when it has done its job (see C<push_read>, 658data arrives (also the first time it is queued) and removes it when it has
563below). 659done its job (see C<push_read>, below).
564 660
565This way you can, for example, push three line-reads, followed by reading 661This way you can, for example, push three line-reads, followed by reading
566a chunk of data, and AnyEvent::Handle will execute them in order. 662a chunk of data, and AnyEvent::Handle will execute them in order.
567 663
568Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 664Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
581 # handle xml 677 # handle xml
582 }); 678 });
583 }); 679 });
584 }); 680 });
585 681
586Example 2: Implement a client for a protocol that replies either with 682Example 2: Implement a client for a protocol that replies either with "OK"
587"OK" and another line or "ERROR" for one request, and 64 bytes for the 683and another line or "ERROR" for the first request that is sent, and 64
588second request. Due tot he availability of a full queue, we can just 684bytes for the second request. Due to the availability of a queue, we can
589pipeline sending both requests and manipulate the queue as necessary in 685just pipeline sending both requests and manipulate the queue as necessary
590the callbacks: 686in the callbacks.
591 687
592 # request one 688When the first callback is called and sees an "OK" response, it will
689C<unshift> another line-read. This line-read will be queued I<before> the
69064-byte chunk callback.
691
692 # request one, returns either "OK + extra line" or "ERROR"
593 $handle->push_write ("request 1\015\012"); 693 $handle->push_write ("request 1\015\012");
594 694
595 # we expect "ERROR" or "OK" as response, so push a line read 695 # we expect "ERROR" or "OK" as response, so push a line read
596 $handle->push_read (line => sub { 696 $handle->push_read (line => sub {
597 # if we got an "OK", we have to _prepend_ another line, 697 # if we got an "OK", we have to _prepend_ another line,
604 ... 704 ...
605 }); 705 });
606 } 706 }
607 }); 707 });
608 708
609 # request two 709 # request two, simply returns 64 octets
610 $handle->push_write ("request 2\015\012"); 710 $handle->push_write ("request 2\015\012");
611 711
612 # simply read 64 bytes, always 712 # simply read 64 bytes, always
613 $handle->push_read (chunk => 64, sub { 713 $handle->push_read (chunk => 64, sub {
614 my $response = $_[1]; 714 my $response = $_[1];
620=cut 720=cut
621 721
622sub _drain_rbuf { 722sub _drain_rbuf {
623 my ($self) = @_; 723 my ($self) = @_;
624 724
725 local $self->{_in_drain} = 1;
726
625 if ( 727 if (
626 defined $self->{rbuf_max} 728 defined $self->{rbuf_max}
627 && $self->{rbuf_max} < length $self->{rbuf} 729 && $self->{rbuf_max} < length $self->{rbuf}
628 ) { 730 ) {
629 return $self->_error (&Errno::ENOSPC, 1); 731 return $self->_error (&Errno::ENOSPC, 1);
630 } 732 }
631 733
632 return if $self->{in_drain}; 734 while () {
633 local $self->{in_drain} = 1;
634
635 while (my $len = length $self->{rbuf}) { 735 my $len = length $self->{rbuf};
636 no strict 'refs'; 736
637 if (my $cb = shift @{ $self->{_queue} }) { 737 if (my $cb = shift @{ $self->{_queue} }) {
638 unless ($cb->($self)) { 738 unless ($cb->($self)) {
639 if ($self->{_eof}) { 739 if ($self->{_eof}) {
640 # no progress can be made (not enough data and no data forthcoming) 740 # no progress can be made (not enough data and no data forthcoming)
641 return $self->_error (&Errno::EPIPE, 1); 741 $self->_error (&Errno::EPIPE, 1), last;
642 } 742 }
643 743
644 unshift @{ $self->{_queue} }, $cb; 744 unshift @{ $self->{_queue} }, $cb;
645 last; 745 last;
646 } 746 }
647 } elsif ($self->{on_read}) { 747 } elsif ($self->{on_read}) {
748 last unless $len;
749
648 $self->{on_read}($self); 750 $self->{on_read}($self);
649 751
650 if ( 752 if (
651 $len == length $self->{rbuf} # if no data has been consumed 753 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 754 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # but we still have on_read 755 && $self->{on_read} # but we still have on_read
654 ) { 756 ) {
655 # no further data will arrive 757 # no further data will arrive
656 # so no progress can be made 758 # so no progress can be made
657 return $self->_error (&Errno::EPIPE, 1) 759 $self->_error (&Errno::EPIPE, 1), last
658 if $self->{_eof}; 760 if $self->{_eof};
659 761
660 last; # more data might arrive 762 last; # more data might arrive
661 } 763 }
662 } else { 764 } else {
686 788
687sub on_read { 789sub on_read {
688 my ($self, $cb) = @_; 790 my ($self, $cb) = @_;
689 791
690 $self->{on_read} = $cb; 792 $self->{on_read} = $cb;
793 $self->_drain_rbuf if $cb && !$self->{_in_drain};
691} 794}
692 795
693=item $handle->rbuf 796=item $handle->rbuf
694 797
695Returns the read buffer (as a modifiable lvalue). 798Returns the read buffer (as a modifiable lvalue).
744 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 847 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
745 ->($self, $cb, @_); 848 ->($self, $cb, @_);
746 } 849 }
747 850
748 push @{ $self->{_queue} }, $cb; 851 push @{ $self->{_queue} }, $cb;
749 $self->_drain_rbuf; 852 $self->_drain_rbuf unless $self->{_in_drain};
750} 853}
751 854
752sub unshift_read { 855sub unshift_read {
753 my $self = shift; 856 my $self = shift;
754 my $cb = pop; 857 my $cb = pop;
760 ->($self, $cb, @_); 863 ->($self, $cb, @_);
761 } 864 }
762 865
763 866
764 unshift @{ $self->{_queue} }, $cb; 867 unshift @{ $self->{_queue} }, $cb;
765 $self->_drain_rbuf; 868 $self->_drain_rbuf unless $self->{_in_drain};
766} 869}
767 870
768=item $handle->push_read (type => @args, $cb) 871=item $handle->push_read (type => @args, $cb)
769 872
770=item $handle->unshift_read (type => @args, $cb) 873=item $handle->unshift_read (type => @args, $cb)
800 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 903 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
801 1 904 1
802 } 905 }
803}; 906};
804 907
805# compatibility with older API
806sub push_read_chunk {
807 $_[0]->push_read (chunk => $_[1], $_[2]);
808}
809
810sub unshift_read_chunk {
811 $_[0]->unshift_read (chunk => $_[1], $_[2]);
812}
813
814=item line => [$eol, ]$cb->($handle, $line, $eol) 908=item line => [$eol, ]$cb->($handle, $line, $eol)
815 909
816The callback will be called only once a full line (including the end of 910The callback will be called only once a full line (including the end of
817line marker, C<$eol>) has been read. This line (excluding the end of line 911line marker, C<$eol>) has been read. This line (excluding the end of line
818marker) will be passed to the callback as second argument (C<$line>), and 912marker) will be passed to the callback as second argument (C<$line>), and
833=cut 927=cut
834 928
835register_read_type line => sub { 929register_read_type line => sub {
836 my ($self, $cb, $eol) = @_; 930 my ($self, $cb, $eol) = @_;
837 931
838 $eol = qr|(\015?\012)| if @_ < 3; 932 if (@_ < 3) {
839 $eol = quotemeta $eol unless ref $eol; 933 # this is more than twice as fast as the generic code below
840 $eol = qr|^(.*?)($eol)|s;
841
842 sub { 934 sub {
843 $_[0]{rbuf} =~ s/$eol// or return; 935 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
844 936
845 $cb->($_[0], $1, $2); 937 $cb->($_[0], $1, $2);
846 1
847 }
848};
849
850# compatibility with older API
851sub push_read_line {
852 my $self = shift;
853 $self->push_read (line => @_);
854}
855
856sub unshift_read_line {
857 my $self = shift;
858 $self->unshift_read (line => @_);
859}
860
861=item netstring => $cb->($handle, $string)
862
863A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
864
865Throws an error with C<$!> set to EBADMSG on format violations.
866
867=cut
868
869register_read_type netstring => sub {
870 my ($self, $cb) = @_;
871
872 sub {
873 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
874 if ($_[0]{rbuf} =~ /[^0-9]/) {
875 $self->_error (&Errno::EBADMSG);
876 } 938 1
877 return;
878 } 939 }
940 } else {
941 $eol = quotemeta $eol unless ref $eol;
942 $eol = qr|^(.*?)($eol)|s;
879 943
880 my $len = $1; 944 sub {
945 $_[0]{rbuf} =~ s/$eol// or return;
881 946
882 $self->unshift_read (chunk => $len, sub { 947 $cb->($_[0], $1, $2);
883 my $string = $_[1];
884 $_[0]->unshift_read (chunk => 1, sub {
885 if ($_[1] eq ",") {
886 $cb->($_[0], $string);
887 } else {
888 $self->_error (&Errno::EBADMSG);
889 }
890 }); 948 1
891 }); 949 }
892
893 1
894 } 950 }
895}; 951};
896 952
897=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 953=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
898 954
962 1018
963 () 1019 ()
964 } 1020 }
965}; 1021};
966 1022
1023=item netstring => $cb->($handle, $string)
1024
1025A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1026
1027Throws an error with C<$!> set to EBADMSG on format violations.
1028
1029=cut
1030
1031register_read_type netstring => sub {
1032 my ($self, $cb) = @_;
1033
1034 sub {
1035 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1036 if ($_[0]{rbuf} =~ /[^0-9]/) {
1037 $self->_error (&Errno::EBADMSG);
1038 }
1039 return;
1040 }
1041
1042 my $len = $1;
1043
1044 $self->unshift_read (chunk => $len, sub {
1045 my $string = $_[1];
1046 $_[0]->unshift_read (chunk => 1, sub {
1047 if ($_[1] eq ",") {
1048 $cb->($_[0], $string);
1049 } else {
1050 $self->_error (&Errno::EBADMSG);
1051 }
1052 });
1053 });
1054
1055 1
1056 }
1057};
1058
1059=item packstring => $format, $cb->($handle, $string)
1060
1061An octet string prefixed with an encoded length. The encoding C<$format>
1062uses the same format as a Perl C<pack> format, but must specify a single
1063integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1064optional C<!>, C<< < >> or C<< > >> modifier).
1065
1066DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1067
1068Example: read a block of data prefixed by its length in BER-encoded
1069format (very efficient).
1070
1071 $handle->push_read (packstring => "w", sub {
1072 my ($handle, $data) = @_;
1073 });
1074
1075=cut
1076
1077register_read_type packstring => sub {
1078 my ($self, $cb, $format) = @_;
1079
1080 sub {
1081 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1082 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1083 or return;
1084
1085 $format = length pack $format, $len;
1086
1087 # bypass unshift if we already have the remaining chunk
1088 if ($format + $len <= length $_[0]{rbuf}) {
1089 my $data = substr $_[0]{rbuf}, $format, $len;
1090 substr $_[0]{rbuf}, 0, $format + $len, "";
1091 $cb->($_[0], $data);
1092 } else {
1093 # remove prefix
1094 substr $_[0]{rbuf}, 0, $format, "";
1095
1096 # read remaining chunk
1097 $_[0]->unshift_read (chunk => $len, $cb);
1098 }
1099
1100 1
1101 }
1102};
1103
967=item json => $cb->($handle, $hash_or_arrayref) 1104=item json => $cb->($handle, $hash_or_arrayref)
968 1105
969Reads a JSON object or array, decodes it and passes it to the callback. 1106Reads a JSON object or array, decodes it and passes it to the callback.
970 1107
971If a C<json> object was passed to the constructor, then that will be used 1108If a C<json> object was passed to the constructor, then that will be used
981the C<json> write type description, above, for an actual example. 1118the C<json> write type description, above, for an actual example.
982 1119
983=cut 1120=cut
984 1121
985register_read_type json => sub { 1122register_read_type json => sub {
986 my ($self, $cb, $accept, $reject, $skip) = @_; 1123 my ($self, $cb) = @_;
987 1124
988 require JSON; 1125 require JSON;
989 1126
990 my $data; 1127 my $data;
991 my $rbuf = \$self->{rbuf}; 1128 my $rbuf = \$self->{rbuf};
1006 () 1143 ()
1007 } 1144 }
1008 } 1145 }
1009}; 1146};
1010 1147
1148=item storable => $cb->($handle, $ref)
1149
1150Deserialises a L<Storable> frozen representation as written by the
1151C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1152data).
1153
1154Raises C<EBADMSG> error if the data could not be decoded.
1155
1156=cut
1157
1158register_read_type storable => sub {
1159 my ($self, $cb) = @_;
1160
1161 require Storable;
1162
1163 sub {
1164 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1165 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1166 or return;
1167
1168 my $format = length pack "w", $len;
1169
1170 # bypass unshift if we already have the remaining chunk
1171 if ($format + $len <= length $_[0]{rbuf}) {
1172 my $data = substr $_[0]{rbuf}, $format, $len;
1173 substr $_[0]{rbuf}, 0, $format + $len, "";
1174 $cb->($_[0], Storable::thaw ($data));
1175 } else {
1176 # remove prefix
1177 substr $_[0]{rbuf}, 0, $format, "";
1178
1179 # read remaining chunk
1180 $_[0]->unshift_read (chunk => $len, sub {
1181 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1182 $cb->($_[0], $ref);
1183 } else {
1184 $self->_error (&Errno::EBADMSG);
1185 }
1186 });
1187 }
1188
1189 1
1190 }
1191};
1192
1011=back 1193=back
1012 1194
1013=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1195=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1014 1196
1015This function (not method) lets you add your own types to C<push_read>. 1197This function (not method) lets you add your own types to C<push_read>.
1033=item $handle->stop_read 1215=item $handle->stop_read
1034 1216
1035=item $handle->start_read 1217=item $handle->start_read
1036 1218
1037In rare cases you actually do not want to read anything from the 1219In rare cases you actually do not want to read anything from the
1038socket. In this case you can call C<stop_read>. Neither C<on_read> no 1220socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1039any queued callbacks will be executed then. To start reading again, call 1221any queued callbacks will be executed then. To start reading again, call
1040C<start_read>. 1222C<start_read>.
1041 1223
1042Note that AnyEvent::Handle will automatically C<start_read> for you when 1224Note that AnyEvent::Handle will automatically C<start_read> for you when
1043you change the C<on_read> callback or push/unshift a read callback, and it 1225you change the C<on_read> callback or push/unshift a read callback, and it
1065 if ($len > 0) { 1247 if ($len > 0) {
1066 $self->{_activity} = AnyEvent->now; 1248 $self->{_activity} = AnyEvent->now;
1067 1249
1068 $self->{filter_r} 1250 $self->{filter_r}
1069 ? $self->{filter_r}($self, $rbuf) 1251 ? $self->{filter_r}($self, $rbuf)
1070 : $self->_drain_rbuf; 1252 : $self->{_in_drain} || $self->_drain_rbuf;
1071 1253
1072 } elsif (defined $len) { 1254 } elsif (defined $len) {
1073 delete $self->{_rw}; 1255 delete $self->{_rw};
1074 $self->{_eof} = 1; 1256 $self->{_eof} = 1;
1075 $self->_drain_rbuf; 1257 $self->_drain_rbuf unless $self->{_in_drain};
1076 1258
1077 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1259 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1078 return $self->_error ($!, 1); 1260 return $self->_error ($!, 1);
1079 } 1261 }
1080 }); 1262 });
1098 } 1280 }
1099 1281
1100 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1282 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1101 if (length $buf) { 1283 if (length $buf) {
1102 $self->{rbuf} .= $buf; 1284 $self->{rbuf} .= $buf;
1103 $self->_drain_rbuf; 1285 $self->_drain_rbuf unless $self->{_in_drain};
1104 } else { 1286 } else {
1105 # let's treat SSL-eof as we treat normal EOF 1287 # let's treat SSL-eof as we treat normal EOF
1106 $self->{_eof} = 1; 1288 $self->{_eof} = 1;
1107 $self->_shutdown; 1289 $self->_shutdown;
1108 return; 1290 return;
1200 1382
1201sub DESTROY { 1383sub DESTROY {
1202 my $self = shift; 1384 my $self = shift;
1203 1385
1204 $self->stoptls; 1386 $self->stoptls;
1387
1388 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1389
1390 if ($linger && length $self->{wbuf}) {
1391 my $fh = delete $self->{fh};
1392 my $wbuf = delete $self->{wbuf};
1393
1394 my @linger;
1395
1396 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1397 my $len = syswrite $fh, $wbuf, length $wbuf;
1398
1399 if ($len > 0) {
1400 substr $wbuf, 0, $len, "";
1401 } else {
1402 @linger = (); # end
1403 }
1404 });
1405 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1406 @linger = ();
1407 });
1408 }
1205} 1409}
1206 1410
1207=item AnyEvent::Handle::TLS_CTX 1411=item AnyEvent::Handle::TLS_CTX
1208 1412
1209This function creates and returns the Net::SSLeay::CTX object used by 1413This function creates and returns the Net::SSLeay::CTX object used by
1251=over 4 1455=over 4
1252 1456
1253=item * all constructor arguments become object members. 1457=item * all constructor arguments become object members.
1254 1458
1255At least initially, when you pass a C<tls>-argument to the constructor it 1459At least initially, when you pass a C<tls>-argument to the constructor it
1256will end up in C<< $handle->{tls} >>. Those members might be changes or 1460will end up in C<< $handle->{tls} >>. Those members might be changed or
1257mutated later on (for example C<tls> will hold the TLS connection object). 1461mutated later on (for example C<tls> will hold the TLS connection object).
1258 1462
1259=item * other object member names are prefixed with an C<_>. 1463=item * other object member names are prefixed with an C<_>.
1260 1464
1261All object members not explicitly documented (internal use) are prefixed 1465All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines