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.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
… …
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
166considered empty. 170considered empty.
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.
167 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
… …
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 {
241 my ($self) = @_; 251 my ($self) = @_;
242 252
253 delete $self->{_tw};
243 delete $self->{_rw}; 254 delete $self->{_rw};
244 delete $self->{_ww}; 255 delete $self->{_ww};
245 delete $self->{fh}; 256 delete $self->{fh};
246}
247 257
258 $self->stoptls;
259}
260
248sub error { 261sub _error {
249 my ($self) = @_; 262 my ($self, $errno, $fatal) = @_;
250 263
251 {
252 local $!;
253 $self->_shutdown; 264 $self->_shutdown
254 } 265 if $fatal;
255 266
256 $self->{on_error}($self) 267 $! = $errno;
268
257 if $self->{on_error}; 269 if ($self->{on_error}) {
258 270 $self->{on_error}($self, $fatal);
271 } else {
259 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 272 Carp::croak "AnyEvent::Handle uncaught error: $!";
273 }
260} 274}
261 275
262=item $fh = $handle->fh 276=item $fh = $handle->fh
263 277
264This method returns the file handle of the L<AnyEvent::Handle> object. 278This method returns the file handle of the L<AnyEvent::Handle> object.
… …
328 # now or in the past already? 342 # now or in the past already?
329 if ($after <= 0) { 343 if ($after <= 0) {
330 $self->{_activity} = $NOW; 344 $self->{_activity} = $NOW;
331 345
332 if ($self->{on_timeout}) { 346 if ($self->{on_timeout}) {
333 $self->{on_timeout}->($self); 347 $self->{on_timeout}($self);
334 } else { 348 } else {
335 $! = Errno::ETIMEDOUT; 349 $self->_error (&Errno::ETIMEDOUT);
336 $self->error;
337 } 350 }
338 351
339 # callbakx could have changed timeout value, optimise 352 # callback could have changed timeout value, optimise
340 return unless $self->{timeout}; 353 return unless $self->{timeout};
341 354
342 # calculate new after 355 # calculate new after
343 $after = $self->{timeout}; 356 $after = $self->{timeout};
344 } 357 }
345 358
346 Scalar::Util::weaken $self; 359 Scalar::Util::weaken $self;
360 return unless $self; # ->error could have destroyed $self
347 361
348 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 362 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
349 delete $self->{_tw}; 363 delete $self->{_tw};
350 $self->_timeout; 364 $self->_timeout;
351 }); 365 });
… …
414 if $self->{low_water_mark} >= length $self->{wbuf} 428 if $self->{low_water_mark} >= length $self->{wbuf}
415 && $self->{on_drain}; 429 && $self->{on_drain};
416 430
417 delete $self->{_ww} unless length $self->{wbuf}; 431 delete $self->{_ww} unless length $self->{wbuf};
418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 432 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
419 $self->error; 433 $self->_error ($!, 1);
420 } 434 }
421 }; 435 };
422 436
423 # try to write data immediately 437 # try to write data immediately
424 $cb->(); 438 $cb->();
… …
444 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 458 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
445 ->($self, @_); 459 ->($self, @_);
446 } 460 }
447 461
448 if ($self->{filter_w}) { 462 if ($self->{filter_w}) {
449 $self->{filter_w}->($self, \$_[0]); 463 $self->{filter_w}($self, \$_[0]);
450 } else { 464 } else {
451 $self->{wbuf} .= $_[0]; 465 $self->{wbuf} .= $_[0];
452 $self->_drain_wbuf; 466 $self->_drain_wbuf;
453 } 467 }
454} 468}
455 469
456=item $handle->push_write (type => @args) 470=item $handle->push_write (type => @args)
457 471
458=item $handle->unshift_write (type => @args)
459
460Instead 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
461the job by specifying a type and type-specific arguments. 473the job by specifying a type and type-specific arguments.
462 474
463Predefined 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
464drop by and tell us): 476drop by and tell us):
… …
468=item netstring => $string 480=item netstring => $string
469 481
470Formats the given value as netstring 482Formats the given value as netstring
471(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).
472 484
473=back
474
475=cut 485=cut
476 486
477register_write_type netstring => sub { 487register_write_type netstring => sub {
478 my ($self, $string) = @_; 488 my ($self, $string) = @_;
479 489
480 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
481}; 506};
482 507
483=item json => $array_or_hashref 508=item json => $array_or_hashref
484 509
485Encodes 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
… …
519 544
520 $self->{json} ? $self->{json}->encode ($ref) 545 $self->{json} ? $self->{json}->encode ($ref)
521 : JSON::encode_json ($ref) 546 : JSON::encode_json ($ref)
522}; 547};
523 548
549=back
550
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 551=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525 552
526This 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>.
527Whenever 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
528reference with the handle object and the remaining arguments. 555reference with the handle object and the remaining arguments.
… …
553enough 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
554or not. 581or not.
555 582
556In 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
557case, AnyEvent::Handle will call the first queued callback each time new 584case, 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>, 585data arrives (also the first time it is queued) and removes it when it has
559below). 586done its job (see C<push_read>, below).
560 587
561This 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
562a chunk of data, and AnyEvent::Handle will execute them in order. 589a chunk of data, and AnyEvent::Handle will execute them in order.
563 590
564Example 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
565the specified number of bytes which give an XML datagram. 592the specified number of bytes which give an XML datagram.
566 593
567 # in the default state, expect some header bytes 594 # in the default state, expect some header bytes
568 $handle->on_read (sub { 595 $handle->on_read (sub {
569 # 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)
570 shift->unshift_read_chunk (4, sub { 597 shift->unshift_read (chunk => 4, sub {
571 # header arrived, decode 598 # header arrived, decode
572 my $len = unpack "N", $_[1]; 599 my $len = unpack "N", $_[1];
573 600
574 # now read the payload 601 # now read the payload
575 shift->unshift_read_chunk ($len, sub { 602 shift->unshift_read (chunk => $len, sub {
576 my $xml = $_[1]; 603 my $xml = $_[1];
577 # handle xml 604 # handle xml
578 }); 605 });
579 }); 606 });
580 }); 607 });
… …
587 614
588 # request one 615 # request one
589 $handle->push_write ("request 1\015\012"); 616 $handle->push_write ("request 1\015\012");
590 617
591 # 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
592 $handle->push_read_line (sub { 619 $handle->push_read (line => sub {
593 # if we got an "OK", we have to _prepend_ another line, 620 # 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 621 # 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 622 # which are already in the queue when this callback is called
596 # we don't do this in case we got an error 623 # we don't do this in case we got an error
597 if ($_[1] eq "OK") { 624 if ($_[1] eq "OK") {
598 $_[0]->unshift_read_line (sub { 625 $_[0]->unshift_read (line => sub {
599 my $response = $_[1]; 626 my $response = $_[1];
600 ... 627 ...
601 }); 628 });
602 } 629 }
603 }); 630 });
604 631
605 # request two 632 # request two
606 $handle->push_write ("request 2\015\012"); 633 $handle->push_write ("request 2\015\012");
607 634
608 # simply read 64 bytes, always 635 # simply read 64 bytes, always
609 $handle->push_read_chunk (64, sub { 636 $handle->push_read (chunk => 64, sub {
610 my $response = $_[1]; 637 my $response = $_[1];
611 ... 638 ...
612 }); 639 });
613 640
614=over 4 641=over 4
615 642
616=cut 643=cut
617 644
618sub _drain_rbuf { 645sub _drain_rbuf {
619 my ($self) = @_; 646 my ($self) = @_;
647
648 local $self->{_in_drain} = 1;
620 649
621 if ( 650 if (
622 defined $self->{rbuf_max} 651 defined $self->{rbuf_max}
623 && $self->{rbuf_max} < length $self->{rbuf} 652 && $self->{rbuf_max} < length $self->{rbuf}
624 ) { 653 ) {
625 $! = &Errno::ENOSPC; 654 return $self->_error (&Errno::ENOSPC, 1);
626 $self->error;
627 } 655 }
628 656
629 return if $self->{in_drain}; 657 while () {
630 local $self->{in_drain} = 1;
631
632 while (my $len = length $self->{rbuf}) {
633 no strict 'refs'; 658 no strict 'refs';
659
660 my $len = length $self->{rbuf};
661
634 if (my $cb = shift @{ $self->{_queue} }) { 662 if (my $cb = shift @{ $self->{_queue} }) {
635 unless ($cb->($self)) { 663 unless ($cb->($self)) {
636 if ($self->{_eof}) { 664 if ($self->{_eof}) {
637 # 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)
638 $! = &Errno::EPIPE; 666 $self->_error (&Errno::EPIPE, 1), last;
639 $self->error;
640 } 667 }
641 668
642 unshift @{ $self->{_queue} }, $cb; 669 unshift @{ $self->{_queue} }, $cb;
643 return; 670 last;
644 } 671 }
645 } elsif ($self->{on_read}) { 672 } elsif ($self->{on_read}) {
673 last unless $len;
674
646 $self->{on_read}($self); 675 $self->{on_read}($self);
647 676
648 if ( 677 if (
649 $self->{_eof} # if no further data will arrive
650 && $len == length $self->{rbuf} # and no data has been consumed 678 $len == length $self->{rbuf} # if no data has been consumed
651 && !@{ $self->{_queue} } # and the queue is still empty 679 && !@{ $self->{_queue} } # and the queue is still empty
652 && $self->{on_read} # and we still want to read data 680 && $self->{on_read} # but we still have on_read
653 ) { 681 ) {
682 # no further data will arrive
654 # then no progress can be made 683 # so no progress can be made
655 $! = &Errno::EPIPE; 684 $self->_error (&Errno::EPIPE, 1), last
656 $self->error; 685 if $self->{_eof};
686
687 last; # more data might arrive
657 } 688 }
658 } else { 689 } else {
659 # read side becomes idle 690 # read side becomes idle
660 delete $self->{_rw}; 691 delete $self->{_rw};
661 return; 692 last;
662 } 693 }
663 } 694 }
664 695
665 if ($self->{_eof}) {
666 $self->_shutdown;
667 $self->{on_eof}($self) 696 $self->{on_eof}($self)
668 if $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} };
669 } 703 }
670} 704}
671 705
672=item $handle->on_read ($cb) 706=item $handle->on_read ($cb)
673 707
… …
679 713
680sub on_read { 714sub on_read {
681 my ($self, $cb) = @_; 715 my ($self, $cb) = @_;
682 716
683 $self->{on_read} = $cb; 717 $self->{on_read} = $cb;
718 $self->_drain_rbuf if $cb && !$self->{_in_drain};
684} 719}
685 720
686=item $handle->rbuf 721=item $handle->rbuf
687 722
688Returns the read buffer (as a modifiable lvalue). 723Returns the read buffer (as a modifiable lvalue).
… …
737 $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")
738 ->($self, $cb, @_); 773 ->($self, $cb, @_);
739 } 774 }
740 775
741 push @{ $self->{_queue} }, $cb; 776 push @{ $self->{_queue} }, $cb;
742 $self->_drain_rbuf; 777 $self->_drain_rbuf unless $self->{_in_drain};
743} 778}
744 779
745sub unshift_read { 780sub unshift_read {
746 my $self = shift; 781 my $self = shift;
747 my $cb = pop; 782 my $cb = pop;
… …
753 ->($self, $cb, @_); 788 ->($self, $cb, @_);
754 } 789 }
755 790
756 791
757 unshift @{ $self->{_queue} }, $cb; 792 unshift @{ $self->{_queue} }, $cb;
758 $self->_drain_rbuf; 793 $self->_drain_rbuf unless $self->{_in_drain};
759} 794}
760 795
761=item $handle->push_read (type => @args, $cb) 796=item $handle->push_read (type => @args, $cb)
762 797
763=item $handle->unshift_read (type => @args, $cb) 798=item $handle->unshift_read (type => @args, $cb)
… …
848 883
849sub unshift_read_line { 884sub unshift_read_line {
850 my $self = shift; 885 my $self = shift;
851 $self->unshift_read (line => @_); 886 $self->unshift_read (line => @_);
852} 887}
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 888
892=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 889=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
893 890
894Makes a regex match against the regex object C<$accept> and returns 891Makes a regex match against the regex object C<$accept> and returns
895everything up to and including the match. 892everything up to and including the match.
… …
945 return 1; 942 return 1;
946 } 943 }
947 944
948 # reject 945 # reject
949 if ($reject && $$rbuf =~ $reject) { 946 if ($reject && $$rbuf =~ $reject) {
950 $! = &Errno::EBADMSG; 947 $self->_error (&Errno::EBADMSG);
951 $self->error;
952 } 948 }
953 949
954 # skip 950 # skip
955 if ($skip && $$rbuf =~ $skip) { 951 if ($skip && $$rbuf =~ $skip) {
956 $data .= substr $$rbuf, 0, $+[0], ""; 952 $data .= substr $$rbuf, 0, $+[0], "";
957 } 953 }
958 954
959 () 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
960 } 1028 }
961}; 1029};
962 1030
963=item json => $cb->($handle, $hash_or_arrayref) 1031=item json => $cb->($handle, $hash_or_arrayref)
964 1032
… …
1029=item $handle->stop_read 1097=item $handle->stop_read
1030 1098
1031=item $handle->start_read 1099=item $handle->start_read
1032 1100
1033In 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
1034socket. 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
1035any queued callbacks will be executed then. To start reading again, call 1103any queued callbacks will be executed then. To start reading again, call
1036C<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.
1037 1110
1038=cut 1111=cut
1039 1112
1040sub stop_read { 1113sub stop_read {
1041 my ($self) = @_; 1114 my ($self) = @_;
… …
1055 1128
1056 if ($len > 0) { 1129 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now; 1130 $self->{_activity} = AnyEvent->now;
1058 1131
1059 $self->{filter_r} 1132 $self->{filter_r}
1060 ? $self->{filter_r}->($self, $rbuf) 1133 ? $self->{filter_r}($self, $rbuf)
1061 : $self->_drain_rbuf; 1134 : $self->{_in_drain} || $self->_drain_rbuf;
1062 1135
1063 } elsif (defined $len) { 1136 } elsif (defined $len) {
1064 delete $self->{_rw}; 1137 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
1067 $self->{_eof} = 1; 1138 $self->{_eof} = 1;
1068 $self->_drain_rbuf; 1139 $self->_drain_rbuf unless $self->{_in_drain};
1069 1140
1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1141 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1071 return $self->error; 1142 return $self->_error ($!, 1);
1072 } 1143 }
1073 }); 1144 });
1074 } 1145 }
1075} 1146}
1076 1147
1077sub _dotls { 1148sub _dotls {
1078 my ($self) = @_; 1149 my ($self) = @_;
1150
1151 my $buf;
1079 1152
1080 if (length $self->{_tls_wbuf}) { 1153 if (length $self->{_tls_wbuf}) {
1081 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) {
1082 substr $self->{_tls_wbuf}, 0, $len, ""; 1155 substr $self->{_tls_wbuf}, 0, $len, "";
1083 } 1156 }
1084 } 1157 }
1085 1158
1086 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1159 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1087 $self->{wbuf} .= $buf; 1160 $self->{wbuf} .= $buf;
1088 $self->_drain_wbuf; 1161 $self->_drain_wbuf;
1089 } 1162 }
1090 1163
1091 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1164 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1165 if (length $buf) {
1092 $self->{rbuf} .= $buf; 1166 $self->{rbuf} .= $buf;
1093 $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 }
1094 } 1174 }
1095 1175
1096 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1176 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1097 1177
1098 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1178 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1099 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1179 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1100 $self->error; 1180 return $self->_error ($!, 1);
1101 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1181 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1102 $! = &Errno::EIO; 1182 return $self->_error (&Errno::EIO, 1);
1103 $self->error;
1104 } 1183 }
1105 1184
1106 # all others are fine for our purposes 1185 # all others are fine for our purposes
1107 } 1186 }
1108} 1187}
… …
1123call 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
1124might have already started when this function returns. 1203might have already started when this function returns.
1125 1204
1126=cut 1205=cut
1127 1206
1128# TODO: maybe document...
1129sub starttls { 1207sub starttls {
1130 my ($self, $ssl, $ctx) = @_; 1208 my ($self, $ssl, $ctx) = @_;
1131 1209
1132 $self->stoptls; 1210 $self->stoptls;
1133 1211
… …
1186 1264
1187sub DESTROY { 1265sub DESTROY {
1188 my $self = shift; 1266 my $self = shift;
1189 1267
1190 $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 }
1191} 1291}
1192 1292
1193=item AnyEvent::Handle::TLS_CTX 1293=item AnyEvent::Handle::TLS_CTX
1194 1294
1195This 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