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.169 by root, Fri Jul 31 07:57:43 2009 UTC vs.
Revision 1.196 by root, Tue Jun 8 10:04:17 2010 UTC

1package AnyEvent::Handle;
2
3use Scalar::Util ();
4use Carp ();
5use Errno qw(EAGAIN EINTR);
6
7use AnyEvent (); BEGIN { AnyEvent::common_sense }
8use AnyEvent::Util qw(WSAEWOULDBLOCK);
9
10=head1 NAME 1=head1 NAME
11 2
12AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 3AnyEvent::Handle - non-blocking I/O on streaming handles via AnyEvent
13
14=cut
15
16our $VERSION = 4.881;
17 4
18=head1 SYNOPSIS 5=head1 SYNOPSIS
19 6
20 use AnyEvent; 7 use AnyEvent;
21 use AnyEvent::Handle; 8 use AnyEvent::Handle;
… …
27 on_error => sub { 14 on_error => sub {
28 my ($hdl, $fatal, $msg) = @_; 15 my ($hdl, $fatal, $msg) = @_;
29 warn "got error $msg\n"; 16 warn "got error $msg\n";
30 $hdl->destroy; 17 $hdl->destroy;
31 $cv->send; 18 $cv->send;
32 ); 19 };
33 20
34 # send some request line 21 # send some request line
35 $hdl->push_write ("getinfo\015\012"); 22 $hdl->push_write ("getinfo\015\012");
36 23
37 # read the response line 24 # read the response line
… …
44 $cv->recv; 31 $cv->recv;
45 32
46=head1 DESCRIPTION 33=head1 DESCRIPTION
47 34
48This module is a helper module to make it easier to do event-based I/O on 35This module is a helper module to make it easier to do event-based I/O on
49filehandles. 36stream-based filehandles (sockets, pipes or other stream things).
50 37
51The L<AnyEvent::Intro> tutorial contains some well-documented 38The L<AnyEvent::Intro> tutorial contains some well-documented
52AnyEvent::Handle examples. 39AnyEvent::Handle examples.
53 40
54In the following, when the documentation refers to of "bytes" then this 41In the following, when the documentation refers to of "bytes" then this
… …
59C<on_error> callback. 46C<on_error> callback.
60 47
61All callbacks will be invoked with the handle object as their first 48All callbacks will be invoked with the handle object as their first
62argument. 49argument.
63 50
51=cut
52
53package AnyEvent::Handle;
54
55use Scalar::Util ();
56use List::Util ();
57use Carp ();
58use Errno qw(EAGAIN EINTR);
59
60use AnyEvent (); BEGIN { AnyEvent::common_sense }
61use AnyEvent::Util qw(WSAEWOULDBLOCK);
62
63our $VERSION = $AnyEvent::VERSION;
64
65sub _load_func($) {
66 my $func = $_[0];
67
68 unless (defined &$func) {
69 my $pkg = $func;
70 do {
71 $pkg =~ s/::[^:]+$//
72 or return;
73 eval "require $pkg";
74 } until defined &$func;
75 }
76
77 \&$func
78}
79
64=head1 METHODS 80=head1 METHODS
65 81
66=over 4 82=over 4
67 83
68=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value... 84=item $handle = B<new> AnyEvent::Handle fh => $filehandle, key => value...
69 85
70The constructor supports these arguments (all as C<< key => value >> pairs). 86The constructor supports these arguments (all as C<< key => value >> pairs).
71 87
72=over 4 88=over 4
73 89
… …
113 129
114The actual numeric host and port (the socket peername) are passed as 130The actual numeric host and port (the socket peername) are passed as
115parameters, together with a retry callback. 131parameters, together with a retry callback.
116 132
117When, for some reason, the handle is not acceptable, then calling 133When, for some reason, the handle is not acceptable, then calling
118C<$retry> will continue with the next conenction target (in case of 134C<$retry> will continue with the next connection target (in case of
119multi-homed hosts or SRV records there can be multiple connection 135multi-homed hosts or SRV records there can be multiple connection
120endpoints). When it is called then the read and write queues, eof status, 136endpoints). At the time it is called the read and write queues, eof
121tls status and similar properties of the handle are being reset. 137status, tls status and similar properties of the handle will have been
138reset.
122 139
123In most cases, ignoring the C<$retry> parameter is the way to go. 140In most cases, ignoring the C<$retry> parameter is the way to go.
124 141
125=item on_connect_error => $cb->($handle, $message) 142=item on_connect_error => $cb->($handle, $message)
126 143
127This callback is called when the conenction could not be 144This callback is called when the connection could not be
128established. C<$!> will contain the relevant error code, and C<$message> a 145established. C<$!> will contain the relevant error code, and C<$message> a
129message describing it (usually the same as C<"$!">). 146message describing it (usually the same as C<"$!">).
130 147
131If this callback isn't specified, then C<on_error> will be called with a 148If this callback isn't specified, then C<on_error> will be called with a
132fatal error instead. 149fatal error instead.
… …
216memory and push it into the queue, but instead only read more data from 233memory and push it into the queue, but instead only read more data from
217the file when the write queue becomes empty. 234the file when the write queue becomes empty.
218 235
219=item timeout => $fractional_seconds 236=item timeout => $fractional_seconds
220 237
238=item rtimeout => $fractional_seconds
239
240=item wtimeout => $fractional_seconds
241
221If non-zero, then this enables an "inactivity" timeout: whenever this many 242If non-zero, then these enables an "inactivity" timeout: whenever this
222seconds pass without a successful read or write on the underlying file 243many seconds pass without a successful read or write on the underlying
223handle, the C<on_timeout> callback will be invoked (and if that one is 244file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
224missing, a non-fatal C<ETIMEDOUT> error will be raised). 245will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
246error will be raised).
247
248There are three variants of the timeouts that work fully independent
249of each other, for both read and write, just read, and just write:
250C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
251C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
252C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
225 253
226Note that timeout processing is also active when you currently do not have 254Note that timeout processing is also active when you currently do not have
227any outstanding read or write requests: If you plan to keep the connection 255any outstanding read or write requests: If you plan to keep the connection
228idle then you should disable the timout temporarily or ignore the timeout 256idle then you should disable the timout temporarily or ignore the timeout
229in the C<on_timeout> callback, in which case AnyEvent::Handle will simply 257in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
… …
273accomplishd by setting this option to a true value. 301accomplishd by setting this option to a true value.
274 302
275The default is your opertaing system's default behaviour (most likely 303The default is your opertaing system's default behaviour (most likely
276enabled), this option explicitly enables or disables it, if possible. 304enabled), this option explicitly enables or disables it, if possible.
277 305
306=item keepalive => <boolean>
307
308Enables (default disable) the SO_KEEPALIVE option on the stream socket:
309normally, TCP connections have no time-out once established, so TCP
310connections, once established, can stay alive forever even when the other
311side has long gone. TCP keepalives are a cheap way to take down long-lived
312TCP connections whent he other side becomes unreachable. While the default
313is OS-dependent, TCP keepalives usually kick in after around two hours,
314and, if the other side doesn't reply, take down the TCP connection some 10
315to 15 minutes later.
316
317It is harmless to specify this option for file handles that do not support
318keepalives, and enabling it on connections that are potentially long-lived
319is usually a good idea.
320
321=item oobinline => <boolean>
322
323BSD majorly fucked up the implementation of TCP urgent data. The result
324is that almost no OS implements TCP according to the specs, and every OS
325implements it slightly differently.
326
327If you want to handle TCP urgent data, then setting this flag (the default
328is enabled) gives you the most portable way of getting urgent data, by
329putting it into the stream.
330
331Since BSD emulation of OOB data on top of TCP's urgent data can have
332security implications, AnyEvent::Handle sets this flag automatically
333unless explicitly specified. Note that setting this flag after
334establishing a connection I<may> be a bit too late (data loss could
335already have occured on BSD systems), but at least it will protect you
336from most attacks.
337
278=item read_size => <bytes> 338=item read_size => <bytes>
279 339
280The default read block size (the amount of bytes this module will 340The default read block size (the amount of bytes this module will
281try to read during each loop iteration, which affects memory 341try to read during each loop iteration, which affects memory
282requirements). Default: C<8192>. 342requirements). Default: C<8192>.
… …
315C<undef>. 375C<undef>.
316 376
317=item tls => "accept" | "connect" | Net::SSLeay::SSL object 377=item tls => "accept" | "connect" | Net::SSLeay::SSL object
318 378
319When this parameter is given, it enables TLS (SSL) mode, that means 379When this parameter is given, it enables TLS (SSL) mode, that means
320AnyEvent will start a TLS handshake as soon as the conenction has been 380AnyEvent will start a TLS handshake as soon as the connection has been
321established and will transparently encrypt/decrypt data afterwards. 381established and will transparently encrypt/decrypt data afterwards.
322 382
323All TLS protocol errors will be signalled as C<EPROTO>, with an 383All TLS protocol errors will be signalled as C<EPROTO>, with an
324appropriate error message. 384appropriate error message.
325 385
… …
438 delete $self->{_skip_drain_rbuf}; 498 delete $self->{_skip_drain_rbuf};
439 $self->_start; 499 $self->_start;
440 500
441 $self->{on_connect} 501 $self->{on_connect}
442 and $self->{on_connect}($self, $host, $port, sub { 502 and $self->{on_connect}($self, $host, $port, sub {
443 delete @$self{qw(fh _tw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)}; 503 delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
444 $self->{_skip_drain_rbuf} = 1; 504 $self->{_skip_drain_rbuf} = 1;
445 &$retry; 505 &$retry;
446 }); 506 });
447 507
448 } else { 508 } else {
… …
472} 532}
473 533
474sub _start { 534sub _start {
475 my ($self) = @_; 535 my ($self) = @_;
476 536
537 # too many clueless people try to use udp and similar sockets
538 # with AnyEvent::Handle, do them a favour.
539 my $type = getsockopt $self->{fh}, Socket::SOL_SOCKET (), Socket::SO_TYPE ();
540 Carp::croak "AnyEvent::Handle: only stream sockets supported, anything else will NOT work!"
541 if Socket::SOCK_STREAM () != (unpack "I", $type) && defined $type;
542
477 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 543 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
478 544
545 $self->{_activity} =
546 $self->{_ractivity} =
479 $self->{_activity} = AnyEvent->now; 547 $self->{_wactivity} = AE::now;
480 $self->_timeout;
481 548
549 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
550 $self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout};
551 $self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout};
552
482 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay}; 553 $self->no_delay (delete $self->{no_delay} ) if exists $self->{no_delay} && $self->{no_delay};
554 $self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive} && $self->{keepalive};
483 555
556 $self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
557
484 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 558 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
485 if $self->{tls}; 559 if $self->{tls};
486 560
487 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 561 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
488 562
489 $self->start_read 563 $self->start_read
490 if $self->{on_read} || @{ $self->{_queue} }; 564 if $self->{on_read} || @{ $self->{_queue} };
491 565
492 $self->_drain_wbuf; 566 $self->_drain_wbuf;
493} 567}
494
495#sub _shutdown {
496# my ($self) = @_;
497#
498# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
499# $self->{_eof} = 1; # tell starttls et. al to stop trying
500#
501# &_freetls;
502#}
503 568
504sub _error { 569sub _error {
505 my ($self, $errno, $fatal, $message) = @_; 570 my ($self, $errno, $fatal, $message) = @_;
506 571
507 $! = $errno; 572 $! = $errno;
508 $message ||= "$!"; 573 $message ||= "$!";
509 574
510 if ($self->{on_error}) { 575 if ($self->{on_error}) {
511 $self->{on_error}($self, $fatal, $message); 576 $self->{on_error}($self, $fatal, $message);
512 $self->destroy if $fatal; 577 $self->destroy if $fatal;
513 } elsif ($self->{fh}) { 578 } elsif ($self->{fh} || $self->{connect}) {
514 $self->destroy; 579 $self->destroy;
515 Carp::croak "AnyEvent::Handle uncaught error: $message"; 580 Carp::croak "AnyEvent::Handle uncaught error: $message";
516 } 581 }
517} 582}
518 583
… …
544 $_[0]{on_eof} = $_[1]; 609 $_[0]{on_eof} = $_[1];
545} 610}
546 611
547=item $handle->on_timeout ($cb) 612=item $handle->on_timeout ($cb)
548 613
549Replace the current C<on_timeout> callback, or disables the callback (but 614=item $handle->on_rtimeout ($cb)
550not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551argument and method.
552 615
553=cut 616=item $handle->on_wtimeout ($cb)
554 617
555sub on_timeout { 618Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556 $_[0]{on_timeout} = $_[1]; 619callback, or disables the callback (but not the timeout) if C<$cb> =
557} 620C<undef>. See the C<timeout> constructor argument and method.
621
622=cut
623
624# see below
558 625
559=item $handle->autocork ($boolean) 626=item $handle->autocork ($boolean)
560 627
561Enables or disables the current autocork behaviour (see C<autocork> 628Enables or disables the current autocork behaviour (see C<autocork>
562constructor argument). Changes will only take effect on the next write. 629constructor argument). Changes will only take effect on the next write.
… …
577sub no_delay { 644sub no_delay {
578 $_[0]{no_delay} = $_[1]; 645 $_[0]{no_delay} = $_[1];
579 646
580 eval { 647 eval {
581 local $SIG{__DIE__}; 648 local $SIG{__DIE__};
582 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1] 649 setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
583 if $_[0]{fh}; 650 if $_[0]{fh};
584 }; 651 };
585} 652}
586 653
654=item $handle->keepalive ($boolean)
655
656Enables or disables the C<keepalive> setting (see constructor argument of
657the same name for details).
658
659=cut
660
661sub keepalive {
662 $_[0]{keepalive} = $_[1];
663
664 eval {
665 local $SIG{__DIE__};
666 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
667 if $_[0]{fh};
668 };
669}
670
671=item $handle->oobinline ($boolean)
672
673Enables or disables the C<oobinline> setting (see constructor argument of
674the same name for details).
675
676=cut
677
678sub oobinline {
679 $_[0]{oobinline} = $_[1];
680
681 eval {
682 local $SIG{__DIE__};
683 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
684 if $_[0]{fh};
685 };
686}
687
688=item $handle->keepalive ($boolean)
689
690Enables or disables the C<keepalive> setting (see constructor argument of
691the same name for details).
692
693=cut
694
695sub keepalive {
696 $_[0]{keepalive} = $_[1];
697
698 eval {
699 local $SIG{__DIE__};
700 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
701 if $_[0]{fh};
702 };
703}
704
587=item $handle->on_starttls ($cb) 705=item $handle->on_starttls ($cb)
588 706
589Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument). 707Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
590 708
591=cut 709=cut
… …
598 716
599Replace the current C<on_stoptls> callback (see the C<on_stoptls> constructor argument). 717Replace the current C<on_stoptls> callback (see the C<on_stoptls> constructor argument).
600 718
601=cut 719=cut
602 720
603sub on_starttls { 721sub on_stoptls {
604 $_[0]{on_stoptls} = $_[1]; 722 $_[0]{on_stoptls} = $_[1];
605} 723}
606 724
607=item $handle->rbuf_max ($max_octets) 725=item $handle->rbuf_max ($max_octets)
608 726
… …
616 734
617############################################################################# 735#############################################################################
618 736
619=item $handle->timeout ($seconds) 737=item $handle->timeout ($seconds)
620 738
739=item $handle->rtimeout ($seconds)
740
741=item $handle->wtimeout ($seconds)
742
621Configures (or disables) the inactivity timeout. 743Configures (or disables) the inactivity timeout.
622 744
623=cut 745=item $handle->timeout_reset
624 746
625sub timeout { 747=item $handle->rtimeout_reset
748
749=item $handle->wtimeout_reset
750
751Reset the activity timeout, as if data was received or sent.
752
753These methods are cheap to call.
754
755=cut
756
757for my $dir ("", "r", "w") {
758 my $timeout = "${dir}timeout";
759 my $tw = "_${dir}tw";
760 my $on_timeout = "on_${dir}timeout";
761 my $activity = "_${dir}activity";
762 my $cb;
763
764 *$on_timeout = sub {
765 $_[0]{$on_timeout} = $_[1];
766 };
767
768 *$timeout = sub {
626 my ($self, $timeout) = @_; 769 my ($self, $new_value) = @_;
627 770
628 $self->{timeout} = $timeout; 771 $self->{$timeout} = $new_value;
629 $self->_timeout; 772 delete $self->{$tw}; &$cb;
630} 773 };
631 774
775 *{"${dir}timeout_reset"} = sub {
776 $_[0]{$activity} = AE::now;
777 };
778
779 # main workhorse:
632# reset the timeout watcher, as neccessary 780 # reset the timeout watcher, as neccessary
633# also check for time-outs 781 # also check for time-outs
634sub _timeout { 782 $cb = sub {
635 my ($self) = @_; 783 my ($self) = @_;
636 784
637 if ($self->{timeout} && $self->{fh}) { 785 if ($self->{$timeout} && $self->{fh}) {
638 my $NOW = AnyEvent->now; 786 my $NOW = AE::now;
639 787
640 # when would the timeout trigger? 788 # when would the timeout trigger?
641 my $after = $self->{_activity} + $self->{timeout} - $NOW; 789 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
642 790
643 # now or in the past already? 791 # now or in the past already?
644 if ($after <= 0) { 792 if ($after <= 0) {
645 $self->{_activity} = $NOW; 793 $self->{$activity} = $NOW;
646 794
647 if ($self->{on_timeout}) { 795 if ($self->{$on_timeout}) {
648 $self->{on_timeout}($self); 796 $self->{$on_timeout}($self);
649 } else { 797 } else {
650 $self->_error (Errno::ETIMEDOUT); 798 $self->_error (Errno::ETIMEDOUT);
799 }
800
801 # callback could have changed timeout value, optimise
802 return unless $self->{$timeout};
803
804 # calculate new after
805 $after = $self->{$timeout};
651 } 806 }
652 807
653 # callback could have changed timeout value, optimise 808 Scalar::Util::weaken $self;
654 return unless $self->{timeout}; 809 return unless $self; # ->error could have destroyed $self
655 810
656 # calculate new after 811 $self->{$tw} ||= AE::timer $after, 0, sub {
657 $after = $self->{timeout}; 812 delete $self->{$tw};
813 $cb->($self);
814 };
815 } else {
816 delete $self->{$tw};
658 } 817 }
659
660 Scalar::Util::weaken $self;
661 return unless $self; # ->error could have destroyed $self
662
663 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
664 delete $self->{_tw};
665 $self->_timeout;
666 });
667 } else {
668 delete $self->{_tw};
669 } 818 }
670} 819}
671 820
672############################################################################# 821#############################################################################
673 822
… …
688 837
689=item $handle->on_drain ($cb) 838=item $handle->on_drain ($cb)
690 839
691Sets the C<on_drain> callback or clears it (see the description of 840Sets the C<on_drain> callback or clears it (see the description of
692C<on_drain> in the constructor). 841C<on_drain> in the constructor).
842
843This method may invoke callbacks (and therefore the handle might be
844destroyed after it returns).
693 845
694=cut 846=cut
695 847
696sub on_drain { 848sub on_drain {
697 my ($self, $cb) = @_; 849 my ($self, $cb) = @_;
… …
706 858
707Queues the given scalar to be written. You can push as much data as you 859Queues the given scalar to be written. You can push as much data as you
708want (only limited by the available memory), as C<AnyEvent::Handle> 860want (only limited by the available memory), as C<AnyEvent::Handle>
709buffers it independently of the kernel. 861buffers it independently of the kernel.
710 862
863This method may invoke callbacks (and therefore the handle might be
864destroyed after it returns).
865
711=cut 866=cut
712 867
713sub _drain_wbuf { 868sub _drain_wbuf {
714 my ($self) = @_; 869 my ($self) = @_;
715 870
… …
721 my $len = syswrite $self->{fh}, $self->{wbuf}; 876 my $len = syswrite $self->{fh}, $self->{wbuf};
722 877
723 if (defined $len) { 878 if (defined $len) {
724 substr $self->{wbuf}, 0, $len, ""; 879 substr $self->{wbuf}, 0, $len, "";
725 880
726 $self->{_activity} = AnyEvent->now; 881 $self->{_activity} = $self->{_wactivity} = AE::now;
727 882
728 $self->{on_drain}($self) 883 $self->{on_drain}($self)
729 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 884 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
730 && $self->{on_drain}; 885 && $self->{on_drain};
731 886
… …
737 892
738 # try to write data immediately 893 # try to write data immediately
739 $cb->() unless $self->{autocork}; 894 $cb->() unless $self->{autocork};
740 895
741 # if still data left in wbuf, we need to poll 896 # if still data left in wbuf, we need to poll
742 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 897 $self->{_ww} = AE::io $self->{fh}, 1, $cb
743 if length $self->{wbuf}; 898 if length $self->{wbuf};
744 }; 899 };
745} 900}
746 901
747our %WH; 902our %WH;
748 903
904# deprecated
749sub register_write_type($$) { 905sub register_write_type($$) {
750 $WH{$_[0]} = $_[1]; 906 $WH{$_[0]} = $_[1];
751} 907}
752 908
753sub push_write { 909sub push_write {
754 my $self = shift; 910 my $self = shift;
755 911
756 if (@_ > 1) { 912 if (@_ > 1) {
757 my $type = shift; 913 my $type = shift;
758 914
915 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
759 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 916 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_write")
760 ->($self, @_); 917 ->($self, @_);
761 } 918 }
762 919
920 # we downgrade here to avoid hard-to-track-down bugs,
921 # and diagnose the problem earlier and better.
922
763 if ($self->{tls}) { 923 if ($self->{tls}) {
764 $self->{_tls_wbuf} .= $_[0]; 924 utf8::downgrade $self->{_tls_wbuf} .= $_[0];
765 &_dotls ($self) if $self->{fh}; 925 &_dotls ($self) if $self->{fh};
766 } else { 926 } else {
767 $self->{wbuf} .= $_[0]; 927 utf8::downgrade $self->{wbuf} .= $_[0];
768 $self->_drain_wbuf if $self->{fh}; 928 $self->_drain_wbuf if $self->{fh};
769 } 929 }
770} 930}
771 931
772=item $handle->push_write (type => @args) 932=item $handle->push_write (type => @args)
773 933
774Instead of formatting your data yourself, you can also let this module do 934Instead of formatting your data yourself, you can also let this module
775the job by specifying a type and type-specific arguments. 935do the job by specifying a type and type-specific arguments. You
936can also specify the (fully qualified) name of a package, in which
937case AnyEvent tries to load the package and then expects to find the
938C<anyevent_read_type> function inside (see "custom write types", below).
776 939
777Predefined types are (if you have ideas for additional types, feel free to 940Predefined types are (if you have ideas for additional types, feel free to
778drop by and tell us): 941drop by and tell us):
779 942
780=over 4 943=over 4
… …
837Other languages could read single lines terminated by a newline and pass 1000Other languages could read single lines terminated by a newline and pass
838this line into their JSON decoder of choice. 1001this line into their JSON decoder of choice.
839 1002
840=cut 1003=cut
841 1004
1005sub json_coder() {
1006 eval { require JSON::XS; JSON::XS->new->utf8 }
1007 || do { require JSON; JSON->new->utf8 }
1008}
1009
842register_write_type json => sub { 1010register_write_type json => sub {
843 my ($self, $ref) = @_; 1011 my ($self, $ref) = @_;
844 1012
845 require JSON; 1013 my $json = $self->{json} ||= json_coder;
846 1014
847 $self->{json} ? $self->{json}->encode ($ref) 1015 $json->encode ($ref)
848 : JSON::encode_json ($ref)
849}; 1016};
850 1017
851=item storable => $reference 1018=item storable => $reference
852 1019
853Freezes the given reference using L<Storable> and writes it to the 1020Freezes the given reference using L<Storable> and writes it to the
… …
879the peer. 1046the peer.
880 1047
881You can rely on the normal read queue and C<on_eof> handling 1048You can rely on the normal read queue and C<on_eof> handling
882afterwards. This is the cleanest way to close a connection. 1049afterwards. This is the cleanest way to close a connection.
883 1050
1051This method may invoke callbacks (and therefore the handle might be
1052destroyed after it returns).
1053
884=cut 1054=cut
885 1055
886sub push_shutdown { 1056sub push_shutdown {
887 my ($self) = @_; 1057 my ($self) = @_;
888 1058
889 delete $self->{low_water_mark}; 1059 delete $self->{low_water_mark};
890 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1060 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
891} 1061}
892 1062
893=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1063=item custom write types - Package::anyevent_write_type $handle, @args
894 1064
895This function (not method) lets you add your own types to C<push_write>. 1065Instead of one of the predefined types, you can also specify the name of
1066a package. AnyEvent will try to load the package and then expects to find
1067a function named C<anyevent_write_type> inside. If it isn't found, it
1068progressively tries to load the parent package until it either finds the
1069function (good) or runs out of packages (bad).
1070
896Whenever the given C<type> is used, C<push_write> will invoke the code 1071Whenever the given C<type> is used, C<push_write> will the function with
897reference with the handle object and the remaining arguments. 1072the handle object and the remaining arguments.
898 1073
899The code reference is supposed to return a single octet string that will 1074The function is supposed to return a single octet string that will be
900be appended to the write buffer. 1075appended to the write buffer, so you cna mentally treat this function as a
1076"arguments to on-the-wire-format" converter.
901 1077
902Note that this is a function, and all types registered this way will be 1078Example: implement a custom write type C<join> that joins the remaining
903global, so try to use unique names. 1079arguments using the first one.
1080
1081 $handle->push_write (My::Type => " ", 1,2,3);
1082
1083 # uses the following package, which can be defined in the "My::Type" or in
1084 # the "My" modules to be auto-loaded, or just about anywhere when the
1085 # My::Type::anyevent_write_type is defined before invoking it.
1086
1087 package My::Type;
1088
1089 sub anyevent_write_type {
1090 my ($handle, $delim, @args) = @_;
1091
1092 join $delim, @args
1093 }
904 1094
905=cut 1095=cut
906 1096
907############################################################################# 1097#############################################################################
908 1098
… …
1064 1254
1065This replaces the currently set C<on_read> callback, or clears it (when 1255This replaces the currently set C<on_read> callback, or clears it (when
1066the new callback is C<undef>). See the description of C<on_read> in the 1256the new callback is C<undef>). See the description of C<on_read> in the
1067constructor. 1257constructor.
1068 1258
1259This method may invoke callbacks (and therefore the handle might be
1260destroyed after it returns).
1261
1069=cut 1262=cut
1070 1263
1071sub on_read { 1264sub on_read {
1072 my ($self, $cb) = @_; 1265 my ($self, $cb) = @_;
1073 1266
… …
1112 1305
1113If enough data was available, then the callback must remove all data it is 1306If enough data was available, then the callback must remove all data it is
1114interested in (which can be none at all) and return a true value. After returning 1307interested in (which can be none at all) and return a true value. After returning
1115true, it will be removed from the queue. 1308true, it will be removed from the queue.
1116 1309
1310These methods may invoke callbacks (and therefore the handle might be
1311destroyed after it returns).
1312
1117=cut 1313=cut
1118 1314
1119our %RH; 1315our %RH;
1120 1316
1121sub register_read_type($$) { 1317sub register_read_type($$) {
… …
1127 my $cb = pop; 1323 my $cb = pop;
1128 1324
1129 if (@_) { 1325 if (@_) {
1130 my $type = shift; 1326 my $type = shift;
1131 1327
1328 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1132 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 1329 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
1133 ->($self, $cb, @_); 1330 ->($self, $cb, @_);
1134 } 1331 }
1135 1332
1136 push @{ $self->{_queue} }, $cb; 1333 push @{ $self->{_queue} }, $cb;
1137 $self->_drain_rbuf; 1334 $self->_drain_rbuf;
… …
1146 1343
1147 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1344 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1148 ->($self, $cb, @_); 1345 ->($self, $cb, @_);
1149 } 1346 }
1150 1347
1151
1152 unshift @{ $self->{_queue} }, $cb; 1348 unshift @{ $self->{_queue} }, $cb;
1153 $self->_drain_rbuf; 1349 $self->_drain_rbuf;
1154} 1350}
1155 1351
1156=item $handle->push_read (type => @args, $cb) 1352=item $handle->push_read (type => @args, $cb)
1157 1353
1158=item $handle->unshift_read (type => @args, $cb) 1354=item $handle->unshift_read (type => @args, $cb)
1159 1355
1160Instead of providing a callback that parses the data itself you can chose 1356Instead of providing a callback that parses the data itself you can chose
1161between a number of predefined parsing formats, for chunks of data, lines 1357between a number of predefined parsing formats, for chunks of data, lines
1162etc. 1358etc. You can also specify the (fully qualified) name of a package, in
1359which case AnyEvent tries to load the package and then expects to find the
1360C<anyevent_read_type> function inside (see "custom read types", below).
1163 1361
1164Predefined types are (if you have ideas for additional types, feel free to 1362Predefined types are (if you have ideas for additional types, feel free to
1165drop by and tell us): 1363drop by and tell us):
1166 1364
1167=over 4 1365=over 4
… …
1407=cut 1605=cut
1408 1606
1409register_read_type json => sub { 1607register_read_type json => sub {
1410 my ($self, $cb) = @_; 1608 my ($self, $cb) = @_;
1411 1609
1412 my $json = $self->{json} ||= 1610 my $json = $self->{json} ||= json_coder;
1413 eval { require JSON::XS; JSON::XS->new->utf8 }
1414 || do { require JSON; JSON->new->utf8 };
1415 1611
1416 my $data; 1612 my $data;
1417 my $rbuf = \$self->{rbuf}; 1613 my $rbuf = \$self->{rbuf};
1418 1614
1419 sub { 1615 sub {
… …
1488 } 1684 }
1489}; 1685};
1490 1686
1491=back 1687=back
1492 1688
1493=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1689=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1494 1690
1495This function (not method) lets you add your own types to C<push_read>. 1691Instead of one of the predefined types, you can also specify the name
1692of a package. AnyEvent will try to load the package and then expects to
1693find a function named C<anyevent_read_type> inside. If it isn't found, it
1694progressively tries to load the parent package until it either finds the
1695function (good) or runs out of packages (bad).
1496 1696
1497Whenever the given C<type> is used, C<push_read> will invoke the code 1697Whenever this type is used, C<push_read> will invoke the function with the
1498reference with the handle object, the callback and the remaining 1698handle object, the original callback and the remaining arguments.
1499arguments.
1500 1699
1501The code reference is supposed to return a callback (usually a closure) 1700The function is supposed to return a callback (usually a closure) that
1502that works as a plain read callback (see C<< ->push_read ($cb) >>). 1701works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
1702mentally treat the function as a "configurable read type to read callback"
1703converter.
1503 1704
1504It should invoke the passed callback when it is done reading (remember to 1705It should invoke the original callback when it is done reading (remember
1505pass C<$handle> as first argument as all other callbacks do that). 1706to pass C<$handle> as first argument as all other callbacks do that,
1707although there is no strict requirement on this).
1506 1708
1507Note that this is a function, and all types registered this way will be
1508global, so try to use unique names.
1509
1510For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1709For examples, see the source of this module (F<perldoc -m
1511search for C<register_read_type>)). 1710AnyEvent::Handle>, search for C<register_read_type>)).
1512 1711
1513=item $handle->stop_read 1712=item $handle->stop_read
1514 1713
1515=item $handle->start_read 1714=item $handle->start_read
1516 1715
… …
1536} 1735}
1537 1736
1538sub start_read { 1737sub start_read {
1539 my ($self) = @_; 1738 my ($self) = @_;
1540 1739
1541 unless ($self->{_rw} || $self->{_eof}) { 1740 unless ($self->{_rw} || $self->{_eof} || !$self->{fh}) {
1542 Scalar::Util::weaken $self; 1741 Scalar::Util::weaken $self;
1543 1742
1544 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1743 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1545 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1744 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1546 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1745 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1547 1746
1548 if ($len > 0) { 1747 if ($len > 0) {
1549 $self->{_activity} = AnyEvent->now; 1748 $self->{_activity} = $self->{_ractivity} = AE::now;
1550 1749
1551 if ($self->{tls}) { 1750 if ($self->{tls}) {
1552 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1751 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1553 1752
1554 &_dotls ($self); 1753 &_dotls ($self);
… …
1562 $self->_drain_rbuf; 1761 $self->_drain_rbuf;
1563 1762
1564 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1763 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1565 return $self->_error ($!, 1); 1764 return $self->_error ($!, 1);
1566 } 1765 }
1567 }); 1766 };
1568 } 1767 }
1569} 1768}
1570 1769
1571our $ERROR_SYSCALL; 1770our $ERROR_SYSCALL;
1572our $ERROR_WANT_READ; 1771our $ERROR_WANT_READ;
… …
1639 && ($tmp != $ERROR_SYSCALL || $!); 1838 && ($tmp != $ERROR_SYSCALL || $!);
1640 1839
1641 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1840 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1642 $self->{wbuf} .= $tmp; 1841 $self->{wbuf} .= $tmp;
1643 $self->_drain_wbuf; 1842 $self->_drain_wbuf;
1843 $self->{tls} or return; # tls session might have gone away in callback
1644 } 1844 }
1645 1845
1646 $self->{_on_starttls} 1846 $self->{_on_starttls}
1647 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK () 1847 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK ()
1648 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established"); 1848 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established");
… …
1673 1873
1674Due to bugs in OpenSSL, it might or might not be possible to do multiple 1874Due to bugs in OpenSSL, it might or might not be possible to do multiple
1675handshakes on the same stream. Best do not attempt to use the stream after 1875handshakes on the same stream. Best do not attempt to use the stream after
1676stopping TLS. 1876stopping TLS.
1677 1877
1878This method may invoke callbacks (and therefore the handle might be
1879destroyed after it returns).
1880
1678=cut 1881=cut
1679 1882
1680our %TLS_CACHE; #TODO not yet documented, should we? 1883our %TLS_CACHE; #TODO not yet documented, should we?
1681 1884
1682sub starttls { 1885sub starttls {
… …
1693 require Net::SSLeay; 1896 require Net::SSLeay;
1694 1897
1695 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1898 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1696 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1899 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1697 1900
1698 $tls = $self->{tls}; 1901 $tls = delete $self->{tls};
1699 $ctx = $self->{tls_ctx}; 1902 $ctx = $self->{tls_ctx};
1700 1903
1701 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1904 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1702 1905
1703 if ("HASH" eq ref $ctx) { 1906 if ("HASH" eq ref $ctx) {
… …
1732 Net::SSLeay::CTX_set_mode ($tls, 1|2); 1935 Net::SSLeay::CTX_set_mode ($tls, 1|2);
1733 1936
1734 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1937 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1735 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1938 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1736 1939
1940 Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
1941
1737 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio}); 1942 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1738 1943
1739 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) } 1944 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1740 if $self->{on_starttls}; 1945 if $self->{on_starttls};
1741 1946
… …
1745 1950
1746=item $handle->stoptls 1951=item $handle->stoptls
1747 1952
1748Shuts down the SSL connection - this makes a proper EOF handshake by 1953Shuts down the SSL connection - this makes a proper EOF handshake by
1749sending a close notify to the other side, but since OpenSSL doesn't 1954sending a close notify to the other side, but since OpenSSL doesn't
1750support non-blocking shut downs, it is not guarenteed that you can re-use 1955support non-blocking shut downs, it is not guaranteed that you can re-use
1751the stream afterwards. 1956the stream afterwards.
1957
1958This method may invoke callbacks (and therefore the handle might be
1959destroyed after it returns).
1752 1960
1753=cut 1961=cut
1754 1962
1755sub stoptls { 1963sub stoptls {
1756 my ($self) = @_; 1964 my ($self) = @_;
1757 1965
1758 if ($self->{tls}) { 1966 if ($self->{tls} && $self->{fh}) {
1759 Net::SSLeay::shutdown ($self->{tls}); 1967 Net::SSLeay::shutdown ($self->{tls});
1760 1968
1761 &_dotls; 1969 &_dotls;
1762 1970
1763# # we don't give a shit. no, we do, but we can't. no...#d# 1971# # we don't give a shit. no, we do, but we can't. no...#d#
… …
1770 my ($self) = @_; 1978 my ($self) = @_;
1771 1979
1772 return unless $self->{tls}; 1980 return unless $self->{tls};
1773 1981
1774 $self->{tls_ctx}->_put_session (delete $self->{tls}) 1982 $self->{tls_ctx}->_put_session (delete $self->{tls})
1775 if ref $self->{tls}; 1983 if $self->{tls} > 0;
1776 1984
1777 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 1985 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1778} 1986}
1779 1987
1780sub DESTROY { 1988sub DESTROY {
… …
1788 my $fh = delete $self->{fh}; 1996 my $fh = delete $self->{fh};
1789 my $wbuf = delete $self->{wbuf}; 1997 my $wbuf = delete $self->{wbuf};
1790 1998
1791 my @linger; 1999 my @linger;
1792 2000
1793 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 2001 push @linger, AE::io $fh, 1, sub {
1794 my $len = syswrite $fh, $wbuf, length $wbuf; 2002 my $len = syswrite $fh, $wbuf, length $wbuf;
1795 2003
1796 if ($len > 0) { 2004 if ($len > 0) {
1797 substr $wbuf, 0, $len, ""; 2005 substr $wbuf, 0, $len, "";
1798 } else { 2006 } else {
1799 @linger = (); # end 2007 @linger = (); # end
1800 } 2008 }
1801 }); 2009 };
1802 push @linger, AnyEvent->timer (after => $linger, cb => sub { 2010 push @linger, AE::timer $linger, 0, sub {
1803 @linger = (); 2011 @linger = ();
1804 }); 2012 };
1805 } 2013 }
1806} 2014}
1807 2015
1808=item $handle->destroy 2016=item $handle->destroy
1809 2017
… …
1840 2048
1841sub AnyEvent::Handle::destroyed::AUTOLOAD { 2049sub AnyEvent::Handle::destroyed::AUTOLOAD {
1842 #nop 2050 #nop
1843} 2051}
1844 2052
2053=item $handle->destroyed
2054
2055Returns false as long as the handle hasn't been destroyed by a call to C<<
2056->destroy >>, true otherwise.
2057
2058Can be useful to decide whether the handle is still valid after some
2059callback possibly destroyed the handle. For example, C<< ->push_write >>,
2060C<< ->starttls >> and other methods can call user callbacks, which in turn
2061can destroy the handle, so work can be avoided by checking sometimes:
2062
2063 $hdl->starttls ("accept");
2064 return if $hdl->destroyed;
2065 $hdl->push_write (...
2066
2067Note that the call to C<push_write> will silently be ignored if the handle
2068has been destroyed, so often you can just ignore the possibility of the
2069handle being destroyed.
2070
2071=cut
2072
2073sub destroyed { 0 }
2074sub AnyEvent::Handle::destroyed::destroyed { 1 }
2075
1845=item AnyEvent::Handle::TLS_CTX 2076=item AnyEvent::Handle::TLS_CTX
1846 2077
1847This function creates and returns the AnyEvent::TLS object used by default 2078This function creates and returns the AnyEvent::TLS object used by default
1848for TLS mode. 2079for TLS mode.
1849 2080

Diff Legend

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