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.45 by root, Thu May 29 00:20:39 2008 UTC vs.
Revision 1.61 by root, Fri Jun 6 10:23:50 2008 UTC

… …
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = '0.04'; 19our $VERSION = 4.14;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
… …
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 on EOF. 80Set the callback to be called when an end-of-file condition is detcted,
81i.e. in the case of a socket, when the other side has closed the
82connection cleanly.
81 83
82While not mandatory, it is highly recommended to set an eof callback, 84While not mandatory, it is highly recommended to set an eof callback,
83otherwise 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
84waiting for data. 86waiting for data.
85 87
86=item on_error => $cb->($handle) 88=item on_error => $cb->($handle, $fatal)
87 89
88This is the fatal error callback, that is called when, well, a fatal error 90This is the error callback, which is called when, well, some error
89occurs, such as not being able to resolve the hostname, failure to connect 91occured, such as not being able to resolve the hostname, failure to
90or a read error. 92connect or a read error.
91 93
92The object will not be in a usable state when this callback has been 94Some errors are fatal (which is indicated by C<$fatal> being true). On
93called. 95fatal errors the handle object will be shut down and will not be
96usable. Non-fatal errors can be retried by simply returning, but it is
97recommended to simply ignore this parameter and instead abondon the handle
98object when this callback is invoked.
94 99
95On callback entrance, the value of C<$!> contains the operating system 100On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 102
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100
101While not mandatory, it is I<highly> recommended to set this callback, as 103While not mandatory, it is I<highly> recommended to set this callback, as
102you will not be notified of errors otherwise. The default simply calls 104you will not be notified of errors otherwise. The default simply calls
103die. 105C<croak>.
104 106
105=item on_read => $cb->($handle) 107=item on_read => $cb->($handle)
106 108
107This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
108and 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).
109 113
110To 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 >>
111method or access the C<$handle->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
112 116
113When 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
… …
155isn't finished). 159isn't finished).
156 160
157=item read_size => <bytes> 161=item read_size => <bytes>
158 162
159The default read block size (the amount of bytes this module will try to read 163The default read block size (the amount of bytes this module will try to read
160on each [loop iteration). Default: C<4096>. 164during each (loop iteration). Default: C<8192>.
161 165
162=item low_water_mark => <bytes> 166=item low_water_mark => <bytes>
163 167
164Sets the amount of bytes (default: C<0>) that make up an "empty" write 168Sets the amount of bytes (default: C<0>) that make up an "empty" write
165buffer: If the write reaches this size or gets even samller it is 169buffer: If the write reaches this size or gets even samller it is
… …
222 if ($self->{tls}) { 226 if ($self->{tls}) {
223 require Net::SSLeay; 227 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 228 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 } 229 }
226 230
227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
228# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
229# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
231
232 $self->{_activity} = AnyEvent->now; 231 $self->{_activity} = AnyEvent->now;
233 $self->_timeout; 232 $self->_timeout;
234 233
235 $self->start_read; 234 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
236 235
237 $self 236 $self
238} 237}
239 238
240sub _shutdown { 239sub _shutdown {
241 my ($self) = @_; 240 my ($self) = @_;
242 241
242 delete $self->{_tw};
243 delete $self->{_rw}; 243 delete $self->{_rw};
244 delete $self->{_ww}; 244 delete $self->{_ww};
245 delete $self->{fh}; 245 delete $self->{fh};
246}
247 246
247 $self->stoptls;
248}
249
248sub error { 250sub _error {
249 my ($self) = @_; 251 my ($self, $errno, $fatal) = @_;
250 252
251 {
252 local $!;
253 $self->_shutdown; 253 $self->_shutdown
254 } 254 if $fatal;
255 255
256 $self->{on_error}($self) 256 $! = $errno;
257
257 if $self->{on_error}; 258 if ($self->{on_error}) {
258 259 $self->{on_error}($self, $fatal);
260 } else {
259 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 261 Carp::croak "AnyEvent::Handle uncaught error: $!";
262 }
260} 263}
261 264
262=item $fh = $handle->fh 265=item $fh = $handle->fh
263 266
264This method returns the file handle of the L<AnyEvent::Handle> object. 267This method returns the file handle of the L<AnyEvent::Handle> object.
… …
328 # now or in the past already? 331 # now or in the past already?
329 if ($after <= 0) { 332 if ($after <= 0) {
330 $self->{_activity} = $NOW; 333 $self->{_activity} = $NOW;
331 334
332 if ($self->{on_timeout}) { 335 if ($self->{on_timeout}) {
333 $self->{on_timeout}->($self); 336 $self->{on_timeout}($self);
334 } else { 337 } else {
335 $! = Errno::ETIMEDOUT; 338 $self->_error (&Errno::ETIMEDOUT);
336 $self->error;
337 } 339 }
338 340
339 # callbakx could have changed timeout value, optimise 341 # callback could have changed timeout value, optimise
340 return unless $self->{timeout}; 342 return unless $self->{timeout};
341 343
342 # calculate new after 344 # calculate new after
343 $after = $self->{timeout}; 345 $after = $self->{timeout};
344 } 346 }
345 347
346 Scalar::Util::weaken $self; 348 Scalar::Util::weaken $self;
349 return unless $self; # ->error could have destroyed $self
347 350
348 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 351 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
349 delete $self->{_tw}; 352 delete $self->{_tw};
350 $self->_timeout; 353 $self->_timeout;
351 }); 354 });
… …
414 if $self->{low_water_mark} >= length $self->{wbuf} 417 if $self->{low_water_mark} >= length $self->{wbuf}
415 && $self->{on_drain}; 418 && $self->{on_drain};
416 419
417 delete $self->{_ww} unless length $self->{wbuf}; 420 delete $self->{_ww} unless length $self->{wbuf};
418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 421 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
419 $self->error; 422 $self->_error ($!, 1);
420 } 423 }
421 }; 424 };
422 425
423 # try to write data immediately 426 # try to write data immediately
424 $cb->(); 427 $cb->();
… …
444 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 447 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
445 ->($self, @_); 448 ->($self, @_);
446 } 449 }
447 450
448 if ($self->{filter_w}) { 451 if ($self->{filter_w}) {
449 $self->{filter_w}->($self, \$_[0]); 452 $self->{filter_w}($self, \$_[0]);
450 } else { 453 } else {
451 $self->{wbuf} .= $_[0]; 454 $self->{wbuf} .= $_[0];
452 $self->_drain_wbuf; 455 $self->_drain_wbuf;
453 } 456 }
454} 457}
455 458
456=item $handle->push_write (type => @args) 459=item $handle->push_write (type => @args)
457 460
458=item $handle->unshift_write (type => @args)
459
460Instead of formatting your data yourself, you can also let this module do 461Instead of formatting your data yourself, you can also let this module do
461the job by specifying a type and type-specific arguments. 462the job by specifying a type and type-specific arguments.
462 463
463Predefined types are (if you have ideas for additional types, feel free to 464Predefined types are (if you have ideas for additional types, feel free to
464drop by and tell us): 465drop by and tell us):
… …
468=item netstring => $string 469=item netstring => $string
469 470
470Formats the given value as netstring 471Formats the given value as netstring
471(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
472 473
473=back
474
475=cut 474=cut
476 475
477register_write_type netstring => sub { 476register_write_type netstring => sub {
478 my ($self, $string) = @_; 477 my ($self, $string) = @_;
479 478
480 sprintf "%d:%s,", (length $string), $string 479 sprintf "%d:%s,", (length $string), $string
480};
481
482=item packstring => $format, $data
483
484An octet string prefixed with an encoded length. The encoding C<$format>
485uses the same format as a Perl C<pack> format, but must specify a single
486integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
487optional C<!>, C<< < >> or C<< > >> modifier).
488
489=cut
490
491register_write_type packstring => sub {
492 my ($self, $format, $string) = @_;
493
494 pack "$format/a", $string
481}; 495};
482 496
483=item json => $array_or_hashref 497=item json => $array_or_hashref
484 498
485Encodes the given hash or array reference into a JSON object. Unless you 499Encodes the given hash or array reference into a JSON object. Unless you
… …
519 533
520 $self->{json} ? $self->{json}->encode ($ref) 534 $self->{json} ? $self->{json}->encode ($ref)
521 : JSON::encode_json ($ref) 535 : JSON::encode_json ($ref)
522}; 536};
523 537
538=back
539
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 540=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525 541
526This function (not method) lets you add your own types to C<push_write>. 542This function (not method) lets you add your own types to C<push_write>.
527Whenever the given C<type> is used, C<push_write> will invoke the code 543Whenever the given C<type> is used, C<push_write> will invoke the code
528reference with the handle object and the remaining arguments. 544reference with the handle object and the remaining arguments.
… …
553enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 569enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
554or not. 570or not.
555 571
556In the more complex case, you want to queue multiple callbacks. In this 572In the more complex case, you want to queue multiple callbacks. In this
557case, AnyEvent::Handle will call the first queued callback each time new 573case, 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>, 574data arrives (also the first time it is queued) and removes it when it has
559below). 575done its job (see C<push_read>, below).
560 576
561This way you can, for example, push three line-reads, followed by reading 577This way you can, for example, push three line-reads, followed by reading
562a chunk of data, and AnyEvent::Handle will execute them in order. 578a chunk of data, and AnyEvent::Handle will execute them in order.
563 579
564Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 580Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
565the specified number of bytes which give an XML datagram. 581the specified number of bytes which give an XML datagram.
566 582
567 # in the default state, expect some header bytes 583 # in the default state, expect some header bytes
568 $handle->on_read (sub { 584 $handle->on_read (sub {
569 # some data is here, now queue the length-header-read (4 octets) 585 # some data is here, now queue the length-header-read (4 octets)
570 shift->unshift_read_chunk (4, sub { 586 shift->unshift_read (chunk => 4, sub {
571 # header arrived, decode 587 # header arrived, decode
572 my $len = unpack "N", $_[1]; 588 my $len = unpack "N", $_[1];
573 589
574 # now read the payload 590 # now read the payload
575 shift->unshift_read_chunk ($len, sub { 591 shift->unshift_read (chunk => $len, sub {
576 my $xml = $_[1]; 592 my $xml = $_[1];
577 # handle xml 593 # handle xml
578 }); 594 });
579 }); 595 });
580 }); 596 });
… …
587 603
588 # request one 604 # request one
589 $handle->push_write ("request 1\015\012"); 605 $handle->push_write ("request 1\015\012");
590 606
591 # we expect "ERROR" or "OK" as response, so push a line read 607 # we expect "ERROR" or "OK" as response, so push a line read
592 $handle->push_read_line (sub { 608 $handle->push_read (line => sub {
593 # if we got an "OK", we have to _prepend_ another line, 609 # if we got an "OK", we have to _prepend_ another line,
594 # so it will be read before the second request reads its 64 bytes 610 # so it will be read before the second request reads its 64 bytes
595 # which are already in the queue when this callback is called 611 # which are already in the queue when this callback is called
596 # we don't do this in case we got an error 612 # we don't do this in case we got an error
597 if ($_[1] eq "OK") { 613 if ($_[1] eq "OK") {
598 $_[0]->unshift_read_line (sub { 614 $_[0]->unshift_read (line => sub {
599 my $response = $_[1]; 615 my $response = $_[1];
600 ... 616 ...
601 }); 617 });
602 } 618 }
603 }); 619 });
604 620
605 # request two 621 # request two
606 $handle->push_write ("request 2\015\012"); 622 $handle->push_write ("request 2\015\012");
607 623
608 # simply read 64 bytes, always 624 # simply read 64 bytes, always
609 $handle->push_read_chunk (64, sub { 625 $handle->push_read (chunk => 64, sub {
610 my $response = $_[1]; 626 my $response = $_[1];
611 ... 627 ...
612 }); 628 });
613 629
614=over 4 630=over 4
615 631
616=cut 632=cut
617 633
618sub _drain_rbuf { 634sub _drain_rbuf {
619 my ($self) = @_; 635 my ($self) = @_;
636
637 local $self->{_in_drain} = 1;
620 638
621 if ( 639 if (
622 defined $self->{rbuf_max} 640 defined $self->{rbuf_max}
623 && $self->{rbuf_max} < length $self->{rbuf} 641 && $self->{rbuf_max} < length $self->{rbuf}
624 ) { 642 ) {
625 $! = &Errno::ENOSPC; 643 return $self->_error (&Errno::ENOSPC, 1);
626 $self->error;
627 } 644 }
628 645
629 return if $self->{in_drain}; 646 while () {
630 local $self->{in_drain} = 1;
631
632 while (my $len = length $self->{rbuf}) {
633 no strict 'refs'; 647 no strict 'refs';
648
649 my $len = length $self->{rbuf};
650
634 if (my $cb = shift @{ $self->{_queue} }) { 651 if (my $cb = shift @{ $self->{_queue} }) {
635 unless ($cb->($self)) { 652 unless ($cb->($self)) {
636 if ($self->{_eof}) { 653 if ($self->{_eof}) {
637 # no progress can be made (not enough data and no data forthcoming) 654 # no progress can be made (not enough data and no data forthcoming)
638 $! = &Errno::EPIPE; 655 $self->_error (&Errno::EPIPE, 1), last;
639 $self->error;
640 } 656 }
641 657
642 unshift @{ $self->{_queue} }, $cb; 658 unshift @{ $self->{_queue} }, $cb;
643 return; 659 last;
644 } 660 }
645 } elsif ($self->{on_read}) { 661 } elsif ($self->{on_read}) {
662 last unless $len;
663
646 $self->{on_read}($self); 664 $self->{on_read}($self);
647 665
648 if ( 666 if (
649 $self->{_eof} # if no further data will arrive
650 && $len == length $self->{rbuf} # and no data has been consumed 667 $len == length $self->{rbuf} # if no data has been consumed
651 && !@{ $self->{_queue} } # and the queue is still empty 668 && !@{ $self->{_queue} } # and the queue is still empty
652 && $self->{on_read} # and we still want to read data 669 && $self->{on_read} # but we still have on_read
653 ) { 670 ) {
671 # no further data will arrive
654 # then no progress can be made 672 # so no progress can be made
655 $! = &Errno::EPIPE; 673 $self->_error (&Errno::EPIPE, 1), last
656 $self->error; 674 if $self->{_eof};
675
676 last; # more data might arrive
657 } 677 }
658 } else { 678 } else {
659 # read side becomes idle 679 # read side becomes idle
660 delete $self->{_rw}; 680 delete $self->{_rw};
661 return; 681 last;
662 } 682 }
663 } 683 }
664 684
665 if ($self->{_eof}) {
666 $self->_shutdown;
667 $self->{on_eof}($self) 685 $self->{on_eof}($self)
668 if $self->{on_eof}; 686 if $self->{_eof} && $self->{on_eof};
687
688 # may need to restart read watcher
689 unless ($self->{_rw}) {
690 $self->start_read
691 if $self->{on_read} || @{ $self->{_queue} };
669 } 692 }
670} 693}
671 694
672=item $handle->on_read ($cb) 695=item $handle->on_read ($cb)
673 696
… …
679 702
680sub on_read { 703sub on_read {
681 my ($self, $cb) = @_; 704 my ($self, $cb) = @_;
682 705
683 $self->{on_read} = $cb; 706 $self->{on_read} = $cb;
707 $self->_drain_rbuf if $cb && !$self->{_in_drain};
684} 708}
685 709
686=item $handle->rbuf 710=item $handle->rbuf
687 711
688Returns the read buffer (as a modifiable lvalue). 712Returns the read buffer (as a modifiable lvalue).
… …
737 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 761 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
738 ->($self, $cb, @_); 762 ->($self, $cb, @_);
739 } 763 }
740 764
741 push @{ $self->{_queue} }, $cb; 765 push @{ $self->{_queue} }, $cb;
742 $self->_drain_rbuf; 766 $self->_drain_rbuf unless $self->{_in_drain};
743} 767}
744 768
745sub unshift_read { 769sub unshift_read {
746 my $self = shift; 770 my $self = shift;
747 my $cb = pop; 771 my $cb = pop;
… …
753 ->($self, $cb, @_); 777 ->($self, $cb, @_);
754 } 778 }
755 779
756 780
757 unshift @{ $self->{_queue} }, $cb; 781 unshift @{ $self->{_queue} }, $cb;
758 $self->_drain_rbuf; 782 $self->_drain_rbuf unless $self->{_in_drain};
759} 783}
760 784
761=item $handle->push_read (type => @args, $cb) 785=item $handle->push_read (type => @args, $cb)
762 786
763=item $handle->unshift_read (type => @args, $cb) 787=item $handle->unshift_read (type => @args, $cb)
… …
848 872
849sub unshift_read_line { 873sub unshift_read_line {
850 my $self = shift; 874 my $self = shift;
851 $self->unshift_read (line => @_); 875 $self->unshift_read (line => @_);
852} 876}
853
854=item netstring => $cb->($handle, $string)
855
856A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
857
858Throws an error with C<$!> set to EBADMSG on format violations.
859
860=cut
861
862register_read_type netstring => sub {
863 my ($self, $cb) = @_;
864
865 sub {
866 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
867 if ($_[0]{rbuf} =~ /[^0-9]/) {
868 $! = &Errno::EBADMSG;
869 $self->error;
870 }
871 return;
872 }
873
874 my $len = $1;
875
876 $self->unshift_read (chunk => $len, sub {
877 my $string = $_[1];
878 $_[0]->unshift_read (chunk => 1, sub {
879 if ($_[1] eq ",") {
880 $cb->($_[0], $string);
881 } else {
882 $! = &Errno::EBADMSG;
883 $self->error;
884 }
885 });
886 });
887
888 1
889 }
890};
891 877
892=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 878=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
893 879
894Makes a regex match against the regex object C<$accept> and returns 880Makes a regex match against the regex object C<$accept> and returns
895everything up to and including the match. 881everything up to and including the match.
… …
945 return 1; 931 return 1;
946 } 932 }
947 933
948 # reject 934 # reject
949 if ($reject && $$rbuf =~ $reject) { 935 if ($reject && $$rbuf =~ $reject) {
950 $! = &Errno::EBADMSG; 936 $self->_error (&Errno::EBADMSG);
951 $self->error;
952 } 937 }
953 938
954 # skip 939 # skip
955 if ($skip && $$rbuf =~ $skip) { 940 if ($skip && $$rbuf =~ $skip) {
956 $data .= substr $$rbuf, 0, $+[0], ""; 941 $data .= substr $$rbuf, 0, $+[0], "";
957 } 942 }
958 943
959 () 944 ()
945 }
946};
947
948=item netstring => $cb->($handle, $string)
949
950A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
951
952Throws an error with C<$!> set to EBADMSG on format violations.
953
954=cut
955
956register_read_type netstring => sub {
957 my ($self, $cb) = @_;
958
959 sub {
960 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
961 if ($_[0]{rbuf} =~ /[^0-9]/) {
962 $self->_error (&Errno::EBADMSG);
963 }
964 return;
965 }
966
967 my $len = $1;
968
969 $self->unshift_read (chunk => $len, sub {
970 my $string = $_[1];
971 $_[0]->unshift_read (chunk => 1, sub {
972 if ($_[1] eq ",") {
973 $cb->($_[0], $string);
974 } else {
975 $self->_error (&Errno::EBADMSG);
976 }
977 });
978 });
979
980 1
981 }
982};
983
984=item packstring => $format, $cb->($handle, $string)
985
986An octet string prefixed with an encoded length. The encoding C<$format>
987uses the same format as a Perl C<pack> format, but must specify a single
988integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
989optional C<!>, C<< < >> or C<< > >> modifier).
990
991DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
992
993Example: read a block of data prefixed by its length in BER-encoded
994format (very efficient).
995
996 $handle->push_read (packstring => "w", sub {
997 my ($handle, $data) = @_;
998 });
999
1000=cut
1001
1002register_read_type packstring => sub {
1003 my ($self, $cb, $format) = @_;
1004
1005 sub {
1006 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1007 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1008 or return;
1009
1010 # remove prefix
1011 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1012
1013 # read rest
1014 $_[0]->unshift_read (chunk => $len, $cb);
1015
1016 1
960 } 1017 }
961}; 1018};
962 1019
963=item json => $cb->($handle, $hash_or_arrayref) 1020=item json => $cb->($handle, $hash_or_arrayref)
964 1021
… …
1029=item $handle->stop_read 1086=item $handle->stop_read
1030 1087
1031=item $handle->start_read 1088=item $handle->start_read
1032 1089
1033In rare cases you actually do not want to read anything from the 1090In rare cases you actually do not want to read anything from the
1034socket. In this case you can call C<stop_read>. Neither C<on_read> no 1091socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1035any queued callbacks will be executed then. To start reading again, call 1092any queued callbacks will be executed then. To start reading again, call
1036C<start_read>. 1093C<start_read>.
1094
1095Note that AnyEvent::Handle will automatically C<start_read> for you when
1096you change the C<on_read> callback or push/unshift a read callback, and it
1097will automatically C<stop_read> for you when neither C<on_read> is set nor
1098there are any read requests in the queue.
1037 1099
1038=cut 1100=cut
1039 1101
1040sub stop_read { 1102sub stop_read {
1041 my ($self) = @_; 1103 my ($self) = @_;
… …
1055 1117
1056 if ($len > 0) { 1118 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now; 1119 $self->{_activity} = AnyEvent->now;
1058 1120
1059 $self->{filter_r} 1121 $self->{filter_r}
1060 ? $self->{filter_r}->($self, $rbuf) 1122 ? $self->{filter_r}($self, $rbuf)
1061 : $self->_drain_rbuf; 1123 : $self->{_in_drain} || $self->_drain_rbuf;
1062 1124
1063 } elsif (defined $len) { 1125 } elsif (defined $len) {
1064 delete $self->{_rw}; 1126 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
1067 $self->{_eof} = 1; 1127 $self->{_eof} = 1;
1068 $self->_drain_rbuf; 1128 $self->_drain_rbuf unless $self->{_in_drain};
1069 1129
1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1130 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1071 return $self->error; 1131 return $self->_error ($!, 1);
1072 } 1132 }
1073 }); 1133 });
1074 } 1134 }
1075} 1135}
1076 1136
1077sub _dotls { 1137sub _dotls {
1078 my ($self) = @_; 1138 my ($self) = @_;
1139
1140 my $buf;
1079 1141
1080 if (length $self->{_tls_wbuf}) { 1142 if (length $self->{_tls_wbuf}) {
1081 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1143 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1082 substr $self->{_tls_wbuf}, 0, $len, ""; 1144 substr $self->{_tls_wbuf}, 0, $len, "";
1083 } 1145 }
1084 } 1146 }
1085 1147
1086 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1148 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1087 $self->{wbuf} .= $buf; 1149 $self->{wbuf} .= $buf;
1088 $self->_drain_wbuf; 1150 $self->_drain_wbuf;
1089 } 1151 }
1090 1152
1091 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1153 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1154 if (length $buf) {
1092 $self->{rbuf} .= $buf; 1155 $self->{rbuf} .= $buf;
1093 $self->_drain_rbuf; 1156 $self->_drain_rbuf unless $self->{_in_drain};
1157 } else {
1158 # let's treat SSL-eof as we treat normal EOF
1159 $self->{_eof} = 1;
1160 $self->_shutdown;
1161 return;
1162 }
1094 } 1163 }
1095 1164
1096 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1165 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1097 1166
1098 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1167 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1099 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1168 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1100 $self->error; 1169 return $self->_error ($!, 1);
1101 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1170 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1102 $! = &Errno::EIO; 1171 return $self->_error (&Errno::EIO, 1);
1103 $self->error;
1104 } 1172 }
1105 1173
1106 # all others are fine for our purposes 1174 # all others are fine for our purposes
1107 } 1175 }
1108} 1176}
… …
1123call and can be used or changed to your liking. Note that the handshake 1191call and can be used or changed to your liking. Note that the handshake
1124might have already started when this function returns. 1192might have already started when this function returns.
1125 1193
1126=cut 1194=cut
1127 1195
1128# TODO: maybe document...
1129sub starttls { 1196sub starttls {
1130 my ($self, $ssl, $ctx) = @_; 1197 my ($self, $ssl, $ctx) = @_;
1131 1198
1132 $self->stoptls; 1199 $self->stoptls;
1133 1200

Diff Legend

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