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.49 by root, Thu May 29 03:45:37 2008 UTC vs.
Revision 1.58 by root, Wed Jun 4 22:51:15 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 = '1.0'; 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
… …
621 620
622 if ( 621 if (
623 defined $self->{rbuf_max} 622 defined $self->{rbuf_max}
624 && $self->{rbuf_max} < length $self->{rbuf} 623 && $self->{rbuf_max} < length $self->{rbuf}
625 ) { 624 ) {
626 $! = &Errno::ENOSPC; 625 return $self->_error (&Errno::ENOSPC, 1);
627 $self->error;
628 } 626 }
629 627
630 return if $self->{in_drain}; 628 return if $self->{in_drain};
631 local $self->{in_drain} = 1; 629 local $self->{in_drain} = 1;
632 630
… …
634 no strict 'refs'; 632 no strict 'refs';
635 if (my $cb = shift @{ $self->{_queue} }) { 633 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 634 unless ($cb->($self)) {
637 if ($self->{_eof}) { 635 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming) 636 # no progress can be made (not enough data and no data forthcoming)
639 $! = &Errno::EPIPE; 637 return $self->_error (&Errno::EPIPE, 1);
640 $self->error;
641 } 638 }
642 639
643 unshift @{ $self->{_queue} }, $cb; 640 unshift @{ $self->{_queue} }, $cb;
644 return; 641 last;
645 } 642 }
646 } elsif ($self->{on_read}) { 643 } elsif ($self->{on_read}) {
647 $self->{on_read}($self); 644 $self->{on_read}($self);
648 645
649 if ( 646 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed 647 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 648 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data 649 && $self->{on_read} # but we still have on_read
654 ) { 650 ) {
651 # no further data will arrive
655 # then no progress can be made 652 # so no progress can be made
656 $! = &Errno::EPIPE; 653 return $self->_error (&Errno::EPIPE, 1)
657 $self->error; 654 if $self->{_eof};
655
656 last; # more data might arrive
658 } 657 }
659 } else { 658 } else {
660 # read side becomes idle 659 # read side becomes idle
661 delete $self->{_rw}; 660 delete $self->{_rw};
662 return; 661 last;
663 } 662 }
664 } 663 }
665 664
666 $self->{on_eof}($self) 665 $self->{on_eof}($self)
667 if $self->{_eof} && $self->{on_eof}; 666 if $self->{_eof} && $self->{on_eof};
667
668 # may need to restart read watcher
669 unless ($self->{_rw}) {
670 $self->start_read
671 if $self->{on_read} || @{ $self->{_queue} };
672 }
668} 673}
669 674
670=item $handle->on_read ($cb) 675=item $handle->on_read ($cb)
671 676
672This replaces the currently set C<on_read> callback, or clears it (when 677This replaces the currently set C<on_read> callback, or clears it (when
… …
677 682
678sub on_read { 683sub on_read {
679 my ($self, $cb) = @_; 684 my ($self, $cb) = @_;
680 685
681 $self->{on_read} = $cb; 686 $self->{on_read} = $cb;
687 $self->_drain_rbuf if $cb;
682} 688}
683 689
684=item $handle->rbuf 690=item $handle->rbuf
685 691
686Returns the read buffer (as a modifiable lvalue). 692Returns the read buffer (as a modifiable lvalue).
… …
861 my ($self, $cb) = @_; 867 my ($self, $cb) = @_;
862 868
863 sub { 869 sub {
864 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 870 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
865 if ($_[0]{rbuf} =~ /[^0-9]/) { 871 if ($_[0]{rbuf} =~ /[^0-9]/) {
866 $! = &Errno::EBADMSG; 872 $self->_error (&Errno::EBADMSG);
867 $self->error;
868 } 873 }
869 return; 874 return;
870 } 875 }
871 876
872 my $len = $1; 877 my $len = $1;
… …
875 my $string = $_[1]; 880 my $string = $_[1];
876 $_[0]->unshift_read (chunk => 1, sub { 881 $_[0]->unshift_read (chunk => 1, sub {
877 if ($_[1] eq ",") { 882 if ($_[1] eq ",") {
878 $cb->($_[0], $string); 883 $cb->($_[0], $string);
879 } else { 884 } else {
880 $! = &Errno::EBADMSG; 885 $self->_error (&Errno::EBADMSG);
881 $self->error;
882 } 886 }
883 }); 887 });
884 }); 888 });
885 889
886 1 890 1
… …
943 return 1; 947 return 1;
944 } 948 }
945 949
946 # reject 950 # reject
947 if ($reject && $$rbuf =~ $reject) { 951 if ($reject && $$rbuf =~ $reject) {
948 $! = &Errno::EBADMSG; 952 $self->_error (&Errno::EBADMSG);
949 $self->error;
950 } 953 }
951 954
952 # skip 955 # skip
953 if ($skip && $$rbuf =~ $skip) { 956 if ($skip && $$rbuf =~ $skip) {
954 $data .= substr $$rbuf, 0, $+[0], ""; 957 $data .= substr $$rbuf, 0, $+[0], "";
… …
1027=item $handle->stop_read 1030=item $handle->stop_read
1028 1031
1029=item $handle->start_read 1032=item $handle->start_read
1030 1033
1031In rare cases you actually do not want to read anything from the 1034In 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 1035socket. 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 1036any queued callbacks will be executed then. To start reading again, call
1034C<start_read>. 1037C<start_read>.
1038
1039Note that AnyEvent::Handle will automatically C<start_read> for you when
1040you change the C<on_read> callback or push/unshift a read callback, and it
1041will automatically C<stop_read> for you when neither C<on_read> is set nor
1042there are any read requests in the queue.
1035 1043
1036=cut 1044=cut
1037 1045
1038sub stop_read { 1046sub stop_read {
1039 my ($self) = @_; 1047 my ($self) = @_;
… …
1062 delete $self->{_rw}; 1070 delete $self->{_rw};
1063 $self->{_eof} = 1; 1071 $self->{_eof} = 1;
1064 $self->_drain_rbuf; 1072 $self->_drain_rbuf;
1065 1073
1066 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1074 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1067 return $self->error; 1075 return $self->_error ($!, 1);
1068 } 1076 }
1069 }); 1077 });
1070 } 1078 }
1071} 1079}
1072 1080
1073sub _dotls { 1081sub _dotls {
1074 my ($self) = @_; 1082 my ($self) = @_;
1083
1084 my $buf;
1075 1085
1076 if (length $self->{_tls_wbuf}) { 1086 if (length $self->{_tls_wbuf}) {
1077 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1087 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1078 substr $self->{_tls_wbuf}, 0, $len, ""; 1088 substr $self->{_tls_wbuf}, 0, $len, "";
1079 } 1089 }
1080 } 1090 }
1081 1091
1082 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1092 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1083 $self->{wbuf} .= $buf; 1093 $self->{wbuf} .= $buf;
1084 $self->_drain_wbuf; 1094 $self->_drain_wbuf;
1085 } 1095 }
1086 1096
1087 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1097 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1098 if (length $buf) {
1088 $self->{rbuf} .= $buf; 1099 $self->{rbuf} .= $buf;
1089 $self->_drain_rbuf; 1100 $self->_drain_rbuf;
1101 } else {
1102 # let's treat SSL-eof as we treat normal EOF
1103 $self->{_eof} = 1;
1104 $self->_shutdown;
1105 return;
1106 }
1090 } 1107 }
1091 1108
1092 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1109 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1093 1110
1094 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1111 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1095 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1112 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1096 $self->error; 1113 return $self->_error ($!, 1);
1097 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1114 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1098 $! = &Errno::EIO; 1115 return $self->_error (&Errno::EIO, 1);
1099 $self->error;
1100 } 1116 }
1101 1117
1102 # all others are fine for our purposes 1118 # all others are fine for our purposes
1103 } 1119 }
1104} 1120}
… …
1119call and can be used or changed to your liking. Note that the handshake 1135call and can be used or changed to your liking. Note that the handshake
1120might have already started when this function returns. 1136might have already started when this function returns.
1121 1137
1122=cut 1138=cut
1123 1139
1124# TODO: maybe document...
1125sub starttls { 1140sub starttls {
1126 my ($self, $ssl, $ctx) = @_; 1141 my ($self, $ssl, $ctx) = @_;
1127 1142
1128 $self->stoptls; 1143 $self->stoptls;
1129 1144

Diff Legend

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