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.48 by root, Thu May 29 00:27:06 2008 UTC vs.
Revision 1.62 by root, Fri Jun 6 10:49:20 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
… …
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
166considered empty. 170considered empty.
167 171
172=item linger => <seconds>
173
174If non-zero (default: C<3600>), then the destructor of the
175AnyEvent::Handle object will check wether there is still outstanding write
176data and will install a watcher that will write out this data. No errors
177will be reported (this mostly matches how the operating system treats
178outstanding data at socket close time).
179
180This will not work for partial TLS data that could not yet been
181encoded. This data will be lost.
182
168=item tls => "accept" | "connect" | Net::SSLeay::SSL object 183=item tls => "accept" | "connect" | Net::SSLeay::SSL object
169 184
170When this parameter is given, it enables TLS (SSL) mode, that means it 185When this parameter is given, it enables TLS (SSL) mode, that means it
171will start making tls handshake and will transparently encrypt/decrypt 186will start making tls handshake and will transparently encrypt/decrypt
172data. 187data.
… …
222 if ($self->{tls}) { 237 if ($self->{tls}) {
223 require Net::SSLeay; 238 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 239 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 } 240 }
226 241
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; 242 $self->{_activity} = AnyEvent->now;
233 $self->_timeout; 243 $self->_timeout;
234 244
235 $self->start_read; 245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
236 246
237 $self 247 $self
238} 248}
239 249
240sub _shutdown { 250sub _shutdown {
… …
242 252
243 delete $self->{_tw}; 253 delete $self->{_tw};
244 delete $self->{_rw}; 254 delete $self->{_rw};
245 delete $self->{_ww}; 255 delete $self->{_ww};
246 delete $self->{fh}; 256 delete $self->{fh};
247}
248 257
258 $self->stoptls;
259}
260
249sub error { 261sub _error {
250 my ($self) = @_; 262 my ($self, $errno, $fatal) = @_;
251 263
252 {
253 local $!;
254 $self->_shutdown; 264 $self->_shutdown
255 } 265 if $fatal;
256 266
257 $self->{on_error}($self) 267 $! = $errno;
268
258 if $self->{on_error}; 269 if ($self->{on_error}) {
259 270 $self->{on_error}($self, $fatal);
271 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 272 Carp::croak "AnyEvent::Handle uncaught error: $!";
273 }
261} 274}
262 275
263=item $fh = $handle->fh 276=item $fh = $handle->fh
264 277
265This method returns the file handle of the L<AnyEvent::Handle> object. 278This method returns the file handle of the L<AnyEvent::Handle> object.
… …
331 $self->{_activity} = $NOW; 344 $self->{_activity} = $NOW;
332 345
333 if ($self->{on_timeout}) { 346 if ($self->{on_timeout}) {
334 $self->{on_timeout}($self); 347 $self->{on_timeout}($self);
335 } else { 348 } else {
336 $! = Errno::ETIMEDOUT; 349 $self->_error (&Errno::ETIMEDOUT);
337 $self->error;
338 } 350 }
339 351
340 # callbakx could have changed timeout value, optimise 352 # callback could have changed timeout value, optimise
341 return unless $self->{timeout}; 353 return unless $self->{timeout};
342 354
343 # calculate new after 355 # calculate new after
344 $after = $self->{timeout}; 356 $after = $self->{timeout};
345 } 357 }
346 358
347 Scalar::Util::weaken $self; 359 Scalar::Util::weaken $self;
360 return unless $self; # ->error could have destroyed $self
348 361
349 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 362 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
350 delete $self->{_tw}; 363 delete $self->{_tw};
351 $self->_timeout; 364 $self->_timeout;
352 }); 365 });
… …
415 if $self->{low_water_mark} >= length $self->{wbuf} 428 if $self->{low_water_mark} >= length $self->{wbuf}
416 && $self->{on_drain}; 429 && $self->{on_drain};
417 430
418 delete $self->{_ww} unless length $self->{wbuf}; 431 delete $self->{_ww} unless length $self->{wbuf};
419 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 432 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
420 $self->error; 433 $self->_error ($!, 1);
421 } 434 }
422 }; 435 };
423 436
424 # try to write data immediately 437 # try to write data immediately
425 $cb->(); 438 $cb->();
… …
454 } 467 }
455} 468}
456 469
457=item $handle->push_write (type => @args) 470=item $handle->push_write (type => @args)
458 471
459=item $handle->unshift_write (type => @args)
460
461Instead of formatting your data yourself, you can also let this module do 472Instead of formatting your data yourself, you can also let this module do
462the job by specifying a type and type-specific arguments. 473the job by specifying a type and type-specific arguments.
463 474
464Predefined types are (if you have ideas for additional types, feel free to 475Predefined types are (if you have ideas for additional types, feel free to
465drop by and tell us): 476drop by and tell us):
… …
469=item netstring => $string 480=item netstring => $string
470 481
471Formats the given value as netstring 482Formats the given value as netstring
472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 483(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
473 484
474=back
475
476=cut 485=cut
477 486
478register_write_type netstring => sub { 487register_write_type netstring => sub {
479 my ($self, $string) = @_; 488 my ($self, $string) = @_;
480 489
481 sprintf "%d:%s,", (length $string), $string 490 sprintf "%d:%s,", (length $string), $string
491};
492
493=item packstring => $format, $data
494
495An octet string prefixed with an encoded length. The encoding C<$format>
496uses the same format as a Perl C<pack> format, but must specify a single
497integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
498optional C<!>, C<< < >> or C<< > >> modifier).
499
500=cut
501
502register_write_type packstring => sub {
503 my ($self, $format, $string) = @_;
504
505 pack "$format/a", $string
482}; 506};
483 507
484=item json => $array_or_hashref 508=item json => $array_or_hashref
485 509
486Encodes the given hash or array reference into a JSON object. Unless you 510Encodes the given hash or array reference into a JSON object. Unless you
… …
520 544
521 $self->{json} ? $self->{json}->encode ($ref) 545 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref) 546 : JSON::encode_json ($ref)
523}; 547};
524 548
549=back
550
525=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 551=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
526 552
527This function (not method) lets you add your own types to C<push_write>. 553This function (not method) lets you add your own types to C<push_write>.
528Whenever the given C<type> is used, C<push_write> will invoke the code 554Whenever the given C<type> is used, C<push_write> will invoke the code
529reference with the handle object and the remaining arguments. 555reference with the handle object and the remaining arguments.
… …
554enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 580enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
555or not. 581or not.
556 582
557In the more complex case, you want to queue multiple callbacks. In this 583In the more complex case, you want to queue multiple callbacks. In this
558case, AnyEvent::Handle will call the first queued callback each time new 584case, AnyEvent::Handle will call the first queued callback each time new
559data arrives and removes it when it has done its job (see C<push_read>, 585data arrives (also the first time it is queued) and removes it when it has
560below). 586done its job (see C<push_read>, below).
561 587
562This way you can, for example, push three line-reads, followed by reading 588This way you can, for example, push three line-reads, followed by reading
563a chunk of data, and AnyEvent::Handle will execute them in order. 589a chunk of data, and AnyEvent::Handle will execute them in order.
564 590
565Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 591Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
566the specified number of bytes which give an XML datagram. 592the specified number of bytes which give an XML datagram.
567 593
568 # in the default state, expect some header bytes 594 # in the default state, expect some header bytes
569 $handle->on_read (sub { 595 $handle->on_read (sub {
570 # some data is here, now queue the length-header-read (4 octets) 596 # some data is here, now queue the length-header-read (4 octets)
571 shift->unshift_read_chunk (4, sub { 597 shift->unshift_read (chunk => 4, sub {
572 # header arrived, decode 598 # header arrived, decode
573 my $len = unpack "N", $_[1]; 599 my $len = unpack "N", $_[1];
574 600
575 # now read the payload 601 # now read the payload
576 shift->unshift_read_chunk ($len, sub { 602 shift->unshift_read (chunk => $len, sub {
577 my $xml = $_[1]; 603 my $xml = $_[1];
578 # handle xml 604 # handle xml
579 }); 605 });
580 }); 606 });
581 }); 607 });
… …
588 614
589 # request one 615 # request one
590 $handle->push_write ("request 1\015\012"); 616 $handle->push_write ("request 1\015\012");
591 617
592 # we expect "ERROR" or "OK" as response, so push a line read 618 # we expect "ERROR" or "OK" as response, so push a line read
593 $handle->push_read_line (sub { 619 $handle->push_read (line => sub {
594 # if we got an "OK", we have to _prepend_ another line, 620 # if we got an "OK", we have to _prepend_ another line,
595 # so it will be read before the second request reads its 64 bytes 621 # so it will be read before the second request reads its 64 bytes
596 # which are already in the queue when this callback is called 622 # which are already in the queue when this callback is called
597 # we don't do this in case we got an error 623 # we don't do this in case we got an error
598 if ($_[1] eq "OK") { 624 if ($_[1] eq "OK") {
599 $_[0]->unshift_read_line (sub { 625 $_[0]->unshift_read (line => sub {
600 my $response = $_[1]; 626 my $response = $_[1];
601 ... 627 ...
602 }); 628 });
603 } 629 }
604 }); 630 });
605 631
606 # request two 632 # request two
607 $handle->push_write ("request 2\015\012"); 633 $handle->push_write ("request 2\015\012");
608 634
609 # simply read 64 bytes, always 635 # simply read 64 bytes, always
610 $handle->push_read_chunk (64, sub { 636 $handle->push_read (chunk => 64, sub {
611 my $response = $_[1]; 637 my $response = $_[1];
612 ... 638 ...
613 }); 639 });
614 640
615=over 4 641=over 4
616 642
617=cut 643=cut
618 644
619sub _drain_rbuf { 645sub _drain_rbuf {
620 my ($self) = @_; 646 my ($self) = @_;
647
648 local $self->{_in_drain} = 1;
621 649
622 if ( 650 if (
623 defined $self->{rbuf_max} 651 defined $self->{rbuf_max}
624 && $self->{rbuf_max} < length $self->{rbuf} 652 && $self->{rbuf_max} < length $self->{rbuf}
625 ) { 653 ) {
626 $! = &Errno::ENOSPC; 654 return $self->_error (&Errno::ENOSPC, 1);
627 $self->error;
628 } 655 }
629 656
630 return if $self->{in_drain}; 657 while () {
631 local $self->{in_drain} = 1;
632
633 while (my $len = length $self->{rbuf}) {
634 no strict 'refs'; 658 no strict 'refs';
659
660 my $len = length $self->{rbuf};
661
635 if (my $cb = shift @{ $self->{_queue} }) { 662 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 663 unless ($cb->($self)) {
637 if ($self->{_eof}) { 664 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming) 665 # no progress can be made (not enough data and no data forthcoming)
639 $! = &Errno::EPIPE; 666 $self->_error (&Errno::EPIPE, 1), last;
640 $self->error;
641 } 667 }
642 668
643 unshift @{ $self->{_queue} }, $cb; 669 unshift @{ $self->{_queue} }, $cb;
644 return; 670 last;
645 } 671 }
646 } elsif ($self->{on_read}) { 672 } elsif ($self->{on_read}) {
673 last unless $len;
674
647 $self->{on_read}($self); 675 $self->{on_read}($self);
648 676
649 if ( 677 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed 678 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 679 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data 680 && $self->{on_read} # but we still have on_read
654 ) { 681 ) {
682 # no further data will arrive
655 # then no progress can be made 683 # so no progress can be made
656 $! = &Errno::EPIPE; 684 $self->_error (&Errno::EPIPE, 1), last
657 $self->error; 685 if $self->{_eof};
686
687 last; # more data might arrive
658 } 688 }
659 } else { 689 } else {
660 # read side becomes idle 690 # read side becomes idle
661 delete $self->{_rw}; 691 delete $self->{_rw};
662 return; 692 last;
663 } 693 }
664 } 694 }
665 695
666 $self->{on_eof}($self) 696 $self->{on_eof}($self)
667 if $self->{_eof} && $self->{on_eof}; 697 if $self->{_eof} && $self->{on_eof};
698
699 # may need to restart read watcher
700 unless ($self->{_rw}) {
701 $self->start_read
702 if $self->{on_read} || @{ $self->{_queue} };
703 }
668} 704}
669 705
670=item $handle->on_read ($cb) 706=item $handle->on_read ($cb)
671 707
672This replaces the currently set C<on_read> callback, or clears it (when 708This replaces the currently set C<on_read> callback, or clears it (when
… …
677 713
678sub on_read { 714sub on_read {
679 my ($self, $cb) = @_; 715 my ($self, $cb) = @_;
680 716
681 $self->{on_read} = $cb; 717 $self->{on_read} = $cb;
718 $self->_drain_rbuf if $cb && !$self->{_in_drain};
682} 719}
683 720
684=item $handle->rbuf 721=item $handle->rbuf
685 722
686Returns the read buffer (as a modifiable lvalue). 723Returns the read buffer (as a modifiable lvalue).
… …
735 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 772 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
736 ->($self, $cb, @_); 773 ->($self, $cb, @_);
737 } 774 }
738 775
739 push @{ $self->{_queue} }, $cb; 776 push @{ $self->{_queue} }, $cb;
740 $self->_drain_rbuf; 777 $self->_drain_rbuf unless $self->{_in_drain};
741} 778}
742 779
743sub unshift_read { 780sub unshift_read {
744 my $self = shift; 781 my $self = shift;
745 my $cb = pop; 782 my $cb = pop;
… …
751 ->($self, $cb, @_); 788 ->($self, $cb, @_);
752 } 789 }
753 790
754 791
755 unshift @{ $self->{_queue} }, $cb; 792 unshift @{ $self->{_queue} }, $cb;
756 $self->_drain_rbuf; 793 $self->_drain_rbuf unless $self->{_in_drain};
757} 794}
758 795
759=item $handle->push_read (type => @args, $cb) 796=item $handle->push_read (type => @args, $cb)
760 797
761=item $handle->unshift_read (type => @args, $cb) 798=item $handle->unshift_read (type => @args, $cb)
… …
846 883
847sub unshift_read_line { 884sub unshift_read_line {
848 my $self = shift; 885 my $self = shift;
849 $self->unshift_read (line => @_); 886 $self->unshift_read (line => @_);
850} 887}
851
852=item netstring => $cb->($handle, $string)
853
854A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
855
856Throws an error with C<$!> set to EBADMSG on format violations.
857
858=cut
859
860register_read_type netstring => sub {
861 my ($self, $cb) = @_;
862
863 sub {
864 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
865 if ($_[0]{rbuf} =~ /[^0-9]/) {
866 $! = &Errno::EBADMSG;
867 $self->error;
868 }
869 return;
870 }
871
872 my $len = $1;
873
874 $self->unshift_read (chunk => $len, sub {
875 my $string = $_[1];
876 $_[0]->unshift_read (chunk => 1, sub {
877 if ($_[1] eq ",") {
878 $cb->($_[0], $string);
879 } else {
880 $! = &Errno::EBADMSG;
881 $self->error;
882 }
883 });
884 });
885
886 1
887 }
888};
889 888
890=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 889=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
891 890
892Makes a regex match against the regex object C<$accept> and returns 891Makes a regex match against the regex object C<$accept> and returns
893everything up to and including the match. 892everything up to and including the match.
… …
943 return 1; 942 return 1;
944 } 943 }
945 944
946 # reject 945 # reject
947 if ($reject && $$rbuf =~ $reject) { 946 if ($reject && $$rbuf =~ $reject) {
948 $! = &Errno::EBADMSG; 947 $self->_error (&Errno::EBADMSG);
949 $self->error;
950 } 948 }
951 949
952 # skip 950 # skip
953 if ($skip && $$rbuf =~ $skip) { 951 if ($skip && $$rbuf =~ $skip) {
954 $data .= substr $$rbuf, 0, $+[0], ""; 952 $data .= substr $$rbuf, 0, $+[0], "";
955 } 953 }
956 954
957 () 955 ()
956 }
957};
958
959=item netstring => $cb->($handle, $string)
960
961A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
962
963Throws an error with C<$!> set to EBADMSG on format violations.
964
965=cut
966
967register_read_type netstring => sub {
968 my ($self, $cb) = @_;
969
970 sub {
971 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
972 if ($_[0]{rbuf} =~ /[^0-9]/) {
973 $self->_error (&Errno::EBADMSG);
974 }
975 return;
976 }
977
978 my $len = $1;
979
980 $self->unshift_read (chunk => $len, sub {
981 my $string = $_[1];
982 $_[0]->unshift_read (chunk => 1, sub {
983 if ($_[1] eq ",") {
984 $cb->($_[0], $string);
985 } else {
986 $self->_error (&Errno::EBADMSG);
987 }
988 });
989 });
990
991 1
992 }
993};
994
995=item packstring => $format, $cb->($handle, $string)
996
997An octet string prefixed with an encoded length. The encoding C<$format>
998uses the same format as a Perl C<pack> format, but must specify a single
999integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1000optional C<!>, C<< < >> or C<< > >> modifier).
1001
1002DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1003
1004Example: read a block of data prefixed by its length in BER-encoded
1005format (very efficient).
1006
1007 $handle->push_read (packstring => "w", sub {
1008 my ($handle, $data) = @_;
1009 });
1010
1011=cut
1012
1013register_read_type packstring => sub {
1014 my ($self, $cb, $format) = @_;
1015
1016 sub {
1017 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1018 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1019 or return;
1020
1021 # remove prefix
1022 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1023
1024 # read rest
1025 $_[0]->unshift_read (chunk => $len, $cb);
1026
1027 1
958 } 1028 }
959}; 1029};
960 1030
961=item json => $cb->($handle, $hash_or_arrayref) 1031=item json => $cb->($handle, $hash_or_arrayref)
962 1032
… …
1027=item $handle->stop_read 1097=item $handle->stop_read
1028 1098
1029=item $handle->start_read 1099=item $handle->start_read
1030 1100
1031In rare cases you actually do not want to read anything from the 1101In rare cases you actually do not want to read anything from the
1032socket. In this case you can call C<stop_read>. Neither C<on_read> no 1102socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1033any queued callbacks will be executed then. To start reading again, call 1103any queued callbacks will be executed then. To start reading again, call
1034C<start_read>. 1104C<start_read>.
1105
1106Note that AnyEvent::Handle will automatically C<start_read> for you when
1107you change the C<on_read> callback or push/unshift a read callback, and it
1108will automatically C<stop_read> for you when neither C<on_read> is set nor
1109there are any read requests in the queue.
1035 1110
1036=cut 1111=cut
1037 1112
1038sub stop_read { 1113sub stop_read {
1039 my ($self) = @_; 1114 my ($self) = @_;
… …
1054 if ($len > 0) { 1129 if ($len > 0) {
1055 $self->{_activity} = AnyEvent->now; 1130 $self->{_activity} = AnyEvent->now;
1056 1131
1057 $self->{filter_r} 1132 $self->{filter_r}
1058 ? $self->{filter_r}($self, $rbuf) 1133 ? $self->{filter_r}($self, $rbuf)
1059 : $self->_drain_rbuf; 1134 : $self->{_in_drain} || $self->_drain_rbuf;
1060 1135
1061 } elsif (defined $len) { 1136 } elsif (defined $len) {
1062 delete $self->{_rw}; 1137 delete $self->{_rw};
1063 $self->{_eof} = 1; 1138 $self->{_eof} = 1;
1064 $self->_drain_rbuf; 1139 $self->_drain_rbuf unless $self->{_in_drain};
1065 1140
1066 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1141 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1067 return $self->error; 1142 return $self->_error ($!, 1);
1068 } 1143 }
1069 }); 1144 });
1070 } 1145 }
1071} 1146}
1072 1147
1073sub _dotls { 1148sub _dotls {
1074 my ($self) = @_; 1149 my ($self) = @_;
1150
1151 my $buf;
1075 1152
1076 if (length $self->{_tls_wbuf}) { 1153 if (length $self->{_tls_wbuf}) {
1077 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1154 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1078 substr $self->{_tls_wbuf}, 0, $len, ""; 1155 substr $self->{_tls_wbuf}, 0, $len, "";
1079 } 1156 }
1080 } 1157 }
1081 1158
1082 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1159 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1083 $self->{wbuf} .= $buf; 1160 $self->{wbuf} .= $buf;
1084 $self->_drain_wbuf; 1161 $self->_drain_wbuf;
1085 } 1162 }
1086 1163
1087 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1164 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1165 if (length $buf) {
1088 $self->{rbuf} .= $buf; 1166 $self->{rbuf} .= $buf;
1089 $self->_drain_rbuf; 1167 $self->_drain_rbuf unless $self->{_in_drain};
1168 } else {
1169 # let's treat SSL-eof as we treat normal EOF
1170 $self->{_eof} = 1;
1171 $self->_shutdown;
1172 return;
1173 }
1090 } 1174 }
1091 1175
1092 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1176 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1093 1177
1094 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1178 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1095 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1179 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1096 $self->error; 1180 return $self->_error ($!, 1);
1097 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1181 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1098 $! = &Errno::EIO; 1182 return $self->_error (&Errno::EIO, 1);
1099 $self->error;
1100 } 1183 }
1101 1184
1102 # all others are fine for our purposes 1185 # all others are fine for our purposes
1103 } 1186 }
1104} 1187}
… …
1119call and can be used or changed to your liking. Note that the handshake 1202call and can be used or changed to your liking. Note that the handshake
1120might have already started when this function returns. 1203might have already started when this function returns.
1121 1204
1122=cut 1205=cut
1123 1206
1124# TODO: maybe document...
1125sub starttls { 1207sub starttls {
1126 my ($self, $ssl, $ctx) = @_; 1208 my ($self, $ssl, $ctx) = @_;
1127 1209
1128 $self->stoptls; 1210 $self->stoptls;
1129 1211
… …
1182 1264
1183sub DESTROY { 1265sub DESTROY {
1184 my $self = shift; 1266 my $self = shift;
1185 1267
1186 $self->stoptls; 1268 $self->stoptls;
1269
1270 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1271
1272 if ($linger && length $self->{wbuf}) {
1273 my $fh = delete $self->{fh};
1274 my $wbuf = delete $self->{wbuf};
1275
1276 my @linger;
1277
1278 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1279 my $len = syswrite $fh, $wbuf, length $wbuf;
1280
1281 if ($len > 0) {
1282 substr $wbuf, 0, $len, "";
1283 } else {
1284 @linger = (); # end
1285 }
1286 });
1287 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1288 @linger = ();
1289 });
1290 }
1187} 1291}
1188 1292
1189=item AnyEvent::Handle::TLS_CTX 1293=item AnyEvent::Handle::TLS_CTX
1190 1294
1191This function creates and returns the Net::SSLeay::CTX object used by 1295This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

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