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.47 by root, Thu May 29 00:25:28 2008 UTC vs.
Revision 1.55 by root, Tue Jun 3 16:15:30 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.
… …
242 244
243 delete $self->{_tw}; 245 delete $self->{_tw};
244 delete $self->{_rw}; 246 delete $self->{_rw};
245 delete $self->{_ww}; 247 delete $self->{_ww};
246 delete $self->{fh}; 248 delete $self->{fh};
247}
248 249
250 $self->stoptls;
251}
252
249sub error { 253sub _error {
250 my ($self) = @_; 254 my ($self, $errno, $fatal) = @_;
251 255
252 {
253 local $!;
254 $self->_shutdown; 256 $self->_shutdown
255 } 257 if $fatal;
256 258
257 $self->{on_error}($self) 259 $! = $errno;
260
258 if $self->{on_error}; 261 if ($self->{on_error}) {
259 262 $self->{on_error}($self, $fatal);
263 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 264 Carp::croak "AnyEvent::Handle uncaught error: $!";
265 }
261} 266}
262 267
263=item $fh = $handle->fh 268=item $fh = $handle->fh
264 269
265This method returns the file handle of the L<AnyEvent::Handle> object. 270This method returns the file handle of the L<AnyEvent::Handle> object.
… …
329 # now or in the past already? 334 # now or in the past already?
330 if ($after <= 0) { 335 if ($after <= 0) {
331 $self->{_activity} = $NOW; 336 $self->{_activity} = $NOW;
332 337
333 if ($self->{on_timeout}) { 338 if ($self->{on_timeout}) {
334 $self->{on_timeout}->($self); 339 $self->{on_timeout}($self);
335 } else { 340 } else {
336 $! = Errno::ETIMEDOUT; 341 $self->_error (&Errno::ETIMEDOUT);
337 $self->error;
338 } 342 }
339 343
340 # callbakx could have changed timeout value, optimise 344 # callbakx could have changed timeout value, optimise
341 return unless $self->{timeout}; 345 return unless $self->{timeout};
342 346
… …
415 if $self->{low_water_mark} >= length $self->{wbuf} 419 if $self->{low_water_mark} >= length $self->{wbuf}
416 && $self->{on_drain}; 420 && $self->{on_drain};
417 421
418 delete $self->{_ww} unless length $self->{wbuf}; 422 delete $self->{_ww} unless length $self->{wbuf};
419 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 423 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
420 $self->error; 424 $self->_error ($!, 1);
421 } 425 }
422 }; 426 };
423 427
424 # try to write data immediately 428 # try to write data immediately
425 $cb->(); 429 $cb->();
… …
445 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 449 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
446 ->($self, @_); 450 ->($self, @_);
447 } 451 }
448 452
449 if ($self->{filter_w}) { 453 if ($self->{filter_w}) {
450 $self->{filter_w}->($self, \$_[0]); 454 $self->{filter_w}($self, \$_[0]);
451 } else { 455 } else {
452 $self->{wbuf} .= $_[0]; 456 $self->{wbuf} .= $_[0];
453 $self->_drain_wbuf; 457 $self->_drain_wbuf;
454 } 458 }
455} 459}
456 460
457=item $handle->push_write (type => @args) 461=item $handle->push_write (type => @args)
458 462
459=item $handle->unshift_write (type => @args)
460
461Instead of formatting your data yourself, you can also let this module do 463Instead of formatting your data yourself, you can also let this module do
462the job by specifying a type and type-specific arguments. 464the job by specifying a type and type-specific arguments.
463 465
464Predefined types are (if you have ideas for additional types, feel free to 466Predefined types are (if you have ideas for additional types, feel free to
465drop by and tell us): 467drop by and tell us):
… …
468 470
469=item netstring => $string 471=item netstring => $string
470 472
471Formats the given value as netstring 473Formats the given value as netstring
472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 474(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
473
474=back
475 475
476=cut 476=cut
477 477
478register_write_type netstring => sub { 478register_write_type netstring => sub {
479 my ($self, $string) = @_; 479 my ($self, $string) = @_;
… …
520 520
521 $self->{json} ? $self->{json}->encode ($ref) 521 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref) 522 : JSON::encode_json ($ref)
523}; 523};
524 524
525=back
526
525=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 527=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
526 528
527This function (not method) lets you add your own types to C<push_write>. 529This 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 530Whenever the given C<type> is used, C<push_write> will invoke the code
529reference with the handle object and the remaining arguments. 531reference with the handle object and the remaining arguments.
… …
566the specified number of bytes which give an XML datagram. 568the specified number of bytes which give an XML datagram.
567 569
568 # in the default state, expect some header bytes 570 # in the default state, expect some header bytes
569 $handle->on_read (sub { 571 $handle->on_read (sub {
570 # some data is here, now queue the length-header-read (4 octets) 572 # some data is here, now queue the length-header-read (4 octets)
571 shift->unshift_read_chunk (4, sub { 573 shift->unshift_read (chunk => 4, sub {
572 # header arrived, decode 574 # header arrived, decode
573 my $len = unpack "N", $_[1]; 575 my $len = unpack "N", $_[1];
574 576
575 # now read the payload 577 # now read the payload
576 shift->unshift_read_chunk ($len, sub { 578 shift->unshift_read (chunk => $len, sub {
577 my $xml = $_[1]; 579 my $xml = $_[1];
578 # handle xml 580 # handle xml
579 }); 581 });
580 }); 582 });
581 }); 583 });
… …
588 590
589 # request one 591 # request one
590 $handle->push_write ("request 1\015\012"); 592 $handle->push_write ("request 1\015\012");
591 593
592 # we expect "ERROR" or "OK" as response, so push a line read 594 # we expect "ERROR" or "OK" as response, so push a line read
593 $handle->push_read_line (sub { 595 $handle->push_read (line => sub {
594 # if we got an "OK", we have to _prepend_ another line, 596 # 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 597 # 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 598 # which are already in the queue when this callback is called
597 # we don't do this in case we got an error 599 # we don't do this in case we got an error
598 if ($_[1] eq "OK") { 600 if ($_[1] eq "OK") {
599 $_[0]->unshift_read_line (sub { 601 $_[0]->unshift_read (line => sub {
600 my $response = $_[1]; 602 my $response = $_[1];
601 ... 603 ...
602 }); 604 });
603 } 605 }
604 }); 606 });
605 607
606 # request two 608 # request two
607 $handle->push_write ("request 2\015\012"); 609 $handle->push_write ("request 2\015\012");
608 610
609 # simply read 64 bytes, always 611 # simply read 64 bytes, always
610 $handle->push_read_chunk (64, sub { 612 $handle->push_read (chunk => 64, sub {
611 my $response = $_[1]; 613 my $response = $_[1];
612 ... 614 ...
613 }); 615 });
614 616
615=over 4 617=over 4
… …
621 623
622 if ( 624 if (
623 defined $self->{rbuf_max} 625 defined $self->{rbuf_max}
624 && $self->{rbuf_max} < length $self->{rbuf} 626 && $self->{rbuf_max} < length $self->{rbuf}
625 ) { 627 ) {
626 $! = &Errno::ENOSPC; 628 return $self->_error (&Errno::ENOSPC, 1);
627 $self->error;
628 } 629 }
629 630
630 return if $self->{in_drain}; 631 return if $self->{in_drain};
631 local $self->{in_drain} = 1; 632 local $self->{in_drain} = 1;
632 633
… …
634 no strict 'refs'; 635 no strict 'refs';
635 if (my $cb = shift @{ $self->{_queue} }) { 636 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 637 unless ($cb->($self)) {
637 if ($self->{_eof}) { 638 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming) 639 # no progress can be made (not enough data and no data forthcoming)
639 $! = &Errno::EPIPE; 640 return $self->_error (&Errno::EPIPE, 1);
640 $self->error;
641 } 641 }
642 642
643 unshift @{ $self->{_queue} }, $cb; 643 unshift @{ $self->{_queue} }, $cb;
644 return; 644 last;
645 } 645 }
646 } elsif ($self->{on_read}) { 646 } elsif ($self->{on_read}) {
647 $self->{on_read}($self); 647 $self->{on_read}($self);
648 648
649 if ( 649 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed 650 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 651 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data 652 && $self->{on_read} # but we still have on_read
654 ) { 653 ) {
654 # no further data will arrive
655 # then no progress can be made 655 # so no progress can be made
656 $! = &Errno::EPIPE; 656 return $self->_error (&Errno::EPIPE, 1)
657 $self->error; 657 if $self->{_eof};
658
659 last; # more data might arrive
658 } 660 }
659 } else { 661 } else {
660 # read side becomes idle 662 # read side becomes idle
661 delete $self->{_rw}; 663 delete $self->{_rw};
662 return; 664 last;
663 } 665 }
664 } 666 }
665 667
666 if ($self->{_eof}) {
667 $self->_shutdown;
668 $self->{on_eof}($self) 668 $self->{on_eof}($self)
669 if $self->{on_eof}; 669 if $self->{_eof} && $self->{on_eof};
670
671 # may need to restart read watcher
672 unless ($self->{_rw}) {
673 $self->start_read
674 if $self->{on_read} || @{ $self->{_queue} };
670 } 675 }
671} 676}
672 677
673=item $handle->on_read ($cb) 678=item $handle->on_read ($cb)
674 679
… …
864 my ($self, $cb) = @_; 869 my ($self, $cb) = @_;
865 870
866 sub { 871 sub {
867 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 872 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
868 if ($_[0]{rbuf} =~ /[^0-9]/) { 873 if ($_[0]{rbuf} =~ /[^0-9]/) {
869 $! = &Errno::EBADMSG; 874 $self->_error (&Errno::EBADMSG);
870 $self->error;
871 } 875 }
872 return; 876 return;
873 } 877 }
874 878
875 my $len = $1; 879 my $len = $1;
… …
878 my $string = $_[1]; 882 my $string = $_[1];
879 $_[0]->unshift_read (chunk => 1, sub { 883 $_[0]->unshift_read (chunk => 1, sub {
880 if ($_[1] eq ",") { 884 if ($_[1] eq ",") {
881 $cb->($_[0], $string); 885 $cb->($_[0], $string);
882 } else { 886 } else {
883 $! = &Errno::EBADMSG; 887 $self->_error (&Errno::EBADMSG);
884 $self->error;
885 } 888 }
886 }); 889 });
887 }); 890 });
888 891
889 1 892 1
… …
946 return 1; 949 return 1;
947 } 950 }
948 951
949 # reject 952 # reject
950 if ($reject && $$rbuf =~ $reject) { 953 if ($reject && $$rbuf =~ $reject) {
951 $! = &Errno::EBADMSG; 954 $self->_error (&Errno::EBADMSG);
952 $self->error;
953 } 955 }
954 956
955 # skip 957 # skip
956 if ($skip && $$rbuf =~ $skip) { 958 if ($skip && $$rbuf =~ $skip) {
957 $data .= substr $$rbuf, 0, $+[0], ""; 959 $data .= substr $$rbuf, 0, $+[0], "";
… …
1056 1058
1057 if ($len > 0) { 1059 if ($len > 0) {
1058 $self->{_activity} = AnyEvent->now; 1060 $self->{_activity} = AnyEvent->now;
1059 1061
1060 $self->{filter_r} 1062 $self->{filter_r}
1061 ? $self->{filter_r}->($self, $rbuf) 1063 ? $self->{filter_r}($self, $rbuf)
1062 : $self->_drain_rbuf; 1064 : $self->_drain_rbuf;
1063 1065
1064 } elsif (defined $len) { 1066 } elsif (defined $len) {
1065 delete $self->{_rw}; 1067 delete $self->{_rw};
1066 $self->{_eof} = 1; 1068 $self->{_eof} = 1;
1067 $self->_drain_rbuf; 1069 $self->_drain_rbuf;
1068 1070
1069 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1071 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1070 return $self->error; 1072 return $self->_error ($!, 1);
1071 } 1073 }
1072 }); 1074 });
1073 } 1075 }
1074} 1076}
1075 1077
… …
1094 1096
1095 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1097 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1096 1098
1097 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1099 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1098 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1100 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1099 $self->error; 1101 return $self->_error ($!, 1);
1100 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1102 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1101 $! = &Errno::EIO; 1103 return $self->_error (&Errno::EIO, 1);
1102 $self->error;
1103 } 1104 }
1104 1105
1105 # all others are fine for our purposes 1106 # all others are fine for our purposes
1106 } 1107 }
1107} 1108}
… …
1122call and can be used or changed to your liking. Note that the handshake 1123call and can be used or changed to your liking. Note that the handshake
1123might have already started when this function returns. 1124might have already started when this function returns.
1124 1125
1125=cut 1126=cut
1126 1127
1127# TODO: maybe document...
1128sub starttls { 1128sub starttls {
1129 my ($self, $ssl, $ctx) = @_; 1129 my ($self, $ssl, $ctx) = @_;
1130 1130
1131 $self->stoptls; 1131 $self->stoptls;
1132 1132

Diff Legend

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