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.175 by root, Sat Aug 8 22:20:43 2009 UTC vs.
Revision 1.188 by root, Thu Sep 17 08:20:14 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} = AE::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
… …
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 = AE::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} ||= AE::timer $after, 0, 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} = AE::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
… …
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
764 if ($self->{tls}) { 908 if ($self->{tls}) {
765 $self->{_tls_wbuf} .= $_[0]; 909 $self->{_tls_wbuf} .= $_[0];
… …
770 } 914 }
771} 915}
772 916
773=item $handle->push_write (type => @args) 917=item $handle->push_write (type => @args)
774 918
775Instead of formatting your data yourself, you can also let this module do 919Instead of formatting your data yourself, you can also let this module
776the job by specifying a type and type-specific arguments. 920do the job by specifying a type and type-specific arguments. You
921can also specify the (fully qualified) name of a package, in which
922case AnyEvent tries to load the package and then expects to find the
923C<anyevent_read_type> function inside (see "custom write types", below).
777 924
778Predefined types are (if you have ideas for additional types, feel free to 925Predefined types are (if you have ideas for additional types, feel free to
779drop by and tell us): 926drop by and tell us):
780 927
781=over 4 928=over 4
… …
838Other languages could read single lines terminated by a newline and pass 985Other languages could read single lines terminated by a newline and pass
839this line into their JSON decoder of choice. 986this line into their JSON decoder of choice.
840 987
841=cut 988=cut
842 989
990sub json_coder() {
991 eval { require JSON::XS; JSON::XS->new->utf8 }
992 || do { require JSON; JSON->new->utf8 }
993}
994
843register_write_type json => sub { 995register_write_type json => sub {
844 my ($self, $ref) = @_; 996 my ($self, $ref) = @_;
845 997
846 require JSON; 998 my $json = $self->{json} ||= json_coder;
847 999
848 $self->{json} ? $self->{json}->encode ($ref) 1000 $json->encode ($ref)
849 : JSON::encode_json ($ref)
850}; 1001};
851 1002
852=item storable => $reference 1003=item storable => $reference
853 1004
854Freezes the given reference using L<Storable> and writes it to the 1005Freezes the given reference using L<Storable> and writes it to the
… …
889 1040
890 delete $self->{low_water_mark}; 1041 delete $self->{low_water_mark};
891 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1042 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
892} 1043}
893 1044
894=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1045=item custom write types - Package::anyevent_write_type $handle, @args
895 1046
896This function (not method) lets you add your own types to C<push_write>. 1047Instead of one of the predefined types, you can also specify the name of
1048a package. AnyEvent will try to load the package and then expects to find
1049a function named C<anyevent_write_type> inside. If it isn't found, it
1050progressively tries to load the parent package until it either finds the
1051function (good) or runs out of packages (bad).
1052
897Whenever the given C<type> is used, C<push_write> will invoke the code 1053Whenever the given C<type> is used, C<push_write> will the function with
898reference with the handle object and the remaining arguments. 1054the handle object and the remaining arguments.
899 1055
900The code reference is supposed to return a single octet string that will 1056The function is supposed to return a single octet string that will be
901be appended to the write buffer. 1057appended to the write buffer, so you cna mentally treat this function as a
1058"arguments to on-the-wire-format" converter.
902 1059
903Note that this is a function, and all types registered this way will be 1060Example: implement a custom write type C<join> that joins the remaining
904global, so try to use unique names. 1061arguments using the first one.
1062
1063 $handle->push_write (My::Type => " ", 1,2,3);
1064
1065 # uses the following package, which can be defined in the "My::Type" or in
1066 # the "My" modules to be auto-loaded, or just about anywhere when the
1067 # My::Type::anyevent_write_type is defined before invoking it.
1068
1069 package My::Type;
1070
1071 sub anyevent_write_type {
1072 my ($handle, $delim, @args) = @_;
1073
1074 join $delim, @args
1075 }
905 1076
906=cut 1077=cut
907 1078
908############################################################################# 1079#############################################################################
909 1080
… …
1128 my $cb = pop; 1299 my $cb = pop;
1129 1300
1130 if (@_) { 1301 if (@_) {
1131 my $type = shift; 1302 my $type = shift;
1132 1303
1304 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1133 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 1305 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
1134 ->($self, $cb, @_); 1306 ->($self, $cb, @_);
1135 } 1307 }
1136 1308
1137 push @{ $self->{_queue} }, $cb; 1309 push @{ $self->{_queue} }, $cb;
1138 $self->_drain_rbuf; 1310 $self->_drain_rbuf;
… …
1147 1319
1148 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1320 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1149 ->($self, $cb, @_); 1321 ->($self, $cb, @_);
1150 } 1322 }
1151 1323
1152
1153 unshift @{ $self->{_queue} }, $cb; 1324 unshift @{ $self->{_queue} }, $cb;
1154 $self->_drain_rbuf; 1325 $self->_drain_rbuf;
1155} 1326}
1156 1327
1157=item $handle->push_read (type => @args, $cb) 1328=item $handle->push_read (type => @args, $cb)
1158 1329
1159=item $handle->unshift_read (type => @args, $cb) 1330=item $handle->unshift_read (type => @args, $cb)
1160 1331
1161Instead of providing a callback that parses the data itself you can chose 1332Instead of providing a callback that parses the data itself you can chose
1162between a number of predefined parsing formats, for chunks of data, lines 1333between a number of predefined parsing formats, for chunks of data, lines
1163etc. 1334etc. You can also specify the (fully qualified) name of a package, in
1335which case AnyEvent tries to load the package and then expects to find the
1336C<anyevent_read_type> function inside (see "custom read types", below).
1164 1337
1165Predefined types are (if you have ideas for additional types, feel free to 1338Predefined types are (if you have ideas for additional types, feel free to
1166drop by and tell us): 1339drop by and tell us):
1167 1340
1168=over 4 1341=over 4
… …
1408=cut 1581=cut
1409 1582
1410register_read_type json => sub { 1583register_read_type json => sub {
1411 my ($self, $cb) = @_; 1584 my ($self, $cb) = @_;
1412 1585
1413 my $json = $self->{json} ||= 1586 my $json = $self->{json} ||= json_coder;
1414 eval { require JSON::XS; JSON::XS->new->utf8 }
1415 || do { require JSON; JSON->new->utf8 };
1416 1587
1417 my $data; 1588 my $data;
1418 my $rbuf = \$self->{rbuf}; 1589 my $rbuf = \$self->{rbuf};
1419 1590
1420 sub { 1591 sub {
… …
1489 } 1660 }
1490}; 1661};
1491 1662
1492=back 1663=back
1493 1664
1494=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1665=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1495 1666
1496This function (not method) lets you add your own types to C<push_read>. 1667Instead of one of the predefined types, you can also specify the name
1668of a package. AnyEvent will try to load the package and then expects to
1669find a function named C<anyevent_read_type> inside. If it isn't found, it
1670progressively tries to load the parent package until it either finds the
1671function (good) or runs out of packages (bad).
1497 1672
1498Whenever the given C<type> is used, C<push_read> will invoke the code 1673Whenever this type is used, C<push_read> will invoke the function with the
1499reference with the handle object, the callback and the remaining 1674handle object, the original callback and the remaining arguments.
1500arguments.
1501 1675
1502The code reference is supposed to return a callback (usually a closure) 1676The function is supposed to return a callback (usually a closure) that
1503that works as a plain read callback (see C<< ->push_read ($cb) >>). 1677works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
1678mentally treat the function as a "configurable read type to read callback"
1679converter.
1504 1680
1505It should invoke the passed callback when it is done reading (remember to 1681It should invoke the original callback when it is done reading (remember
1506pass C<$handle> as first argument as all other callbacks do that). 1682to pass C<$handle> as first argument as all other callbacks do that,
1683although there is no strict requirement on this).
1507 1684
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>, 1685For examples, see the source of this module (F<perldoc -m
1512search for C<register_read_type>)). 1686AnyEvent::Handle>, search for C<register_read_type>)).
1513 1687
1514=item $handle->stop_read 1688=item $handle->stop_read
1515 1689
1516=item $handle->start_read 1690=item $handle->start_read
1517 1691
… …
1545 $self->{_rw} = AE::io $self->{fh}, 0, sub { 1719 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1546 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1720 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1547 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1721 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1548 1722
1549 if ($len > 0) { 1723 if ($len > 0) {
1550 $self->{_activity} = AE::now; 1724 $self->{_activity} = $self->{_ractivity} = AE::now;
1551 1725
1552 if ($self->{tls}) { 1726 if ($self->{tls}) {
1553 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1727 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1554 1728
1555 &_dotls ($self); 1729 &_dotls ($self);
… …
1694 require Net::SSLeay; 1868 require Net::SSLeay;
1695 1869
1696 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1870 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1697 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1871 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1698 1872
1699 $tls = $self->{tls}; 1873 $tls = delete $self->{tls};
1700 $ctx = $self->{tls_ctx}; 1874 $ctx = $self->{tls_ctx};
1701 1875
1702 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1876 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1703 1877
1704 if ("HASH" eq ref $ctx) { 1878 if ("HASH" eq ref $ctx) {

Diff Legend

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