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.190 by root, Thu Nov 26 10:30:24 2009 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::TLS fh => $filehandle, key => value...
… …
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
… …
722 my $len = syswrite $self->{fh}, $self->{wbuf}; 864 my $len = syswrite $self->{fh}, $self->{wbuf};
723 865
724 if (defined $len) { 866 if (defined $len) {
725 substr $self->{wbuf}, 0, $len, ""; 867 substr $self->{wbuf}, 0, $len, "";
726 868
727 $self->{_activity} = AnyEvent->now; 869 $self->{_activity} = $self->{_wactivity} = AE::now;
728 870
729 $self->{on_drain}($self) 871 $self->{on_drain}($self)
730 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})
731 && $self->{on_drain}; 873 && $self->{on_drain};
732 874
… …
738 880
739 # try to write data immediately 881 # try to write data immediately
740 $cb->() unless $self->{autocork}; 882 $cb->() unless $self->{autocork};
741 883
742 # if still data left in wbuf, we need to poll 884 # if still data left in wbuf, we need to poll
743 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 885 $self->{_ww} = AE::io $self->{fh}, 1, $cb
744 if length $self->{wbuf}; 886 if length $self->{wbuf};
745 }; 887 };
746} 888}
747 889
748our %WH; 890our %WH;
749 891
892# deprecated
750sub register_write_type($$) { 893sub register_write_type($$) {
751 $WH{$_[0]} = $_[1]; 894 $WH{$_[0]} = $_[1];
752} 895}
753 896
754sub push_write { 897sub push_write {
755 my $self = shift; 898 my $self = shift;
756 899
757 if (@_ > 1) { 900 if (@_ > 1) {
758 my $type = shift; 901 my $type = shift;
759 902
903 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
760 @_ = ($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")
761 ->($self, @_); 905 ->($self, @_);
762 } 906 }
763 907
908 # we downgrade here to avoid hard-to-track-down bugs,
909 # and diagnose the problem earlier and better.
910
764 if ($self->{tls}) { 911 if ($self->{tls}) {
765 $self->{_tls_wbuf} .= $_[0]; 912 utf8::downgrade $self->{_tls_wbuf} .= $_[0];
766 &_dotls ($self) if $self->{fh}; 913 &_dotls ($self) if $self->{fh};
767 } else { 914 } else {
768 $self->{wbuf} .= $_[0]; 915 utf8::downgrade $self->{wbuf} .= $_[0];
769 $self->_drain_wbuf if $self->{fh}; 916 $self->_drain_wbuf if $self->{fh};
770 } 917 }
771} 918}
772 919
773=item $handle->push_write (type => @args) 920=item $handle->push_write (type => @args)
774 921
775Instead of formatting your data yourself, you can also let this module do 922Instead of formatting your data yourself, you can also let this module
776the 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).
777 927
778Predefined 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
779drop by and tell us): 929drop by and tell us):
780 930
781=over 4 931=over 4
… …
838Other languages could read single lines terminated by a newline and pass 988Other languages could read single lines terminated by a newline and pass
839this line into their JSON decoder of choice. 989this line into their JSON decoder of choice.
840 990
841=cut 991=cut
842 992
993sub json_coder() {
994 eval { require JSON::XS; JSON::XS->new->utf8 }
995 || do { require JSON; JSON->new->utf8 }
996}
997
843register_write_type json => sub { 998register_write_type json => sub {
844 my ($self, $ref) = @_; 999 my ($self, $ref) = @_;
845 1000
846 require JSON; 1001 my $json = $self->{json} ||= json_coder;
847 1002
848 $self->{json} ? $self->{json}->encode ($ref) 1003 $json->encode ($ref)
849 : JSON::encode_json ($ref)
850}; 1004};
851 1005
852=item storable => $reference 1006=item storable => $reference
853 1007
854Freezes the given reference using L<Storable> and writes it to the 1008Freezes the given reference using L<Storable> and writes it to the
… …
889 1043
890 delete $self->{low_water_mark}; 1044 delete $self->{low_water_mark};
891 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1045 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
892} 1046}
893 1047
894=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1048=item custom write types - Package::anyevent_write_type $handle, @args
895 1049
896This 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
897Whenever 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
898reference with the handle object and the remaining arguments. 1057the handle object and the remaining arguments.
899 1058
900The 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
901be 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.
902 1062
903Note 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
904global, 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 }
905 1079
906=cut 1080=cut
907 1081
908############################################################################# 1082#############################################################################
909 1083
… …
1128 my $cb = pop; 1302 my $cb = pop;
1129 1303
1130 if (@_) { 1304 if (@_) {
1131 my $type = shift; 1305 my $type = shift;
1132 1306
1307 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1133 $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")
1134 ->($self, $cb, @_); 1309 ->($self, $cb, @_);
1135 } 1310 }
1136 1311
1137 push @{ $self->{_queue} }, $cb; 1312 push @{ $self->{_queue} }, $cb;
1138 $self->_drain_rbuf; 1313 $self->_drain_rbuf;
… …
1147 1322
1148 $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")
1149 ->($self, $cb, @_); 1324 ->($self, $cb, @_);
1150 } 1325 }
1151 1326
1152
1153 unshift @{ $self->{_queue} }, $cb; 1327 unshift @{ $self->{_queue} }, $cb;
1154 $self->_drain_rbuf; 1328 $self->_drain_rbuf;
1155} 1329}
1156 1330
1157=item $handle->push_read (type => @args, $cb) 1331=item $handle->push_read (type => @args, $cb)
1158 1332
1159=item $handle->unshift_read (type => @args, $cb) 1333=item $handle->unshift_read (type => @args, $cb)
1160 1334
1161Instead 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
1162between a number of predefined parsing formats, for chunks of data, lines 1336between a number of predefined parsing formats, for chunks of data, lines
1163etc. 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).
1164 1340
1165Predefined 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
1166drop by and tell us): 1342drop by and tell us):
1167 1343
1168=over 4 1344=over 4
… …
1408=cut 1584=cut
1409 1585
1410register_read_type json => sub { 1586register_read_type json => sub {
1411 my ($self, $cb) = @_; 1587 my ($self, $cb) = @_;
1412 1588
1413 my $json = $self->{json} ||= 1589 my $json = $self->{json} ||= json_coder;
1414 eval { require JSON::XS; JSON::XS->new->utf8 }
1415 || do { require JSON; JSON->new->utf8 };
1416 1590
1417 my $data; 1591 my $data;
1418 my $rbuf = \$self->{rbuf}; 1592 my $rbuf = \$self->{rbuf};
1419 1593
1420 sub { 1594 sub {
… …
1489 } 1663 }
1490}; 1664};
1491 1665
1492=back 1666=back
1493 1667
1494=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1668=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1495 1669
1496This 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).
1497 1675
1498Whenever 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
1499reference with the handle object, the callback and the remaining 1677handle object, the original callback and the remaining arguments.
1500arguments.
1501 1678
1502The code reference is supposed to return a callback (usually a closure) 1679The function is supposed to return a callback (usually a closure) that
1503that 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.
1504 1683
1505It should invoke the passed callback when it is done reading (remember to 1684It should invoke the original callback when it is done reading (remember
1506pass 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).
1507 1687
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>, 1688For examples, see the source of this module (F<perldoc -m
1512search for C<register_read_type>)). 1689AnyEvent::Handle>, search for C<register_read_type>)).
1513 1690
1514=item $handle->stop_read 1691=item $handle->stop_read
1515 1692
1516=item $handle->start_read 1693=item $handle->start_read
1517 1694
… …
1540 my ($self) = @_; 1717 my ($self) = @_;
1541 1718
1542 unless ($self->{_rw} || $self->{_eof}) { 1719 unless ($self->{_rw} || $self->{_eof}) {
1543 Scalar::Util::weaken $self; 1720 Scalar::Util::weaken $self;
1544 1721
1545 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1722 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1546 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1723 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1547 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;
1548 1725
1549 if ($len > 0) { 1726 if ($len > 0) {
1550 $self->{_activity} = AnyEvent->now; 1727 $self->{_activity} = $self->{_ractivity} = AE::now;
1551 1728
1552 if ($self->{tls}) { 1729 if ($self->{tls}) {
1553 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1730 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1554 1731
1555 &_dotls ($self); 1732 &_dotls ($self);
… …
1563 $self->_drain_rbuf; 1740 $self->_drain_rbuf;
1564 1741
1565 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1742 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1566 return $self->_error ($!, 1); 1743 return $self->_error ($!, 1);
1567 } 1744 }
1568 }); 1745 };
1569 } 1746 }
1570} 1747}
1571 1748
1572our $ERROR_SYSCALL; 1749our $ERROR_SYSCALL;
1573our $ERROR_WANT_READ; 1750our $ERROR_WANT_READ;
… …
1694 require Net::SSLeay; 1871 require Net::SSLeay;
1695 1872
1696 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1873 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1697 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1874 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1698 1875
1699 $tls = $self->{tls}; 1876 $tls = delete $self->{tls};
1700 $ctx = $self->{tls_ctx}; 1877 $ctx = $self->{tls_ctx};
1701 1878
1702 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1879 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1703 1880
1704 if ("HASH" eq ref $ctx) { 1881 if ("HASH" eq ref $ctx) {
… …
1791 my $fh = delete $self->{fh}; 1968 my $fh = delete $self->{fh};
1792 my $wbuf = delete $self->{wbuf}; 1969 my $wbuf = delete $self->{wbuf};
1793 1970
1794 my @linger; 1971 my @linger;
1795 1972
1796 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1973 push @linger, AE::io $fh, 1, sub {
1797 my $len = syswrite $fh, $wbuf, length $wbuf; 1974 my $len = syswrite $fh, $wbuf, length $wbuf;
1798 1975
1799 if ($len > 0) { 1976 if ($len > 0) {
1800 substr $wbuf, 0, $len, ""; 1977 substr $wbuf, 0, $len, "";
1801 } else { 1978 } else {
1802 @linger = (); # end 1979 @linger = (); # end
1803 } 1980 }
1804 }); 1981 };
1805 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1982 push @linger, AE::timer $linger, 0, sub {
1806 @linger = (); 1983 @linger = ();
1807 }); 1984 };
1808 } 1985 }
1809} 1986}
1810 1987
1811=item $handle->destroy 1988=item $handle->destroy
1812 1989

Diff Legend

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