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.56 by root, Wed Jun 4 09:55:16 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.12;
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.
… …
155isn't finished). 157isn't finished).
156 158
157=item read_size => <bytes> 159=item read_size => <bytes>
158 160
159The default read block size (the amount of bytes this module will try to read 161The default read block size (the amount of bytes this module will try to read
160on each [loop iteration). Default: C<4096>. 162during each (loop iteration). Default: C<8192>.
161 163
162=item low_water_mark => <bytes> 164=item low_water_mark => <bytes>
163 165
164Sets the amount of bytes (default: C<0>) that make up an "empty" write 166Sets 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 167buffer: If the write reaches this size or gets even samller it is
… …
238} 240}
239 241
240sub _shutdown { 242sub _shutdown {
241 my ($self) = @_; 243 my ($self) = @_;
242 244
245 delete $self->{_tw};
243 delete $self->{_rw}; 246 delete $self->{_rw};
244 delete $self->{_ww}; 247 delete $self->{_ww};
245 delete $self->{fh}; 248 delete $self->{fh};
246}
247 249
250 $self->stoptls;
251}
252
248sub error { 253sub _error {
249 my ($self) = @_; 254 my ($self, $errno, $fatal) = @_;
250 255
251 {
252 local $!;
253 $self->_shutdown; 256 $self->_shutdown
254 } 257 if $fatal;
255 258
256 $self->{on_error}($self) 259 $! = $errno;
260
257 if $self->{on_error}; 261 if ($self->{on_error}) {
258 262 $self->{on_error}($self, $fatal);
263 } else {
259 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 264 Carp::croak "AnyEvent::Handle uncaught error: $!";
265 }
260} 266}
261 267
262=item $fh = $handle->fh 268=item $fh = $handle->fh
263 269
264This method returns the file handle of the L<AnyEvent::Handle> object. 270This method returns the file handle of the L<AnyEvent::Handle> object.
… …
328 # now or in the past already? 334 # now or in the past already?
329 if ($after <= 0) { 335 if ($after <= 0) {
330 $self->{_activity} = $NOW; 336 $self->{_activity} = $NOW;
331 337
332 if ($self->{on_timeout}) { 338 if ($self->{on_timeout}) {
333 $self->{on_timeout}->($self); 339 $self->{on_timeout}($self);
334 } else { 340 } else {
335 $! = Errno::ETIMEDOUT; 341 $self->_error (&Errno::ETIMEDOUT);
336 $self->error;
337 } 342 }
338 343
339 # callbakx could have changed timeout value, optimise 344 # callback could have changed timeout value, optimise
340 return unless $self->{timeout}; 345 return unless $self->{timeout};
341 346
342 # calculate new after 347 # calculate new after
343 $after = $self->{timeout}; 348 $after = $self->{timeout};
344 } 349 }
345 350
346 Scalar::Util::weaken $self; 351 Scalar::Util::weaken $self;
352 return unless $self; # ->error could have destroyed $self
347 353
348 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 354 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
349 delete $self->{_tw}; 355 delete $self->{_tw};
350 $self->_timeout; 356 $self->_timeout;
351 }); 357 });
… …
414 if $self->{low_water_mark} >= length $self->{wbuf} 420 if $self->{low_water_mark} >= length $self->{wbuf}
415 && $self->{on_drain}; 421 && $self->{on_drain};
416 422
417 delete $self->{_ww} unless length $self->{wbuf}; 423 delete $self->{_ww} unless length $self->{wbuf};
418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 424 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
419 $self->error; 425 $self->_error ($!, 1);
420 } 426 }
421 }; 427 };
422 428
423 # try to write data immediately 429 # try to write data immediately
424 $cb->(); 430 $cb->();
… …
444 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 450 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
445 ->($self, @_); 451 ->($self, @_);
446 } 452 }
447 453
448 if ($self->{filter_w}) { 454 if ($self->{filter_w}) {
449 $self->{filter_w}->($self, \$_[0]); 455 $self->{filter_w}($self, \$_[0]);
450 } else { 456 } else {
451 $self->{wbuf} .= $_[0]; 457 $self->{wbuf} .= $_[0];
452 $self->_drain_wbuf; 458 $self->_drain_wbuf;
453 } 459 }
454} 460}
455 461
456=item $handle->push_write (type => @args) 462=item $handle->push_write (type => @args)
457 463
458=item $handle->unshift_write (type => @args)
459
460Instead of formatting your data yourself, you can also let this module do 464Instead of formatting your data yourself, you can also let this module do
461the job by specifying a type and type-specific arguments. 465the job by specifying a type and type-specific arguments.
462 466
463Predefined types are (if you have ideas for additional types, feel free to 467Predefined types are (if you have ideas for additional types, feel free to
464drop by and tell us): 468drop by and tell us):
… …
467 471
468=item netstring => $string 472=item netstring => $string
469 473
470Formats the given value as netstring 474Formats the given value as netstring
471(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 475(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
472
473=back
474 476
475=cut 477=cut
476 478
477register_write_type netstring => sub { 479register_write_type netstring => sub {
478 my ($self, $string) = @_; 480 my ($self, $string) = @_;
… …
519 521
520 $self->{json} ? $self->{json}->encode ($ref) 522 $self->{json} ? $self->{json}->encode ($ref)
521 : JSON::encode_json ($ref) 523 : JSON::encode_json ($ref)
522}; 524};
523 525
526=back
527
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 528=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525 529
526This function (not method) lets you add your own types to C<push_write>. 530This 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 531Whenever the given C<type> is used, C<push_write> will invoke the code
528reference with the handle object and the remaining arguments. 532reference with the handle object and the remaining arguments.
… …
565the specified number of bytes which give an XML datagram. 569the specified number of bytes which give an XML datagram.
566 570
567 # in the default state, expect some header bytes 571 # in the default state, expect some header bytes
568 $handle->on_read (sub { 572 $handle->on_read (sub {
569 # some data is here, now queue the length-header-read (4 octets) 573 # some data is here, now queue the length-header-read (4 octets)
570 shift->unshift_read_chunk (4, sub { 574 shift->unshift_read (chunk => 4, sub {
571 # header arrived, decode 575 # header arrived, decode
572 my $len = unpack "N", $_[1]; 576 my $len = unpack "N", $_[1];
573 577
574 # now read the payload 578 # now read the payload
575 shift->unshift_read_chunk ($len, sub { 579 shift->unshift_read (chunk => $len, sub {
576 my $xml = $_[1]; 580 my $xml = $_[1];
577 # handle xml 581 # handle xml
578 }); 582 });
579 }); 583 });
580 }); 584 });
… …
587 591
588 # request one 592 # request one
589 $handle->push_write ("request 1\015\012"); 593 $handle->push_write ("request 1\015\012");
590 594
591 # we expect "ERROR" or "OK" as response, so push a line read 595 # we expect "ERROR" or "OK" as response, so push a line read
592 $handle->push_read_line (sub { 596 $handle->push_read (line => sub {
593 # if we got an "OK", we have to _prepend_ another line, 597 # 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 598 # 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 599 # which are already in the queue when this callback is called
596 # we don't do this in case we got an error 600 # we don't do this in case we got an error
597 if ($_[1] eq "OK") { 601 if ($_[1] eq "OK") {
598 $_[0]->unshift_read_line (sub { 602 $_[0]->unshift_read (line => sub {
599 my $response = $_[1]; 603 my $response = $_[1];
600 ... 604 ...
601 }); 605 });
602 } 606 }
603 }); 607 });
604 608
605 # request two 609 # request two
606 $handle->push_write ("request 2\015\012"); 610 $handle->push_write ("request 2\015\012");
607 611
608 # simply read 64 bytes, always 612 # simply read 64 bytes, always
609 $handle->push_read_chunk (64, sub { 613 $handle->push_read (chunk => 64, sub {
610 my $response = $_[1]; 614 my $response = $_[1];
611 ... 615 ...
612 }); 616 });
613 617
614=over 4 618=over 4
… …
620 624
621 if ( 625 if (
622 defined $self->{rbuf_max} 626 defined $self->{rbuf_max}
623 && $self->{rbuf_max} < length $self->{rbuf} 627 && $self->{rbuf_max} < length $self->{rbuf}
624 ) { 628 ) {
625 $! = &Errno::ENOSPC; 629 return $self->_error (&Errno::ENOSPC, 1);
626 $self->error;
627 } 630 }
628 631
629 return if $self->{in_drain}; 632 return if $self->{in_drain};
630 local $self->{in_drain} = 1; 633 local $self->{in_drain} = 1;
631 634
… …
633 no strict 'refs'; 636 no strict 'refs';
634 if (my $cb = shift @{ $self->{_queue} }) { 637 if (my $cb = shift @{ $self->{_queue} }) {
635 unless ($cb->($self)) { 638 unless ($cb->($self)) {
636 if ($self->{_eof}) { 639 if ($self->{_eof}) {
637 # no progress can be made (not enough data and no data forthcoming) 640 # no progress can be made (not enough data and no data forthcoming)
638 $! = &Errno::EPIPE; 641 return $self->_error (&Errno::EPIPE, 1);
639 $self->error;
640 } 642 }
641 643
642 unshift @{ $self->{_queue} }, $cb; 644 unshift @{ $self->{_queue} }, $cb;
643 return; 645 last;
644 } 646 }
645 } elsif ($self->{on_read}) { 647 } elsif ($self->{on_read}) {
646 $self->{on_read}($self); 648 $self->{on_read}($self);
647 649
648 if ( 650 if (
649 $self->{_eof} # if no further data will arrive
650 && $len == length $self->{rbuf} # and no data has been consumed 651 $len == length $self->{rbuf} # if no data has been consumed
651 && !@{ $self->{_queue} } # and the queue is still empty 652 && !@{ $self->{_queue} } # and the queue is still empty
652 && $self->{on_read} # and we still want to read data 653 && $self->{on_read} # but we still have on_read
653 ) { 654 ) {
655 # no further data will arrive
654 # then no progress can be made 656 # so no progress can be made
655 $! = &Errno::EPIPE; 657 return $self->_error (&Errno::EPIPE, 1)
656 $self->error; 658 if $self->{_eof};
659
660 last; # more data might arrive
657 } 661 }
658 } else { 662 } else {
659 # read side becomes idle 663 # read side becomes idle
660 delete $self->{_rw}; 664 delete $self->{_rw};
661 return; 665 last;
662 } 666 }
663 } 667 }
664 668
665 if ($self->{_eof}) {
666 $self->_shutdown;
667 $self->{on_eof}($self) 669 $self->{on_eof}($self)
668 if $self->{on_eof}; 670 if $self->{_eof} && $self->{on_eof};
671
672 # may need to restart read watcher
673 unless ($self->{_rw}) {
674 $self->start_read
675 if $self->{on_read} || @{ $self->{_queue} };
669 } 676 }
670} 677}
671 678
672=item $handle->on_read ($cb) 679=item $handle->on_read ($cb)
673 680
… …
863 my ($self, $cb) = @_; 870 my ($self, $cb) = @_;
864 871
865 sub { 872 sub {
866 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 873 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
867 if ($_[0]{rbuf} =~ /[^0-9]/) { 874 if ($_[0]{rbuf} =~ /[^0-9]/) {
868 $! = &Errno::EBADMSG; 875 $self->_error (&Errno::EBADMSG);
869 $self->error;
870 } 876 }
871 return; 877 return;
872 } 878 }
873 879
874 my $len = $1; 880 my $len = $1;
… …
877 my $string = $_[1]; 883 my $string = $_[1];
878 $_[0]->unshift_read (chunk => 1, sub { 884 $_[0]->unshift_read (chunk => 1, sub {
879 if ($_[1] eq ",") { 885 if ($_[1] eq ",") {
880 $cb->($_[0], $string); 886 $cb->($_[0], $string);
881 } else { 887 } else {
882 $! = &Errno::EBADMSG; 888 $self->_error (&Errno::EBADMSG);
883 $self->error;
884 } 889 }
885 }); 890 });
886 }); 891 });
887 892
888 1 893 1
… …
945 return 1; 950 return 1;
946 } 951 }
947 952
948 # reject 953 # reject
949 if ($reject && $$rbuf =~ $reject) { 954 if ($reject && $$rbuf =~ $reject) {
950 $! = &Errno::EBADMSG; 955 $self->_error (&Errno::EBADMSG);
951 $self->error;
952 } 956 }
953 957
954 # skip 958 # skip
955 if ($skip && $$rbuf =~ $skip) { 959 if ($skip && $$rbuf =~ $skip) {
956 $data .= substr $$rbuf, 0, $+[0], ""; 960 $data .= substr $$rbuf, 0, $+[0], "";
… …
1033In rare cases you actually do not want to read anything from the 1037In 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 1038socket. In this case you can call C<stop_read>. Neither C<on_read> no
1035any queued callbacks will be executed then. To start reading again, call 1039any queued callbacks will be executed then. To start reading again, call
1036C<start_read>. 1040C<start_read>.
1037 1041
1042Note that AnyEvent::Handle will automatically C<start_read> for you when
1043you change the C<on_read> callback or push/unshift a read callback, and it
1044will automatically C<stop_read> for you when neither C<on_read> is set nor
1045there are any read requests in the queue.
1046
1038=cut 1047=cut
1039 1048
1040sub stop_read { 1049sub stop_read {
1041 my ($self) = @_; 1050 my ($self) = @_;
1042 1051
… …
1055 1064
1056 if ($len > 0) { 1065 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now; 1066 $self->{_activity} = AnyEvent->now;
1058 1067
1059 $self->{filter_r} 1068 $self->{filter_r}
1060 ? $self->{filter_r}->($self, $rbuf) 1069 ? $self->{filter_r}($self, $rbuf)
1061 : $self->_drain_rbuf; 1070 : $self->_drain_rbuf;
1062 1071
1063 } elsif (defined $len) { 1072 } elsif (defined $len) {
1064 delete $self->{_rw}; 1073 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
1067 $self->{_eof} = 1; 1074 $self->{_eof} = 1;
1068 $self->_drain_rbuf; 1075 $self->_drain_rbuf;
1069 1076
1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1077 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1071 return $self->error; 1078 return $self->_error ($!, 1);
1072 } 1079 }
1073 }); 1080 });
1074 } 1081 }
1075} 1082}
1076 1083
1077sub _dotls { 1084sub _dotls {
1078 my ($self) = @_; 1085 my ($self) = @_;
1086
1087 my $buf;
1079 1088
1080 if (length $self->{_tls_wbuf}) { 1089 if (length $self->{_tls_wbuf}) {
1081 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1090 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1082 substr $self->{_tls_wbuf}, 0, $len, ""; 1091 substr $self->{_tls_wbuf}, 0, $len, "";
1083 } 1092 }
1084 } 1093 }
1085 1094
1086 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1095 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1087 $self->{wbuf} .= $buf; 1096 $self->{wbuf} .= $buf;
1088 $self->_drain_wbuf; 1097 $self->_drain_wbuf;
1089 } 1098 }
1090 1099
1091 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1100 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1101 if (length $buf) {
1092 $self->{rbuf} .= $buf; 1102 $self->{rbuf} .= $buf;
1093 $self->_drain_rbuf; 1103 $self->_drain_rbuf;
1104 } else {
1105 # let's treat SSL-eof as we treat normal EOF
1106 $self->{_eof} = 1;
1107 $self->_shutdown;
1108 return;
1109 }
1094 } 1110 }
1095 1111
1096 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1112 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1097 1113
1098 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1114 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1099 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1115 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1100 $self->error; 1116 return $self->_error ($!, 1);
1101 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1117 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1102 $! = &Errno::EIO; 1118 return $self->_error (&Errno::EIO, 1);
1103 $self->error;
1104 } 1119 }
1105 1120
1106 # all others are fine for our purposes 1121 # all others are fine for our purposes
1107 } 1122 }
1108} 1123}
… …
1123call and can be used or changed to your liking. Note that the handshake 1138call and can be used or changed to your liking. Note that the handshake
1124might have already started when this function returns. 1139might have already started when this function returns.
1125 1140
1126=cut 1141=cut
1127 1142
1128# TODO: maybe document...
1129sub starttls { 1143sub starttls {
1130 my ($self, $ssl, $ctx) = @_; 1144 my ($self, $ssl, $ctx) = @_;
1131 1145
1132 $self->stoptls; 1146 $self->stoptls;
1133 1147

Diff Legend

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