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.59 by root, Thu Jun 5 16:53:11 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.13;
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.
… …
222 if ($self->{tls}) { 224 if ($self->{tls}) {
223 require Net::SSLeay; 225 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 226 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 } 227 }
226 228
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; 229 $self->{_activity} = AnyEvent->now;
233 $self->_timeout; 230 $self->_timeout;
234 231
235 $self->start_read; 232 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
233 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
236 234
237 $self 235 $self
238} 236}
239 237
240sub _shutdown { 238sub _shutdown {
… …
242 240
243 delete $self->{_tw}; 241 delete $self->{_tw};
244 delete $self->{_rw}; 242 delete $self->{_rw};
245 delete $self->{_ww}; 243 delete $self->{_ww};
246 delete $self->{fh}; 244 delete $self->{fh};
247}
248 245
246 $self->stoptls;
247}
248
249sub error { 249sub _error {
250 my ($self) = @_; 250 my ($self, $errno, $fatal) = @_;
251 251
252 {
253 local $!;
254 $self->_shutdown; 252 $self->_shutdown
255 } 253 if $fatal;
256 254
257 $self->{on_error}($self) 255 $! = $errno;
256
258 if $self->{on_error}; 257 if ($self->{on_error}) {
259 258 $self->{on_error}($self, $fatal);
259 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 260 Carp::croak "AnyEvent::Handle uncaught error: $!";
261 }
261} 262}
262 263
263=item $fh = $handle->fh 264=item $fh = $handle->fh
264 265
265This method returns the file handle of the L<AnyEvent::Handle> object. 266This method returns the file handle of the L<AnyEvent::Handle> object.
… …
331 $self->{_activity} = $NOW; 332 $self->{_activity} = $NOW;
332 333
333 if ($self->{on_timeout}) { 334 if ($self->{on_timeout}) {
334 $self->{on_timeout}($self); 335 $self->{on_timeout}($self);
335 } else { 336 } else {
336 $! = Errno::ETIMEDOUT; 337 $self->_error (&Errno::ETIMEDOUT);
337 $self->error;
338 } 338 }
339 339
340 # callbakx could have changed timeout value, optimise 340 # callback could have changed timeout value, optimise
341 return unless $self->{timeout}; 341 return unless $self->{timeout};
342 342
343 # calculate new after 343 # calculate new after
344 $after = $self->{timeout}; 344 $after = $self->{timeout};
345 } 345 }
346 346
347 Scalar::Util::weaken $self; 347 Scalar::Util::weaken $self;
348 return unless $self; # ->error could have destroyed $self
348 349
349 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 350 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
350 delete $self->{_tw}; 351 delete $self->{_tw};
351 $self->_timeout; 352 $self->_timeout;
352 }); 353 });
… …
415 if $self->{low_water_mark} >= length $self->{wbuf} 416 if $self->{low_water_mark} >= length $self->{wbuf}
416 && $self->{on_drain}; 417 && $self->{on_drain};
417 418
418 delete $self->{_ww} unless length $self->{wbuf}; 419 delete $self->{_ww} unless length $self->{wbuf};
419 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 420 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
420 $self->error; 421 $self->_error ($!, 1);
421 } 422 }
422 }; 423 };
423 424
424 # try to write data immediately 425 # try to write data immediately
425 $cb->(); 426 $cb->();
… …
454 } 455 }
455} 456}
456 457
457=item $handle->push_write (type => @args) 458=item $handle->push_write (type => @args)
458 459
459=item $handle->unshift_write (type => @args)
460
461Instead of formatting your data yourself, you can also let this module do 460Instead of formatting your data yourself, you can also let this module do
462the job by specifying a type and type-specific arguments. 461the job by specifying a type and type-specific arguments.
463 462
464Predefined types are (if you have ideas for additional types, feel free to 463Predefined types are (if you have ideas for additional types, feel free to
465drop by and tell us): 464drop by and tell us):
… …
468 467
469=item netstring => $string 468=item netstring => $string
470 469
471Formats the given value as netstring 470Formats the given value as netstring
472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 471(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
473
474=back
475 472
476=cut 473=cut
477 474
478register_write_type netstring => sub { 475register_write_type netstring => sub {
479 my ($self, $string) = @_; 476 my ($self, $string) = @_;
… …
520 517
521 $self->{json} ? $self->{json}->encode ($ref) 518 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref) 519 : JSON::encode_json ($ref)
523}; 520};
524 521
522=back
523
525=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
526 525
527This function (not method) lets you add your own types to C<push_write>. 526This 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 527Whenever the given C<type> is used, C<push_write> will invoke the code
529reference with the handle object and the remaining arguments. 528reference with the handle object and the remaining arguments.
… …
566the specified number of bytes which give an XML datagram. 565the specified number of bytes which give an XML datagram.
567 566
568 # in the default state, expect some header bytes 567 # in the default state, expect some header bytes
569 $handle->on_read (sub { 568 $handle->on_read (sub {
570 # some data is here, now queue the length-header-read (4 octets) 569 # some data is here, now queue the length-header-read (4 octets)
571 shift->unshift_read_chunk (4, sub { 570 shift->unshift_read (chunk => 4, sub {
572 # header arrived, decode 571 # header arrived, decode
573 my $len = unpack "N", $_[1]; 572 my $len = unpack "N", $_[1];
574 573
575 # now read the payload 574 # now read the payload
576 shift->unshift_read_chunk ($len, sub { 575 shift->unshift_read (chunk => $len, sub {
577 my $xml = $_[1]; 576 my $xml = $_[1];
578 # handle xml 577 # handle xml
579 }); 578 });
580 }); 579 });
581 }); 580 });
… …
588 587
589 # request one 588 # request one
590 $handle->push_write ("request 1\015\012"); 589 $handle->push_write ("request 1\015\012");
591 590
592 # we expect "ERROR" or "OK" as response, so push a line read 591 # we expect "ERROR" or "OK" as response, so push a line read
593 $handle->push_read_line (sub { 592 $handle->push_read (line => sub {
594 # if we got an "OK", we have to _prepend_ another line, 593 # 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 594 # 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 595 # which are already in the queue when this callback is called
597 # we don't do this in case we got an error 596 # we don't do this in case we got an error
598 if ($_[1] eq "OK") { 597 if ($_[1] eq "OK") {
599 $_[0]->unshift_read_line (sub { 598 $_[0]->unshift_read (line => sub {
600 my $response = $_[1]; 599 my $response = $_[1];
601 ... 600 ...
602 }); 601 });
603 } 602 }
604 }); 603 });
605 604
606 # request two 605 # request two
607 $handle->push_write ("request 2\015\012"); 606 $handle->push_write ("request 2\015\012");
608 607
609 # simply read 64 bytes, always 608 # simply read 64 bytes, always
610 $handle->push_read_chunk (64, sub { 609 $handle->push_read (chunk => 64, sub {
611 my $response = $_[1]; 610 my $response = $_[1];
612 ... 611 ...
613 }); 612 });
614 613
615=over 4 614=over 4
616 615
617=cut 616=cut
618 617
619sub _drain_rbuf { 618sub _drain_rbuf {
620 my ($self) = @_; 619 my ($self) = @_;
620
621 local $self->{_in_drain} = 1;
621 622
622 if ( 623 if (
623 defined $self->{rbuf_max} 624 defined $self->{rbuf_max}
624 && $self->{rbuf_max} < length $self->{rbuf} 625 && $self->{rbuf_max} < length $self->{rbuf}
625 ) { 626 ) {
626 $! = &Errno::ENOSPC; 627 return $self->_error (&Errno::ENOSPC, 1);
627 $self->error;
628 } 628 }
629 629
630 return if $self->{in_drain}; 630 while () {
631 local $self->{in_drain} = 1;
632
633 while (my $len = length $self->{rbuf}) {
634 no strict 'refs'; 631 no strict 'refs';
632
633 my $len = length $self->{rbuf};
634
635 if (my $cb = shift @{ $self->{_queue} }) { 635 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 636 unless ($cb->($self)) {
637 if ($self->{_eof}) { 637 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming) 638 # no progress can be made (not enough data and no data forthcoming)
639 $! = &Errno::EPIPE; 639 return $self->_error (&Errno::EPIPE, 1);
640 $self->error;
641 } 640 }
642 641
643 unshift @{ $self->{_queue} }, $cb; 642 unshift @{ $self->{_queue} }, $cb;
644 return; 643 last;
645 } 644 }
646 } elsif ($self->{on_read}) { 645 } elsif ($self->{on_read}) {
647 $self->{on_read}($self); 646 $self->{on_read}($self);
648 647
649 if ( 648 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed 649 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 650 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data 651 && $self->{on_read} # but we still have on_read
654 ) { 652 ) {
653 # no further data will arrive
655 # then no progress can be made 654 # so no progress can be made
656 $! = &Errno::EPIPE; 655 return $self->_error (&Errno::EPIPE, 1)
657 $self->error; 656 if $self->{_eof};
657
658 last; # more data might arrive
658 } 659 }
659 } else { 660 } else {
660 # read side becomes idle 661 # read side becomes idle
661 delete $self->{_rw}; 662 delete $self->{_rw};
662 return; 663 last;
663 } 664 }
664 } 665 }
665 666
666 $self->{on_eof}($self) 667 $self->{on_eof}($self)
667 if $self->{_eof} && $self->{on_eof}; 668 if $self->{_eof} && $self->{on_eof};
669
670 # may need to restart read watcher
671 unless ($self->{_rw}) {
672 $self->start_read
673 if $self->{on_read} || @{ $self->{_queue} };
674 }
668} 675}
669 676
670=item $handle->on_read ($cb) 677=item $handle->on_read ($cb)
671 678
672This replaces the currently set C<on_read> callback, or clears it (when 679This replaces the currently set C<on_read> callback, or clears it (when
… …
677 684
678sub on_read { 685sub on_read {
679 my ($self, $cb) = @_; 686 my ($self, $cb) = @_;
680 687
681 $self->{on_read} = $cb; 688 $self->{on_read} = $cb;
689 $self->_drain_rbuf if $cb && !$self->{_in_drain};
682} 690}
683 691
684=item $handle->rbuf 692=item $handle->rbuf
685 693
686Returns the read buffer (as a modifiable lvalue). 694Returns the read buffer (as a modifiable lvalue).
… …
735 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 743 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
736 ->($self, $cb, @_); 744 ->($self, $cb, @_);
737 } 745 }
738 746
739 push @{ $self->{_queue} }, $cb; 747 push @{ $self->{_queue} }, $cb;
740 $self->_drain_rbuf; 748 $self->_drain_rbuf unless $self->{_in_drain};
741} 749}
742 750
743sub unshift_read { 751sub unshift_read {
744 my $self = shift; 752 my $self = shift;
745 my $cb = pop; 753 my $cb = pop;
… …
751 ->($self, $cb, @_); 759 ->($self, $cb, @_);
752 } 760 }
753 761
754 762
755 unshift @{ $self->{_queue} }, $cb; 763 unshift @{ $self->{_queue} }, $cb;
756 $self->_drain_rbuf; 764 $self->_drain_rbuf unless $self->{_in_drain};
757} 765}
758 766
759=item $handle->push_read (type => @args, $cb) 767=item $handle->push_read (type => @args, $cb)
760 768
761=item $handle->unshift_read (type => @args, $cb) 769=item $handle->unshift_read (type => @args, $cb)
… …
861 my ($self, $cb) = @_; 869 my ($self, $cb) = @_;
862 870
863 sub { 871 sub {
864 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 872 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
865 if ($_[0]{rbuf} =~ /[^0-9]/) { 873 if ($_[0]{rbuf} =~ /[^0-9]/) {
866 $! = &Errno::EBADMSG; 874 $self->_error (&Errno::EBADMSG);
867 $self->error;
868 } 875 }
869 return; 876 return;
870 } 877 }
871 878
872 my $len = $1; 879 my $len = $1;
… …
875 my $string = $_[1]; 882 my $string = $_[1];
876 $_[0]->unshift_read (chunk => 1, sub { 883 $_[0]->unshift_read (chunk => 1, sub {
877 if ($_[1] eq ",") { 884 if ($_[1] eq ",") {
878 $cb->($_[0], $string); 885 $cb->($_[0], $string);
879 } else { 886 } else {
880 $! = &Errno::EBADMSG; 887 $self->_error (&Errno::EBADMSG);
881 $self->error;
882 } 888 }
883 }); 889 });
884 }); 890 });
885 891
886 1 892 1
… …
943 return 1; 949 return 1;
944 } 950 }
945 951
946 # reject 952 # reject
947 if ($reject && $$rbuf =~ $reject) { 953 if ($reject && $$rbuf =~ $reject) {
948 $! = &Errno::EBADMSG; 954 $self->_error (&Errno::EBADMSG);
949 $self->error;
950 } 955 }
951 956
952 # skip 957 # skip
953 if ($skip && $$rbuf =~ $skip) { 958 if ($skip && $$rbuf =~ $skip) {
954 $data .= substr $$rbuf, 0, $+[0], ""; 959 $data .= substr $$rbuf, 0, $+[0], "";
… …
1027=item $handle->stop_read 1032=item $handle->stop_read
1028 1033
1029=item $handle->start_read 1034=item $handle->start_read
1030 1035
1031In rare cases you actually do not want to read anything from the 1036In 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 1037socket. 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 1038any queued callbacks will be executed then. To start reading again, call
1034C<start_read>. 1039C<start_read>.
1040
1041Note that AnyEvent::Handle will automatically C<start_read> for you when
1042you change the C<on_read> callback or push/unshift a read callback, and it
1043will automatically C<stop_read> for you when neither C<on_read> is set nor
1044there are any read requests in the queue.
1035 1045
1036=cut 1046=cut
1037 1047
1038sub stop_read { 1048sub stop_read {
1039 my ($self) = @_; 1049 my ($self) = @_;
… …
1054 if ($len > 0) { 1064 if ($len > 0) {
1055 $self->{_activity} = AnyEvent->now; 1065 $self->{_activity} = AnyEvent->now;
1056 1066
1057 $self->{filter_r} 1067 $self->{filter_r}
1058 ? $self->{filter_r}($self, $rbuf) 1068 ? $self->{filter_r}($self, $rbuf)
1059 : $self->_drain_rbuf; 1069 : $self->{_in_drain} || $self->_drain_rbuf;
1060 1070
1061 } elsif (defined $len) { 1071 } elsif (defined $len) {
1062 delete $self->{_rw}; 1072 delete $self->{_rw};
1063 $self->{_eof} = 1; 1073 $self->{_eof} = 1;
1064 $self->_drain_rbuf; 1074 $self->_drain_rbuf unless $self->{_in_drain};
1065 1075
1066 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1076 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1067 return $self->error; 1077 return $self->_error ($!, 1);
1068 } 1078 }
1069 }); 1079 });
1070 } 1080 }
1071} 1081}
1072 1082
1073sub _dotls { 1083sub _dotls {
1074 my ($self) = @_; 1084 my ($self) = @_;
1085
1086 my $buf;
1075 1087
1076 if (length $self->{_tls_wbuf}) { 1088 if (length $self->{_tls_wbuf}) {
1077 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1089 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1078 substr $self->{_tls_wbuf}, 0, $len, ""; 1090 substr $self->{_tls_wbuf}, 0, $len, "";
1079 } 1091 }
1080 } 1092 }
1081 1093
1082 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1094 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1083 $self->{wbuf} .= $buf; 1095 $self->{wbuf} .= $buf;
1084 $self->_drain_wbuf; 1096 $self->_drain_wbuf;
1085 } 1097 }
1086 1098
1087 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1099 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1100 if (length $buf) {
1088 $self->{rbuf} .= $buf; 1101 $self->{rbuf} .= $buf;
1089 $self->_drain_rbuf; 1102 $self->_drain_rbuf unless $self->{_in_drain};
1103 } else {
1104 # let's treat SSL-eof as we treat normal EOF
1105 $self->{_eof} = 1;
1106 $self->_shutdown;
1107 return;
1108 }
1090 } 1109 }
1091 1110
1092 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1111 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1093 1112
1094 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1113 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1095 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1114 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1096 $self->error; 1115 return $self->_error ($!, 1);
1097 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1116 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1098 $! = &Errno::EIO; 1117 return $self->_error (&Errno::EIO, 1);
1099 $self->error;
1100 } 1118 }
1101 1119
1102 # all others are fine for our purposes 1120 # all others are fine for our purposes
1103 } 1121 }
1104} 1122}
… …
1119call and can be used or changed to your liking. Note that the handshake 1137call and can be used or changed to your liking. Note that the handshake
1120might have already started when this function returns. 1138might have already started when this function returns.
1121 1139
1122=cut 1140=cut
1123 1141
1124# TODO: maybe document...
1125sub starttls { 1142sub starttls {
1126 my ($self, $ssl, $ctx) = @_; 1143 my ($self, $ssl, $ctx) = @_;
1127 1144
1128 $self->stoptls; 1145 $self->stoptls;
1129 1146

Diff Legend

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