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.160 by root, Fri Jul 24 22:47:04 2009 UTC vs.
Revision 1.192 by root, Fri Mar 12 23:22:14 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.86;
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
101attempted, but after the file handle has been created. It could be used to 117attempted, but after the file handle has been created. It could be used to
102prepare the file handle with parameters required for the actual connect 118prepare the file handle with parameters required for the actual connect
103(as opposed to settings that can be changed when the connection is already 119(as opposed to settings that can be changed when the connection is already
104established). 120established).
105 121
122The return value of this callback should be the connect timeout value in
123seconds (or C<0>, or C<undef>, or the empty list, to indicate the default
124timeout is to be used).
125
106=item on_connect => $cb->($handle, $host, $port, $retry->()) 126=item on_connect => $cb->($handle, $host, $port, $retry->())
107 127
108This callback is called when a connection has been successfully established. 128This callback is called when a connection has been successfully established.
109 129
110The actual numeric host and port (the socket peername) are passed as 130The actual numeric host and port (the socket peername) are passed as
111parameters, together with a retry callback. 131parameters, together with a retry callback.
112 132
113When, for some reason, the handle is not acceptable, then calling 133When, for some reason, the handle is not acceptable, then calling
114C<$retry> will continue with the next conenction target (in case of 134C<$retry> will continue with the next connection target (in case of
115multi-homed hosts or SRV records there can be multiple connection 135multi-homed hosts or SRV records there can be multiple connection
116endpoints). 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
117tls status and similar properties of the handle are being reset. 137status, tls status and similar properties of the handle will have been
138reset.
118 139
119In 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.
120 141
121=item on_connect_error => $cb->($handle, $message) 142=item on_connect_error => $cb->($handle, $message)
122 143
123This callback is called when the conenction could not be 144This callback is called when the connection could not be
124established. C<$!> will contain the relevant error code, and C<$message> a 145established. C<$!> will contain the relevant error code, and C<$message> a
125message describing it (usually the same as C<"$!">). 146message describing it (usually the same as C<"$!">).
126 147
127If 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
128fatal error instead. 149fatal error instead.
212memory 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
213the file when the write queue becomes empty. 234the file when the write queue becomes empty.
214 235
215=item timeout => $fractional_seconds 236=item timeout => $fractional_seconds
216 237
238=item rtimeout => $fractional_seconds
239
240=item wtimeout => $fractional_seconds
241
217If non-zero, then this enables an "inactivity" timeout: whenever this many 242If non-zero, then these enables an "inactivity" timeout: whenever this
218seconds pass without a successful read or write on the underlying file 243many seconds pass without a successful read or write on the underlying
219handle, 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
220missing, 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>.
221 253
222Note 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
223any 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
224idle then you should disable the timout temporarily or ignore the timeout 256idle then you should disable the timout temporarily or ignore the timeout
225in 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
269accomplishd by setting this option to a true value. 301accomplishd by setting this option to a true value.
270 302
271The default is your opertaing system's default behaviour (most likely 303The default is your opertaing system's default behaviour (most likely
272enabled), this option explicitly enables or disables it, if possible. 304enabled), this option explicitly enables or disables it, if possible.
273 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
274=item read_size => <bytes> 338=item read_size => <bytes>
275 339
276The default read block size (the amount of bytes this module will 340The default read block size (the amount of bytes this module will
277try to read during each loop iteration, which affects memory 341try to read during each loop iteration, which affects memory
278requirements). Default: C<8192>. 342requirements). Default: C<8192>.
311C<undef>. 375C<undef>.
312 376
313=item tls => "accept" | "connect" | Net::SSLeay::SSL object 377=item tls => "accept" | "connect" | Net::SSLeay::SSL object
314 378
315When this parameter is given, it enables TLS (SSL) mode, that means 379When this parameter is given, it enables TLS (SSL) mode, that means
316AnyEvent 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
317established and will transparently encrypt/decrypt data afterwards. 381established and will transparently encrypt/decrypt data afterwards.
318 382
319All TLS protocol errors will be signalled as C<EPROTO>, with an 383All TLS protocol errors will be signalled as C<EPROTO>, with an
320appropriate error message. 384appropriate error message.
321 385
434 delete $self->{_skip_drain_rbuf}; 498 delete $self->{_skip_drain_rbuf};
435 $self->_start; 499 $self->_start;
436 500
437 $self->{on_connect} 501 $self->{on_connect}
438 and $self->{on_connect}($self, $host, $port, sub { 502 and $self->{on_connect}($self, $host, $port, sub {
439 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)};
440 $self->{_skip_drain_rbuf} = 1; 504 $self->{_skip_drain_rbuf} = 1;
441 &$retry; 505 &$retry;
442 }); 506 });
443 507
444 } else { 508 } else {
445 if ($self->{on_connect_error}) { 509 if ($self->{on_connect_error}) {
446 $self->{on_connect_error}($self, "$!"); 510 $self->{on_connect_error}($self, "$!");
447 $self->destroy; 511 $self->destroy;
448 } else { 512 } else {
449 $self->fatal ($!, 1); 513 $self->_error ($!, 1);
450 } 514 }
451 } 515 }
452 }, 516 },
453 sub { 517 sub {
454 local $self->{fh} = $_[0]; 518 local $self->{fh} = $_[0];
455 519
520 $self->{on_prepare}
456 $self->{on_prepare}->($self) 521 ? $self->{on_prepare}->($self)
457 if $self->{on_prepare}; 522 : ()
458 } 523 }
459 ); 524 );
460 } 525 }
461 526
462 } else { 527 } else {
469sub _start { 534sub _start {
470 my ($self) = @_; 535 my ($self) = @_;
471 536
472 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 537 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
473 538
539 $self->{_activity} =
540 $self->{_ractivity} =
474 $self->{_activity} = AnyEvent->now; 541 $self->{_wactivity} = AE::now;
475 $self->_timeout;
476 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
477 $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};
478 549
550 $self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
551
479 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 552 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
480 if $self->{tls}; 553 if $self->{tls};
481 554
482 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 555 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
483 556
484 $self->start_read 557 $self->start_read
485 if $self->{on_read} || @{ $self->{_queue} }; 558 if $self->{on_read} || @{ $self->{_queue} };
486 559
487 $self->_drain_wbuf; 560 $self->_drain_wbuf;
488} 561}
489
490#sub _shutdown {
491# my ($self) = @_;
492#
493# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
494# $self->{_eof} = 1; # tell starttls et. al to stop trying
495#
496# &_freetls;
497#}
498 562
499sub _error { 563sub _error {
500 my ($self, $errno, $fatal, $message) = @_; 564 my ($self, $errno, $fatal, $message) = @_;
501 565
502 $! = $errno; 566 $! = $errno;
503 $message ||= "$!"; 567 $message ||= "$!";
504 568
505 if ($self->{on_error}) { 569 if ($self->{on_error}) {
506 $self->{on_error}($self, $fatal, $message); 570 $self->{on_error}($self, $fatal, $message);
507 $self->destroy if $fatal; 571 $self->destroy if $fatal;
508 } elsif ($self->{fh}) { 572 } elsif ($self->{fh} || $self->{connect}) {
509 $self->destroy; 573 $self->destroy;
510 Carp::croak "AnyEvent::Handle uncaught error: $message"; 574 Carp::croak "AnyEvent::Handle uncaught error: $message";
511 } 575 }
512} 576}
513 577
539 $_[0]{on_eof} = $_[1]; 603 $_[0]{on_eof} = $_[1];
540} 604}
541 605
542=item $handle->on_timeout ($cb) 606=item $handle->on_timeout ($cb)
543 607
544Replace the current C<on_timeout> callback, or disables the callback (but 608=item $handle->on_rtimeout ($cb)
545not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
546argument and method.
547 609
548=cut 610=item $handle->on_wtimeout ($cb)
549 611
550sub on_timeout { 612Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
551 $_[0]{on_timeout} = $_[1]; 613callback, or disables the callback (but not the timeout) if C<$cb> =
552} 614C<undef>. See the C<timeout> constructor argument and method.
615
616=cut
617
618# see below
553 619
554=item $handle->autocork ($boolean) 620=item $handle->autocork ($boolean)
555 621
556Enables or disables the current autocork behaviour (see C<autocork> 622Enables or disables the current autocork behaviour (see C<autocork>
557constructor argument). Changes will only take effect on the next write. 623constructor argument). Changes will only take effect on the next write.
572sub no_delay { 638sub no_delay {
573 $_[0]{no_delay} = $_[1]; 639 $_[0]{no_delay} = $_[1];
574 640
575 eval { 641 eval {
576 local $SIG{__DIE__}; 642 local $SIG{__DIE__};
577 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1] 643 setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
578 if $_[0]{fh}; 644 if $_[0]{fh};
579 }; 645 };
580} 646}
581 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
582=item $handle->on_starttls ($cb) 699=item $handle->on_starttls ($cb)
583 700
584Replace 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).
585 702
586=cut 703=cut
593 710
594Replace 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).
595 712
596=cut 713=cut
597 714
598sub on_starttls { 715sub on_stoptls {
599 $_[0]{on_stoptls} = $_[1]; 716 $_[0]{on_stoptls} = $_[1];
600} 717}
601 718
719=item $handle->rbuf_max ($max_octets)
720
721Configures the C<rbuf_max> setting (C<undef> disables it).
722
723=cut
724
725sub rbuf_max {
726 $_[0]{rbuf_max} = $_[1];
727}
728
602############################################################################# 729#############################################################################
603 730
604=item $handle->timeout ($seconds) 731=item $handle->timeout ($seconds)
605 732
733=item $handle->rtimeout ($seconds)
734
735=item $handle->wtimeout ($seconds)
736
606Configures (or disables) the inactivity timeout. 737Configures (or disables) the inactivity timeout.
607 738
608=cut 739=item $handle->timeout_reset
609 740
610sub 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 {
611 my ($self, $timeout) = @_; 763 my ($self, $new_value) = @_;
612 764
613 $self->{timeout} = $timeout; 765 $self->{$timeout} = $new_value;
614 $self->_timeout; 766 delete $self->{$tw}; &$cb;
615} 767 };
616 768
769 *{"${dir}timeout_reset"} = sub {
770 $_[0]{$activity} = AE::now;
771 };
772
773 # main workhorse:
617# reset the timeout watcher, as neccessary 774 # reset the timeout watcher, as neccessary
618# also check for time-outs 775 # also check for time-outs
619sub _timeout { 776 $cb = sub {
620 my ($self) = @_; 777 my ($self) = @_;
621 778
622 if ($self->{timeout} && $self->{fh}) { 779 if ($self->{$timeout} && $self->{fh}) {
623 my $NOW = AnyEvent->now; 780 my $NOW = AE::now;
624 781
625 # when would the timeout trigger? 782 # when would the timeout trigger?
626 my $after = $self->{_activity} + $self->{timeout} - $NOW; 783 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
627 784
628 # now or in the past already? 785 # now or in the past already?
629 if ($after <= 0) { 786 if ($after <= 0) {
630 $self->{_activity} = $NOW; 787 $self->{$activity} = $NOW;
631 788
632 if ($self->{on_timeout}) { 789 if ($self->{$on_timeout}) {
633 $self->{on_timeout}($self); 790 $self->{$on_timeout}($self);
634 } else { 791 } else {
635 $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};
636 } 800 }
637 801
638 # callback could have changed timeout value, optimise 802 Scalar::Util::weaken $self;
639 return unless $self->{timeout}; 803 return unless $self; # ->error could have destroyed $self
640 804
641 # calculate new after 805 $self->{$tw} ||= AE::timer $after, 0, sub {
642 $after = $self->{timeout}; 806 delete $self->{$tw};
807 $cb->($self);
808 };
809 } else {
810 delete $self->{$tw};
643 } 811 }
644
645 Scalar::Util::weaken $self;
646 return unless $self; # ->error could have destroyed $self
647
648 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
649 delete $self->{_tw};
650 $self->_timeout;
651 });
652 } else {
653 delete $self->{_tw};
654 } 812 }
655} 813}
656 814
657############################################################################# 815#############################################################################
658 816
706 my $len = syswrite $self->{fh}, $self->{wbuf}; 864 my $len = syswrite $self->{fh}, $self->{wbuf};
707 865
708 if (defined $len) { 866 if (defined $len) {
709 substr $self->{wbuf}, 0, $len, ""; 867 substr $self->{wbuf}, 0, $len, "";
710 868
711 $self->{_activity} = AnyEvent->now; 869 $self->{_activity} = $self->{_wactivity} = AE::now;
712 870
713 $self->{on_drain}($self) 871 $self->{on_drain}($self)
714 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 872 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
715 && $self->{on_drain}; 873 && $self->{on_drain};
716 874
722 880
723 # try to write data immediately 881 # try to write data immediately
724 $cb->() unless $self->{autocork}; 882 $cb->() unless $self->{autocork};
725 883
726 # if still data left in wbuf, we need to poll 884 # if still data left in wbuf, we need to poll
727 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 885 $self->{_ww} = AE::io $self->{fh}, 1, $cb
728 if length $self->{wbuf}; 886 if length $self->{wbuf};
729 }; 887 };
730} 888}
731 889
732our %WH; 890our %WH;
733 891
892# deprecated
734sub register_write_type($$) { 893sub register_write_type($$) {
735 $WH{$_[0]} = $_[1]; 894 $WH{$_[0]} = $_[1];
736} 895}
737 896
738sub push_write { 897sub push_write {
739 my $self = shift; 898 my $self = shift;
740 899
741 if (@_ > 1) { 900 if (@_ > 1) {
742 my $type = shift; 901 my $type = shift;
743 902
903 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
744 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 904 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_write")
745 ->($self, @_); 905 ->($self, @_);
746 } 906 }
747 907
908 # we downgrade here to avoid hard-to-track-down bugs,
909 # and diagnose the problem earlier and better.
910
748 if ($self->{tls}) { 911 if ($self->{tls}) {
749 $self->{_tls_wbuf} .= $_[0]; 912 utf8::downgrade $self->{_tls_wbuf} .= $_[0];
750 &_dotls ($self) if $self->{fh}; 913 &_dotls ($self) if $self->{fh};
751 } else { 914 } else {
752 $self->{wbuf} .= $_[0]; 915 utf8::downgrade $self->{wbuf} .= $_[0];
753 $self->_drain_wbuf if $self->{fh}; 916 $self->_drain_wbuf if $self->{fh};
754 } 917 }
755} 918}
756 919
757=item $handle->push_write (type => @args) 920=item $handle->push_write (type => @args)
758 921
759Instead of formatting your data yourself, you can also let this module do 922Instead of formatting your data yourself, you can also let this module
760the job by specifying a type and type-specific arguments. 923do the job by specifying a type and type-specific arguments. You
924can also specify the (fully qualified) name of a package, in which
925case AnyEvent tries to load the package and then expects to find the
926C<anyevent_read_type> function inside (see "custom write types", below).
761 927
762Predefined types are (if you have ideas for additional types, feel free to 928Predefined types are (if you have ideas for additional types, feel free to
763drop by and tell us): 929drop by and tell us):
764 930
765=over 4 931=over 4
822Other languages could read single lines terminated by a newline and pass 988Other languages could read single lines terminated by a newline and pass
823this line into their JSON decoder of choice. 989this line into their JSON decoder of choice.
824 990
825=cut 991=cut
826 992
993sub json_coder() {
994 eval { require JSON::XS; JSON::XS->new->utf8 }
995 || do { require JSON; JSON->new->utf8 }
996}
997
827register_write_type json => sub { 998register_write_type json => sub {
828 my ($self, $ref) = @_; 999 my ($self, $ref) = @_;
829 1000
830 require JSON; 1001 my $json = $self->{json} ||= json_coder;
831 1002
832 $self->{json} ? $self->{json}->encode ($ref) 1003 $json->encode ($ref)
833 : JSON::encode_json ($ref)
834}; 1004};
835 1005
836=item storable => $reference 1006=item storable => $reference
837 1007
838Freezes the given reference using L<Storable> and writes it to the 1008Freezes the given reference using L<Storable> and writes it to the
873 1043
874 delete $self->{low_water_mark}; 1044 delete $self->{low_water_mark};
875 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1045 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
876} 1046}
877 1047
878=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1048=item custom write types - Package::anyevent_write_type $handle, @args
879 1049
880This function (not method) lets you add your own types to C<push_write>. 1050Instead of one of the predefined types, you can also specify the name of
1051a package. AnyEvent will try to load the package and then expects to find
1052a function named C<anyevent_write_type> inside. If it isn't found, it
1053progressively tries to load the parent package until it either finds the
1054function (good) or runs out of packages (bad).
1055
881Whenever the given C<type> is used, C<push_write> will invoke the code 1056Whenever the given C<type> is used, C<push_write> will the function with
882reference with the handle object and the remaining arguments. 1057the handle object and the remaining arguments.
883 1058
884The code reference is supposed to return a single octet string that will 1059The function is supposed to return a single octet string that will be
885be appended to the write buffer. 1060appended to the write buffer, so you cna mentally treat this function as a
1061"arguments to on-the-wire-format" converter.
886 1062
887Note that this is a function, and all types registered this way will be 1063Example: implement a custom write type C<join> that joins the remaining
888global, so try to use unique names. 1064arguments using the first one.
1065
1066 $handle->push_write (My::Type => " ", 1,2,3);
1067
1068 # uses the following package, which can be defined in the "My::Type" or in
1069 # the "My" modules to be auto-loaded, or just about anywhere when the
1070 # My::Type::anyevent_write_type is defined before invoking it.
1071
1072 package My::Type;
1073
1074 sub anyevent_write_type {
1075 my ($handle, $delim, @args) = @_;
1076
1077 join $delim, @args
1078 }
889 1079
890=cut 1080=cut
891 1081
892############################################################################# 1082#############################################################################
893 1083
976 1166
977sub _drain_rbuf { 1167sub _drain_rbuf {
978 my ($self) = @_; 1168 my ($self) = @_;
979 1169
980 # avoid recursion 1170 # avoid recursion
981 return if exists $self->{_skip_drain_rbuf}; 1171 return if $self->{_skip_drain_rbuf};
982 local $self->{_skip_drain_rbuf} = 1; 1172 local $self->{_skip_drain_rbuf} = 1;
983
984 if (
985 defined $self->{rbuf_max}
986 && $self->{rbuf_max} < length $self->{rbuf}
987 ) {
988 $self->_error (Errno::ENOSPC, 1), return;
989 }
990 1173
991 while () { 1174 while () {
992 # we need to use a separate tls read buffer, as we must not receive data while 1175 # we need to use a separate tls read buffer, as we must not receive data while
993 # we are draining the buffer, and this can only happen with TLS. 1176 # we are draining the buffer, and this can only happen with TLS.
994 $self->{rbuf} .= delete $self->{_tls_rbuf} if exists $self->{_tls_rbuf}; 1177 $self->{rbuf} .= delete $self->{_tls_rbuf}
1178 if exists $self->{_tls_rbuf};
995 1179
996 my $len = length $self->{rbuf}; 1180 my $len = length $self->{rbuf};
997 1181
998 if (my $cb = shift @{ $self->{_queue} }) { 1182 if (my $cb = shift @{ $self->{_queue} }) {
999 unless ($cb->($self)) { 1183 unless ($cb->($self)) {
1000 if ($self->{_eof}) { 1184 # no progress can be made
1001 # no progress can be made (not enough data and no data forthcoming) 1185 # (not enough data and no data forthcoming)
1002 $self->_error (Errno::EPIPE, 1), return; 1186 $self->_error (Errno::EPIPE, 1), return
1003 } 1187 if $self->{_eof};
1004 1188
1005 unshift @{ $self->{_queue} }, $cb; 1189 unshift @{ $self->{_queue} }, $cb;
1006 last; 1190 last;
1007 } 1191 }
1008 } elsif ($self->{on_read}) { 1192 } elsif ($self->{on_read}) {
1028 last; 1212 last;
1029 } 1213 }
1030 } 1214 }
1031 1215
1032 if ($self->{_eof}) { 1216 if ($self->{_eof}) {
1033 if ($self->{on_eof}) { 1217 $self->{on_eof}
1034 $self->{on_eof}($self) 1218 ? $self->{on_eof}($self)
1035 } else {
1036 $self->_error (0, 1, "Unexpected end-of-file"); 1219 : $self->_error (0, 1, "Unexpected end-of-file");
1037 } 1220
1221 return;
1222 }
1223
1224 if (
1225 defined $self->{rbuf_max}
1226 && $self->{rbuf_max} < length $self->{rbuf}
1227 ) {
1228 $self->_error (Errno::ENOSPC, 1), return;
1038 } 1229 }
1039 1230
1040 # may need to restart read watcher 1231 # may need to restart read watcher
1041 unless ($self->{_rw}) { 1232 unless ($self->{_rw}) {
1042 $self->start_read 1233 $self->start_read
1111 my $cb = pop; 1302 my $cb = pop;
1112 1303
1113 if (@_) { 1304 if (@_) {
1114 my $type = shift; 1305 my $type = shift;
1115 1306
1307 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1116 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 1308 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
1117 ->($self, $cb, @_); 1309 ->($self, $cb, @_);
1118 } 1310 }
1119 1311
1120 push @{ $self->{_queue} }, $cb; 1312 push @{ $self->{_queue} }, $cb;
1121 $self->_drain_rbuf; 1313 $self->_drain_rbuf;
1130 1322
1131 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1323 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1132 ->($self, $cb, @_); 1324 ->($self, $cb, @_);
1133 } 1325 }
1134 1326
1135
1136 unshift @{ $self->{_queue} }, $cb; 1327 unshift @{ $self->{_queue} }, $cb;
1137 $self->_drain_rbuf; 1328 $self->_drain_rbuf;
1138} 1329}
1139 1330
1140=item $handle->push_read (type => @args, $cb) 1331=item $handle->push_read (type => @args, $cb)
1141 1332
1142=item $handle->unshift_read (type => @args, $cb) 1333=item $handle->unshift_read (type => @args, $cb)
1143 1334
1144Instead of providing a callback that parses the data itself you can chose 1335Instead of providing a callback that parses the data itself you can chose
1145between a number of predefined parsing formats, for chunks of data, lines 1336between a number of predefined parsing formats, for chunks of data, lines
1146etc. 1337etc. You can also specify the (fully qualified) name of a package, in
1338which case AnyEvent tries to load the package and then expects to find the
1339C<anyevent_read_type> function inside (see "custom read types", below).
1147 1340
1148Predefined types are (if you have ideas for additional types, feel free to 1341Predefined types are (if you have ideas for additional types, feel free to
1149drop by and tell us): 1342drop by and tell us):
1150 1343
1151=over 4 1344=over 4
1391=cut 1584=cut
1392 1585
1393register_read_type json => sub { 1586register_read_type json => sub {
1394 my ($self, $cb) = @_; 1587 my ($self, $cb) = @_;
1395 1588
1396 my $json = $self->{json} ||= 1589 my $json = $self->{json} ||= json_coder;
1397 eval { require JSON::XS; JSON::XS->new->utf8 }
1398 || do { require JSON; JSON->new->utf8 };
1399 1590
1400 my $data; 1591 my $data;
1401 my $rbuf = \$self->{rbuf}; 1592 my $rbuf = \$self->{rbuf};
1402 1593
1403 sub { 1594 sub {
1472 } 1663 }
1473}; 1664};
1474 1665
1475=back 1666=back
1476 1667
1477=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1668=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1478 1669
1479This function (not method) lets you add your own types to C<push_read>. 1670Instead of one of the predefined types, you can also specify the name
1671of a package. AnyEvent will try to load the package and then expects to
1672find a function named C<anyevent_read_type> inside. If it isn't found, it
1673progressively tries to load the parent package until it either finds the
1674function (good) or runs out of packages (bad).
1480 1675
1481Whenever the given C<type> is used, C<push_read> will invoke the code 1676Whenever this type is used, C<push_read> will invoke the function with the
1482reference with the handle object, the callback and the remaining 1677handle object, the original callback and the remaining arguments.
1483arguments.
1484 1678
1485The code reference is supposed to return a callback (usually a closure) 1679The function is supposed to return a callback (usually a closure) that
1486that works as a plain read callback (see C<< ->push_read ($cb) >>). 1680works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
1681mentally treat the function as a "configurable read type to read callback"
1682converter.
1487 1683
1488It should invoke the passed callback when it is done reading (remember to 1684It should invoke the original callback when it is done reading (remember
1489pass C<$handle> as first argument as all other callbacks do that). 1685to pass C<$handle> as first argument as all other callbacks do that,
1686although there is no strict requirement on this).
1490 1687
1491Note that this is a function, and all types registered this way will be
1492global, so try to use unique names.
1493
1494For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1688For examples, see the source of this module (F<perldoc -m
1495search for C<register_read_type>)). 1689AnyEvent::Handle>, search for C<register_read_type>)).
1496 1690
1497=item $handle->stop_read 1691=item $handle->stop_read
1498 1692
1499=item $handle->start_read 1693=item $handle->start_read
1500 1694
1520} 1714}
1521 1715
1522sub start_read { 1716sub start_read {
1523 my ($self) = @_; 1717 my ($self) = @_;
1524 1718
1525 unless ($self->{_rw} || $self->{_eof}) { 1719 unless ($self->{_rw} || $self->{_eof} || !$self->{fh}) {
1526 Scalar::Util::weaken $self; 1720 Scalar::Util::weaken $self;
1527 1721
1528 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1722 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1529 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1723 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1530 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1724 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1531 1725
1532 if ($len > 0) { 1726 if ($len > 0) {
1533 $self->{_activity} = AnyEvent->now; 1727 $self->{_activity} = $self->{_ractivity} = AE::now;
1534 1728
1535 if ($self->{tls}) { 1729 if ($self->{tls}) {
1536 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1730 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1537 1731
1538 &_dotls ($self); 1732 &_dotls ($self);
1546 $self->_drain_rbuf; 1740 $self->_drain_rbuf;
1547 1741
1548 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1742 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1549 return $self->_error ($!, 1); 1743 return $self->_error ($!, 1);
1550 } 1744 }
1551 }); 1745 };
1552 } 1746 }
1553} 1747}
1554 1748
1555our $ERROR_SYSCALL; 1749our $ERROR_SYSCALL;
1556our $ERROR_WANT_READ; 1750our $ERROR_WANT_READ;
1623 && ($tmp != $ERROR_SYSCALL || $!); 1817 && ($tmp != $ERROR_SYSCALL || $!);
1624 1818
1625 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1819 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1626 $self->{wbuf} .= $tmp; 1820 $self->{wbuf} .= $tmp;
1627 $self->_drain_wbuf; 1821 $self->_drain_wbuf;
1822 $self->{tls} or return; # tls session might have gone away in callback
1628 } 1823 }
1629 1824
1630 $self->{_on_starttls} 1825 $self->{_on_starttls}
1631 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK () 1826 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK ()
1632 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established"); 1827 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established");
1677 require Net::SSLeay; 1872 require Net::SSLeay;
1678 1873
1679 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1874 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1680 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1875 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1681 1876
1682 $tls = $self->{tls}; 1877 $tls = delete $self->{tls};
1683 $ctx = $self->{tls_ctx}; 1878 $ctx = $self->{tls_ctx};
1684 1879
1685 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1880 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1686 1881
1687 if ("HASH" eq ref $ctx) { 1882 if ("HASH" eq ref $ctx) {
1716 Net::SSLeay::CTX_set_mode ($tls, 1|2); 1911 Net::SSLeay::CTX_set_mode ($tls, 1|2);
1717 1912
1718 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1913 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1719 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1914 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1720 1915
1916 Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
1917
1721 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio}); 1918 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1722 1919
1723 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) } 1920 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1724 if $self->{on_starttls}; 1921 if $self->{on_starttls};
1725 1922
1729 1926
1730=item $handle->stoptls 1927=item $handle->stoptls
1731 1928
1732Shuts down the SSL connection - this makes a proper EOF handshake by 1929Shuts down the SSL connection - this makes a proper EOF handshake by
1733sending a close notify to the other side, but since OpenSSL doesn't 1930sending a close notify to the other side, but since OpenSSL doesn't
1734support non-blocking shut downs, it is not guarenteed that you can re-use 1931support non-blocking shut downs, it is not guaranteed that you can re-use
1735the stream afterwards. 1932the stream afterwards.
1736 1933
1737=cut 1934=cut
1738 1935
1739sub stoptls { 1936sub stoptls {
1740 my ($self) = @_; 1937 my ($self) = @_;
1741 1938
1742 if ($self->{tls}) { 1939 if ($self->{tls} && $self->{fh}) {
1743 Net::SSLeay::shutdown ($self->{tls}); 1940 Net::SSLeay::shutdown ($self->{tls});
1744 1941
1745 &_dotls; 1942 &_dotls;
1746 1943
1747# # we don't give a shit. no, we do, but we can't. no...#d# 1944# # we don't give a shit. no, we do, but we can't. no...#d#
1754 my ($self) = @_; 1951 my ($self) = @_;
1755 1952
1756 return unless $self->{tls}; 1953 return unless $self->{tls};
1757 1954
1758 $self->{tls_ctx}->_put_session (delete $self->{tls}) 1955 $self->{tls_ctx}->_put_session (delete $self->{tls})
1759 if ref $self->{tls}; 1956 if $self->{tls} > 0;
1760 1957
1761 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 1958 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1762} 1959}
1763 1960
1764sub DESTROY { 1961sub DESTROY {
1772 my $fh = delete $self->{fh}; 1969 my $fh = delete $self->{fh};
1773 my $wbuf = delete $self->{wbuf}; 1970 my $wbuf = delete $self->{wbuf};
1774 1971
1775 my @linger; 1972 my @linger;
1776 1973
1777 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1974 push @linger, AE::io $fh, 1, sub {
1778 my $len = syswrite $fh, $wbuf, length $wbuf; 1975 my $len = syswrite $fh, $wbuf, length $wbuf;
1779 1976
1780 if ($len > 0) { 1977 if ($len > 0) {
1781 substr $wbuf, 0, $len, ""; 1978 substr $wbuf, 0, $len, "";
1782 } else { 1979 } else {
1783 @linger = (); # end 1980 @linger = (); # end
1784 } 1981 }
1785 }); 1982 };
1786 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1983 push @linger, AE::timer $linger, 0, sub {
1787 @linger = (); 1984 @linger = ();
1788 }); 1985 };
1789 } 1986 }
1790} 1987}
1791 1988
1792=item $handle->destroy 1989=item $handle->destroy
1793 1990
1794Shuts down the handle object as much as possible - this call ensures that 1991Shuts down the handle object as much as possible - this call ensures that
1795no further callbacks will be invoked and as many resources as possible 1992no further callbacks will be invoked and as many resources as possible
1796will be freed. You must not call any methods on the object afterwards. 1993will be freed. Any method you will call on the handle object after
1994destroying it in this way will be silently ignored (and it will return the
1995empty list).
1797 1996
1798Normally, you can just "forget" any references to an AnyEvent::Handle 1997Normally, you can just "forget" any references to an AnyEvent::Handle
1799object and it will simply shut down. This works in fatal error and EOF 1998object and it will simply shut down. This works in fatal error and EOF
1800callbacks, as well as code outside. It does I<NOT> work in a read or write 1999callbacks, as well as code outside. It does I<NOT> work in a read or write
1801callback, so when you want to destroy the AnyEvent::Handle object from 2000callback, so when you want to destroy the AnyEvent::Handle object from
1815sub destroy { 2014sub destroy {
1816 my ($self) = @_; 2015 my ($self) = @_;
1817 2016
1818 $self->DESTROY; 2017 $self->DESTROY;
1819 %$self = (); 2018 %$self = ();
2019 bless $self, "AnyEvent::Handle::destroyed";
1820} 2020}
2021
2022sub AnyEvent::Handle::destroyed::AUTOLOAD {
2023 #nop
2024}
2025
2026=item $handle->destroyed
2027
2028Returns false as long as the handle hasn't been destroyed by a call to C<<
2029->destroy >>, true otherwise.
2030
2031Can be useful to decide whether the handle is still valid after some
2032callback possibly destroyed the handle. For example, C<< ->push_write >>,
2033C<< ->starttls >> and other methods can call user callbacks, which in turn
2034can destroy the handle, so work can be avoided by checking sometimes:
2035
2036 $hdl->starttls ("accept");
2037 return if $hdl->destroyed;
2038 $hdl->push_write (...
2039
2040Note that the call to C<push_write> will silently be ignored if the handle
2041has been destroyed, so often you can just ignore the possibility of the
2042handle being destroyed.
2043
2044=cut
2045
2046sub destroyed { 0 }
2047sub AnyEvent::Handle::destroyed::destroyed { 1 }
1821 2048
1822=item AnyEvent::Handle::TLS_CTX 2049=item AnyEvent::Handle::TLS_CTX
1823 2050
1824This function creates and returns the AnyEvent::TLS object used by default 2051This function creates and returns the AnyEvent::TLS object used by default
1825for TLS mode. 2052for TLS mode.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines