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.174 by root, Sat Aug 8 20:52:06 2009 UTC vs.
Revision 1.193 by root, Mon Mar 15 18:51:30 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 file handles via AnyEvent
13
14=cut
15
16our $VERSION = 4.91;
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
… …
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 {
… …
474sub _start { 534sub _start {
475 my ($self) = @_; 535 my ($self) = @_;
476 536
477 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 537 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
478 538
539 $self->{_activity} =
540 $self->{_ractivity} =
479 $self->{_activity} = AnyEvent->now; 541 $self->{_wactivity} = AE::now;
480 $self->_timeout;
481 542
543 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
544 $self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout};
545 $self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout};
546
482 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay}; 547 $self->no_delay (delete $self->{no_delay} ) if exists $self->{no_delay} && $self->{no_delay};
548 $self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive} && $self->{keepalive};
483 549
550 $self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
551
484 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 552 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
485 if $self->{tls}; 553 if $self->{tls};
486 554
487 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 555 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
488 556
489 $self->start_read 557 $self->start_read
490 if $self->{on_read} || @{ $self->{_queue} }; 558 if $self->{on_read} || @{ $self->{_queue} };
491 559
492 $self->_drain_wbuf; 560 $self->_drain_wbuf;
493} 561}
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 562
504sub _error { 563sub _error {
505 my ($self, $errno, $fatal, $message) = @_; 564 my ($self, $errno, $fatal, $message) = @_;
506 565
507 $! = $errno; 566 $! = $errno;
508 $message ||= "$!"; 567 $message ||= "$!";
509 568
510 if ($self->{on_error}) { 569 if ($self->{on_error}) {
511 $self->{on_error}($self, $fatal, $message); 570 $self->{on_error}($self, $fatal, $message);
512 $self->destroy if $fatal; 571 $self->destroy if $fatal;
513 } elsif ($self->{fh}) { 572 } elsif ($self->{fh} || $self->{connect}) {
514 $self->destroy; 573 $self->destroy;
515 Carp::croak "AnyEvent::Handle uncaught error: $message"; 574 Carp::croak "AnyEvent::Handle uncaught error: $message";
516 } 575 }
517} 576}
518 577
… …
544 $_[0]{on_eof} = $_[1]; 603 $_[0]{on_eof} = $_[1];
545} 604}
546 605
547=item $handle->on_timeout ($cb) 606=item $handle->on_timeout ($cb)
548 607
549Replace the current C<on_timeout> callback, or disables the callback (but 608=item $handle->on_rtimeout ($cb)
550not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551argument and method.
552 609
553=cut 610=item $handle->on_wtimeout ($cb)
554 611
555sub on_timeout { 612Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556 $_[0]{on_timeout} = $_[1]; 613callback, or disables the callback (but not the timeout) if C<$cb> =
557} 614C<undef>. See the C<timeout> constructor argument and method.
615
616=cut
617
618# see below
558 619
559=item $handle->autocork ($boolean) 620=item $handle->autocork ($boolean)
560 621
561Enables or disables the current autocork behaviour (see C<autocork> 622Enables or disables the current autocork behaviour (see C<autocork>
562constructor argument). Changes will only take effect on the next write. 623constructor argument). Changes will only take effect on the next write.
… …
577sub no_delay { 638sub no_delay {
578 $_[0]{no_delay} = $_[1]; 639 $_[0]{no_delay} = $_[1];
579 640
580 eval { 641 eval {
581 local $SIG{__DIE__}; 642 local $SIG{__DIE__};
582 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1] 643 setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
583 if $_[0]{fh}; 644 if $_[0]{fh};
584 }; 645 };
585} 646}
586 647
648=item $handle->keepalive ($boolean)
649
650Enables or disables the C<keepalive> setting (see constructor argument of
651the same name for details).
652
653=cut
654
655sub keepalive {
656 $_[0]{keepalive} = $_[1];
657
658 eval {
659 local $SIG{__DIE__};
660 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
661 if $_[0]{fh};
662 };
663}
664
665=item $handle->oobinline ($boolean)
666
667Enables or disables the C<oobinline> setting (see constructor argument of
668the same name for details).
669
670=cut
671
672sub oobinline {
673 $_[0]{oobinline} = $_[1];
674
675 eval {
676 local $SIG{__DIE__};
677 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
678 if $_[0]{fh};
679 };
680}
681
682=item $handle->keepalive ($boolean)
683
684Enables or disables the C<keepalive> setting (see constructor argument of
685the same name for details).
686
687=cut
688
689sub keepalive {
690 $_[0]{keepalive} = $_[1];
691
692 eval {
693 local $SIG{__DIE__};
694 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
695 if $_[0]{fh};
696 };
697}
698
587=item $handle->on_starttls ($cb) 699=item $handle->on_starttls ($cb)
588 700
589Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument). 701Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
590 702
591=cut 703=cut
… …
598 710
599Replace the current C<on_stoptls> callback (see the C<on_stoptls> constructor argument). 711Replace the current C<on_stoptls> callback (see the C<on_stoptls> constructor argument).
600 712
601=cut 713=cut
602 714
603sub on_starttls { 715sub on_stoptls {
604 $_[0]{on_stoptls} = $_[1]; 716 $_[0]{on_stoptls} = $_[1];
605} 717}
606 718
607=item $handle->rbuf_max ($max_octets) 719=item $handle->rbuf_max ($max_octets)
608 720
… …
616 728
617############################################################################# 729#############################################################################
618 730
619=item $handle->timeout ($seconds) 731=item $handle->timeout ($seconds)
620 732
733=item $handle->rtimeout ($seconds)
734
735=item $handle->wtimeout ($seconds)
736
621Configures (or disables) the inactivity timeout. 737Configures (or disables) the inactivity timeout.
622 738
623=cut 739=item $handle->timeout_reset
624 740
625sub timeout { 741=item $handle->rtimeout_reset
742
743=item $handle->wtimeout_reset
744
745Reset the activity timeout, as if data was received or sent.
746
747These methods are cheap to call.
748
749=cut
750
751for my $dir ("", "r", "w") {
752 my $timeout = "${dir}timeout";
753 my $tw = "_${dir}tw";
754 my $on_timeout = "on_${dir}timeout";
755 my $activity = "_${dir}activity";
756 my $cb;
757
758 *$on_timeout = sub {
759 $_[0]{$on_timeout} = $_[1];
760 };
761
762 *$timeout = sub {
626 my ($self, $timeout) = @_; 763 my ($self, $new_value) = @_;
627 764
628 $self->{timeout} = $timeout; 765 $self->{$timeout} = $new_value;
629 delete $self->{_tw}; 766 delete $self->{$tw}; &$cb;
630 $self->_timeout; 767 };
631}
632 768
769 *{"${dir}timeout_reset"} = sub {
770 $_[0]{$activity} = AE::now;
771 };
772
773 # main workhorse:
633# reset the timeout watcher, as neccessary 774 # reset the timeout watcher, as neccessary
634# also check for time-outs 775 # also check for time-outs
635sub _timeout { 776 $cb = sub {
636 my ($self) = @_; 777 my ($self) = @_;
637 778
638 if ($self->{timeout} && $self->{fh}) { 779 if ($self->{$timeout} && $self->{fh}) {
639 my $NOW = AnyEvent->now; 780 my $NOW = AE::now;
640 781
641 # when would the timeout trigger? 782 # when would the timeout trigger?
642 my $after = $self->{_activity} + $self->{timeout} - $NOW; 783 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
643 784
644 # now or in the past already? 785 # now or in the past already?
645 if ($after <= 0) { 786 if ($after <= 0) {
646 $self->{_activity} = $NOW; 787 $self->{$activity} = $NOW;
647 788
648 if ($self->{on_timeout}) { 789 if ($self->{$on_timeout}) {
649 $self->{on_timeout}($self); 790 $self->{$on_timeout}($self);
650 } else { 791 } else {
651 $self->_error (Errno::ETIMEDOUT); 792 $self->_error (Errno::ETIMEDOUT);
793 }
794
795 # callback could have changed timeout value, optimise
796 return unless $self->{$timeout};
797
798 # calculate new after
799 $after = $self->{$timeout};
652 } 800 }
653 801
654 # callback could have changed timeout value, optimise 802 Scalar::Util::weaken $self;
655 return unless $self->{timeout}; 803 return unless $self; # ->error could have destroyed $self
656 804
657 # calculate new after 805 $self->{$tw} ||= AE::timer $after, 0, sub {
658 $after = $self->{timeout}; 806 delete $self->{$tw};
807 $cb->($self);
808 };
809 } else {
810 delete $self->{$tw};
659 } 811 }
660
661 Scalar::Util::weaken $self;
662 return unless $self; # ->error could have destroyed $self
663
664 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
665 delete $self->{_tw};
666 $self->_timeout;
667 });
668 } else {
669 delete $self->{_tw};
670 } 812 }
671} 813}
672 814
673############################################################################# 815#############################################################################
674 816
… …
689 831
690=item $handle->on_drain ($cb) 832=item $handle->on_drain ($cb)
691 833
692Sets the C<on_drain> callback or clears it (see the description of 834Sets the C<on_drain> callback or clears it (see the description of
693C<on_drain> in the constructor). 835C<on_drain> in the constructor).
836
837This method may invoke callbacks (and therefore the handle might be
838destroyed after it returns).
694 839
695=cut 840=cut
696 841
697sub on_drain { 842sub on_drain {
698 my ($self, $cb) = @_; 843 my ($self, $cb) = @_;
… …
707 852
708Queues the given scalar to be written. You can push as much data as you 853Queues the given scalar to be written. You can push as much data as you
709want (only limited by the available memory), as C<AnyEvent::Handle> 854want (only limited by the available memory), as C<AnyEvent::Handle>
710buffers it independently of the kernel. 855buffers it independently of the kernel.
711 856
857This method may invoke callbacks (and therefore the handle might be
858destroyed after it returns).
859
712=cut 860=cut
713 861
714sub _drain_wbuf { 862sub _drain_wbuf {
715 my ($self) = @_; 863 my ($self) = @_;
716 864
… …
722 my $len = syswrite $self->{fh}, $self->{wbuf}; 870 my $len = syswrite $self->{fh}, $self->{wbuf};
723 871
724 if (defined $len) { 872 if (defined $len) {
725 substr $self->{wbuf}, 0, $len, ""; 873 substr $self->{wbuf}, 0, $len, "";
726 874
727 $self->{_activity} = AnyEvent->now; 875 $self->{_activity} = $self->{_wactivity} = AE::now;
728 876
729 $self->{on_drain}($self) 877 $self->{on_drain}($self)
730 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 878 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
731 && $self->{on_drain}; 879 && $self->{on_drain};
732 880
… …
738 886
739 # try to write data immediately 887 # try to write data immediately
740 $cb->() unless $self->{autocork}; 888 $cb->() unless $self->{autocork};
741 889
742 # if still data left in wbuf, we need to poll 890 # if still data left in wbuf, we need to poll
743 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 891 $self->{_ww} = AE::io $self->{fh}, 1, $cb
744 if length $self->{wbuf}; 892 if length $self->{wbuf};
745 }; 893 };
746} 894}
747 895
748our %WH; 896our %WH;
749 897
898# deprecated
750sub register_write_type($$) { 899sub register_write_type($$) {
751 $WH{$_[0]} = $_[1]; 900 $WH{$_[0]} = $_[1];
752} 901}
753 902
754sub push_write { 903sub push_write {
755 my $self = shift; 904 my $self = shift;
756 905
757 if (@_ > 1) { 906 if (@_ > 1) {
758 my $type = shift; 907 my $type = shift;
759 908
909 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
760 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 910 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_write")
761 ->($self, @_); 911 ->($self, @_);
762 } 912 }
763 913
914 # we downgrade here to avoid hard-to-track-down bugs,
915 # and diagnose the problem earlier and better.
916
764 if ($self->{tls}) { 917 if ($self->{tls}) {
765 $self->{_tls_wbuf} .= $_[0]; 918 utf8::downgrade $self->{_tls_wbuf} .= $_[0];
766 &_dotls ($self) if $self->{fh}; 919 &_dotls ($self) if $self->{fh};
767 } else { 920 } else {
768 $self->{wbuf} .= $_[0]; 921 utf8::downgrade $self->{wbuf} .= $_[0];
769 $self->_drain_wbuf if $self->{fh}; 922 $self->_drain_wbuf if $self->{fh};
770 } 923 }
771} 924}
772 925
773=item $handle->push_write (type => @args) 926=item $handle->push_write (type => @args)
774 927
775Instead of formatting your data yourself, you can also let this module do 928Instead of formatting your data yourself, you can also let this module
776the job by specifying a type and type-specific arguments. 929do the job by specifying a type and type-specific arguments. You
930can also specify the (fully qualified) name of a package, in which
931case AnyEvent tries to load the package and then expects to find the
932C<anyevent_read_type> function inside (see "custom write types", below).
777 933
778Predefined types are (if you have ideas for additional types, feel free to 934Predefined types are (if you have ideas for additional types, feel free to
779drop by and tell us): 935drop by and tell us):
780 936
781=over 4 937=over 4
… …
838Other languages could read single lines terminated by a newline and pass 994Other languages could read single lines terminated by a newline and pass
839this line into their JSON decoder of choice. 995this line into their JSON decoder of choice.
840 996
841=cut 997=cut
842 998
999sub json_coder() {
1000 eval { require JSON::XS; JSON::XS->new->utf8 }
1001 || do { require JSON; JSON->new->utf8 }
1002}
1003
843register_write_type json => sub { 1004register_write_type json => sub {
844 my ($self, $ref) = @_; 1005 my ($self, $ref) = @_;
845 1006
846 require JSON; 1007 my $json = $self->{json} ||= json_coder;
847 1008
848 $self->{json} ? $self->{json}->encode ($ref) 1009 $json->encode ($ref)
849 : JSON::encode_json ($ref)
850}; 1010};
851 1011
852=item storable => $reference 1012=item storable => $reference
853 1013
854Freezes the given reference using L<Storable> and writes it to the 1014Freezes the given reference using L<Storable> and writes it to the
… …
880the peer. 1040the peer.
881 1041
882You can rely on the normal read queue and C<on_eof> handling 1042You can rely on the normal read queue and C<on_eof> handling
883afterwards. This is the cleanest way to close a connection. 1043afterwards. This is the cleanest way to close a connection.
884 1044
1045This method may invoke callbacks (and therefore the handle might be
1046destroyed after it returns).
1047
885=cut 1048=cut
886 1049
887sub push_shutdown { 1050sub push_shutdown {
888 my ($self) = @_; 1051 my ($self) = @_;
889 1052
890 delete $self->{low_water_mark}; 1053 delete $self->{low_water_mark};
891 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1054 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
892} 1055}
893 1056
894=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1057=item custom write types - Package::anyevent_write_type $handle, @args
895 1058
896This function (not method) lets you add your own types to C<push_write>. 1059Instead of one of the predefined types, you can also specify the name of
1060a package. AnyEvent will try to load the package and then expects to find
1061a function named C<anyevent_write_type> inside. If it isn't found, it
1062progressively tries to load the parent package until it either finds the
1063function (good) or runs out of packages (bad).
1064
897Whenever the given C<type> is used, C<push_write> will invoke the code 1065Whenever the given C<type> is used, C<push_write> will the function with
898reference with the handle object and the remaining arguments. 1066the handle object and the remaining arguments.
899 1067
900The code reference is supposed to return a single octet string that will 1068The function is supposed to return a single octet string that will be
901be appended to the write buffer. 1069appended to the write buffer, so you cna mentally treat this function as a
1070"arguments to on-the-wire-format" converter.
902 1071
903Note that this is a function, and all types registered this way will be 1072Example: implement a custom write type C<join> that joins the remaining
904global, so try to use unique names. 1073arguments using the first one.
1074
1075 $handle->push_write (My::Type => " ", 1,2,3);
1076
1077 # uses the following package, which can be defined in the "My::Type" or in
1078 # the "My" modules to be auto-loaded, or just about anywhere when the
1079 # My::Type::anyevent_write_type is defined before invoking it.
1080
1081 package My::Type;
1082
1083 sub anyevent_write_type {
1084 my ($handle, $delim, @args) = @_;
1085
1086 join $delim, @args
1087 }
905 1088
906=cut 1089=cut
907 1090
908############################################################################# 1091#############################################################################
909 1092
… …
1065 1248
1066This replaces the currently set C<on_read> callback, or clears it (when 1249This replaces the currently set C<on_read> callback, or clears it (when
1067the new callback is C<undef>). See the description of C<on_read> in the 1250the new callback is C<undef>). See the description of C<on_read> in the
1068constructor. 1251constructor.
1069 1252
1253This method may invoke callbacks (and therefore the handle might be
1254destroyed after it returns).
1255
1070=cut 1256=cut
1071 1257
1072sub on_read { 1258sub on_read {
1073 my ($self, $cb) = @_; 1259 my ($self, $cb) = @_;
1074 1260
… …
1113 1299
1114If enough data was available, then the callback must remove all data it is 1300If enough data was available, then the callback must remove all data it is
1115interested in (which can be none at all) and return a true value. After returning 1301interested in (which can be none at all) and return a true value. After returning
1116true, it will be removed from the queue. 1302true, it will be removed from the queue.
1117 1303
1304These methods may invoke callbacks (and therefore the handle might be
1305destroyed after it returns).
1306
1118=cut 1307=cut
1119 1308
1120our %RH; 1309our %RH;
1121 1310
1122sub register_read_type($$) { 1311sub register_read_type($$) {
… …
1128 my $cb = pop; 1317 my $cb = pop;
1129 1318
1130 if (@_) { 1319 if (@_) {
1131 my $type = shift; 1320 my $type = shift;
1132 1321
1322 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1133 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 1323 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
1134 ->($self, $cb, @_); 1324 ->($self, $cb, @_);
1135 } 1325 }
1136 1326
1137 push @{ $self->{_queue} }, $cb; 1327 push @{ $self->{_queue} }, $cb;
1138 $self->_drain_rbuf; 1328 $self->_drain_rbuf;
… …
1147 1337
1148 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1338 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1149 ->($self, $cb, @_); 1339 ->($self, $cb, @_);
1150 } 1340 }
1151 1341
1152
1153 unshift @{ $self->{_queue} }, $cb; 1342 unshift @{ $self->{_queue} }, $cb;
1154 $self->_drain_rbuf; 1343 $self->_drain_rbuf;
1155} 1344}
1156 1345
1157=item $handle->push_read (type => @args, $cb) 1346=item $handle->push_read (type => @args, $cb)
1158 1347
1159=item $handle->unshift_read (type => @args, $cb) 1348=item $handle->unshift_read (type => @args, $cb)
1160 1349
1161Instead of providing a callback that parses the data itself you can chose 1350Instead of providing a callback that parses the data itself you can chose
1162between a number of predefined parsing formats, for chunks of data, lines 1351between a number of predefined parsing formats, for chunks of data, lines
1163etc. 1352etc. You can also specify the (fully qualified) name of a package, in
1353which case AnyEvent tries to load the package and then expects to find the
1354C<anyevent_read_type> function inside (see "custom read types", below).
1164 1355
1165Predefined types are (if you have ideas for additional types, feel free to 1356Predefined types are (if you have ideas for additional types, feel free to
1166drop by and tell us): 1357drop by and tell us):
1167 1358
1168=over 4 1359=over 4
… …
1408=cut 1599=cut
1409 1600
1410register_read_type json => sub { 1601register_read_type json => sub {
1411 my ($self, $cb) = @_; 1602 my ($self, $cb) = @_;
1412 1603
1413 my $json = $self->{json} ||= 1604 my $json = $self->{json} ||= json_coder;
1414 eval { require JSON::XS; JSON::XS->new->utf8 }
1415 || do { require JSON; JSON->new->utf8 };
1416 1605
1417 my $data; 1606 my $data;
1418 my $rbuf = \$self->{rbuf}; 1607 my $rbuf = \$self->{rbuf};
1419 1608
1420 sub { 1609 sub {
… …
1489 } 1678 }
1490}; 1679};
1491 1680
1492=back 1681=back
1493 1682
1494=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1683=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1495 1684
1496This function (not method) lets you add your own types to C<push_read>. 1685Instead of one of the predefined types, you can also specify the name
1686of a package. AnyEvent will try to load the package and then expects to
1687find a function named C<anyevent_read_type> inside. If it isn't found, it
1688progressively tries to load the parent package until it either finds the
1689function (good) or runs out of packages (bad).
1497 1690
1498Whenever the given C<type> is used, C<push_read> will invoke the code 1691Whenever this type is used, C<push_read> will invoke the function with the
1499reference with the handle object, the callback and the remaining 1692handle object, the original callback and the remaining arguments.
1500arguments.
1501 1693
1502The code reference is supposed to return a callback (usually a closure) 1694The function is supposed to return a callback (usually a closure) that
1503that works as a plain read callback (see C<< ->push_read ($cb) >>). 1695works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
1696mentally treat the function as a "configurable read type to read callback"
1697converter.
1504 1698
1505It should invoke the passed callback when it is done reading (remember to 1699It should invoke the original callback when it is done reading (remember
1506pass C<$handle> as first argument as all other callbacks do that). 1700to pass C<$handle> as first argument as all other callbacks do that,
1701although there is no strict requirement on this).
1507 1702
1508Note that this is a function, and all types registered this way will be
1509global, so try to use unique names.
1510
1511For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1703For examples, see the source of this module (F<perldoc -m
1512search for C<register_read_type>)). 1704AnyEvent::Handle>, search for C<register_read_type>)).
1513 1705
1514=item $handle->stop_read 1706=item $handle->stop_read
1515 1707
1516=item $handle->start_read 1708=item $handle->start_read
1517 1709
… …
1537} 1729}
1538 1730
1539sub start_read { 1731sub start_read {
1540 my ($self) = @_; 1732 my ($self) = @_;
1541 1733
1542 unless ($self->{_rw} || $self->{_eof}) { 1734 unless ($self->{_rw} || $self->{_eof} || !$self->{fh}) {
1543 Scalar::Util::weaken $self; 1735 Scalar::Util::weaken $self;
1544 1736
1545 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1737 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1546 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1738 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1547 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1739 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1548 1740
1549 if ($len > 0) { 1741 if ($len > 0) {
1550 $self->{_activity} = AnyEvent->now; 1742 $self->{_activity} = $self->{_ractivity} = AE::now;
1551 1743
1552 if ($self->{tls}) { 1744 if ($self->{tls}) {
1553 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1745 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1554 1746
1555 &_dotls ($self); 1747 &_dotls ($self);
… …
1563 $self->_drain_rbuf; 1755 $self->_drain_rbuf;
1564 1756
1565 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1757 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1566 return $self->_error ($!, 1); 1758 return $self->_error ($!, 1);
1567 } 1759 }
1568 }); 1760 };
1569 } 1761 }
1570} 1762}
1571 1763
1572our $ERROR_SYSCALL; 1764our $ERROR_SYSCALL;
1573our $ERROR_WANT_READ; 1765our $ERROR_WANT_READ;
… …
1640 && ($tmp != $ERROR_SYSCALL || $!); 1832 && ($tmp != $ERROR_SYSCALL || $!);
1641 1833
1642 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1834 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1643 $self->{wbuf} .= $tmp; 1835 $self->{wbuf} .= $tmp;
1644 $self->_drain_wbuf; 1836 $self->_drain_wbuf;
1837 $self->{tls} or return; # tls session might have gone away in callback
1645 } 1838 }
1646 1839
1647 $self->{_on_starttls} 1840 $self->{_on_starttls}
1648 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK () 1841 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK ()
1649 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established"); 1842 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established");
… …
1674 1867
1675Due to bugs in OpenSSL, it might or might not be possible to do multiple 1868Due to bugs in OpenSSL, it might or might not be possible to do multiple
1676handshakes on the same stream. Best do not attempt to use the stream after 1869handshakes on the same stream. Best do not attempt to use the stream after
1677stopping TLS. 1870stopping TLS.
1678 1871
1872This method may invoke callbacks (and therefore the handle might be
1873destroyed after it returns).
1874
1679=cut 1875=cut
1680 1876
1681our %TLS_CACHE; #TODO not yet documented, should we? 1877our %TLS_CACHE; #TODO not yet documented, should we?
1682 1878
1683sub starttls { 1879sub starttls {
… …
1694 require Net::SSLeay; 1890 require Net::SSLeay;
1695 1891
1696 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1892 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1697 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1893 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1698 1894
1699 $tls = $self->{tls}; 1895 $tls = delete $self->{tls};
1700 $ctx = $self->{tls_ctx}; 1896 $ctx = $self->{tls_ctx};
1701 1897
1702 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1898 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1703 1899
1704 if ("HASH" eq ref $ctx) { 1900 if ("HASH" eq ref $ctx) {
… …
1748 1944
1749=item $handle->stoptls 1945=item $handle->stoptls
1750 1946
1751Shuts down the SSL connection - this makes a proper EOF handshake by 1947Shuts down the SSL connection - this makes a proper EOF handshake by
1752sending a close notify to the other side, but since OpenSSL doesn't 1948sending a close notify to the other side, but since OpenSSL doesn't
1753support non-blocking shut downs, it is not guarenteed that you can re-use 1949support non-blocking shut downs, it is not guaranteed that you can re-use
1754the stream afterwards. 1950the stream afterwards.
1951
1952This method may invoke callbacks (and therefore the handle might be
1953destroyed after it returns).
1755 1954
1756=cut 1955=cut
1757 1956
1758sub stoptls { 1957sub stoptls {
1759 my ($self) = @_; 1958 my ($self) = @_;
1760 1959
1761 if ($self->{tls}) { 1960 if ($self->{tls} && $self->{fh}) {
1762 Net::SSLeay::shutdown ($self->{tls}); 1961 Net::SSLeay::shutdown ($self->{tls});
1763 1962
1764 &_dotls; 1963 &_dotls;
1765 1964
1766# # we don't give a shit. no, we do, but we can't. no...#d# 1965# # we don't give a shit. no, we do, but we can't. no...#d#
… …
1791 my $fh = delete $self->{fh}; 1990 my $fh = delete $self->{fh};
1792 my $wbuf = delete $self->{wbuf}; 1991 my $wbuf = delete $self->{wbuf};
1793 1992
1794 my @linger; 1993 my @linger;
1795 1994
1796 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1995 push @linger, AE::io $fh, 1, sub {
1797 my $len = syswrite $fh, $wbuf, length $wbuf; 1996 my $len = syswrite $fh, $wbuf, length $wbuf;
1798 1997
1799 if ($len > 0) { 1998 if ($len > 0) {
1800 substr $wbuf, 0, $len, ""; 1999 substr $wbuf, 0, $len, "";
1801 } else { 2000 } else {
1802 @linger = (); # end 2001 @linger = (); # end
1803 } 2002 }
1804 }); 2003 };
1805 push @linger, AnyEvent->timer (after => $linger, cb => sub { 2004 push @linger, AE::timer $linger, 0, sub {
1806 @linger = (); 2005 @linger = ();
1807 }); 2006 };
1808 } 2007 }
1809} 2008}
1810 2009
1811=item $handle->destroy 2010=item $handle->destroy
1812 2011
… …
1843 2042
1844sub AnyEvent::Handle::destroyed::AUTOLOAD { 2043sub AnyEvent::Handle::destroyed::AUTOLOAD {
1845 #nop 2044 #nop
1846} 2045}
1847 2046
2047=item $handle->destroyed
2048
2049Returns false as long as the handle hasn't been destroyed by a call to C<<
2050->destroy >>, true otherwise.
2051
2052Can be useful to decide whether the handle is still valid after some
2053callback possibly destroyed the handle. For example, C<< ->push_write >>,
2054C<< ->starttls >> and other methods can call user callbacks, which in turn
2055can destroy the handle, so work can be avoided by checking sometimes:
2056
2057 $hdl->starttls ("accept");
2058 return if $hdl->destroyed;
2059 $hdl->push_write (...
2060
2061Note that the call to C<push_write> will silently be ignored if the handle
2062has been destroyed, so often you can just ignore the possibility of the
2063handle being destroyed.
2064
2065=cut
2066
2067sub destroyed { 0 }
2068sub AnyEvent::Handle::destroyed::destroyed { 1 }
2069
1848=item AnyEvent::Handle::TLS_CTX 2070=item AnyEvent::Handle::TLS_CTX
1849 2071
1850This function creates and returns the AnyEvent::TLS object used by default 2072This function creates and returns the AnyEvent::TLS object used by default
1851for TLS mode. 2073for TLS mode.
1852 2074

Diff Legend

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