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.172 by root, Wed Aug 5 20:50:27 2009 UTC vs.
Revision 1.187 by root, Tue Sep 8 00:01:12 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.901;
17 4
18=head1 SYNOPSIS 5=head1 SYNOPSIS
19 6
20 use AnyEvent; 7 use AnyEvent;
21 use AnyEvent::Handle; 8 use AnyEvent::Handle;
… …
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
… …
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 $self->_timeout; 766 delete $self->{$tw}; &$cb;
630} 767 };
631 768
769 *{"${dir}timeout_reset"} = sub {
770 $_[0]{$activity} = AE::now;
771 };
772
773 # main workhorse:
632# reset the timeout watcher, as neccessary 774 # reset the timeout watcher, as neccessary
633# also check for time-outs 775 # also check for time-outs
634sub _timeout { 776 $cb = sub {
635 my ($self) = @_; 777 my ($self) = @_;
636 778
637 if ($self->{timeout} && $self->{fh}) { 779 if ($self->{$timeout} && $self->{fh}) {
638 my $NOW = AnyEvent->now; 780 my $NOW = AE::now;
639 781
640 # when would the timeout trigger? 782 # when would the timeout trigger?
641 my $after = $self->{_activity} + $self->{timeout} - $NOW; 783 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
642 784
643 # now or in the past already? 785 # now or in the past already?
644 if ($after <= 0) { 786 if ($after <= 0) {
645 $self->{_activity} = $NOW; 787 $self->{$activity} = $NOW;
646 788
647 if ($self->{on_timeout}) { 789 if ($self->{$on_timeout}) {
648 $self->{on_timeout}($self); 790 $self->{$on_timeout}($self);
649 } else { 791 } else {
650 $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};
651 } 800 }
652 801
653 # callback could have changed timeout value, optimise 802 Scalar::Util::weaken $self;
654 return unless $self->{timeout}; 803 return unless $self; # ->error could have destroyed $self
655 804
656 # calculate new after 805 $self->{$tw} ||= AE::timer $after, 0, sub {
657 $after = $self->{timeout}; 806 delete $self->{$tw};
807 $cb->($self);
808 };
809 } else {
810 delete $self->{$tw};
658 } 811 }
659
660 Scalar::Util::weaken $self;
661 return unless $self; # ->error could have destroyed $self
662
663 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
664 delete $self->{_tw};
665 $self->_timeout;
666 });
667 } else {
668 delete $self->{_tw};
669 } 812 }
670} 813}
671 814
672############################################################################# 815#############################################################################
673 816
… …
721 my $len = syswrite $self->{fh}, $self->{wbuf}; 864 my $len = syswrite $self->{fh}, $self->{wbuf};
722 865
723 if (defined $len) { 866 if (defined $len) {
724 substr $self->{wbuf}, 0, $len, ""; 867 substr $self->{wbuf}, 0, $len, "";
725 868
726 $self->{_activity} = AnyEvent->now; 869 $self->{_activity} = $self->{_wactivity} = AE::now;
727 870
728 $self->{on_drain}($self) 871 $self->{on_drain}($self)
729 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})
730 && $self->{on_drain}; 873 && $self->{on_drain};
731 874
… …
737 880
738 # try to write data immediately 881 # try to write data immediately
739 $cb->() unless $self->{autocork}; 882 $cb->() unless $self->{autocork};
740 883
741 # if still data left in wbuf, we need to poll 884 # if still data left in wbuf, we need to poll
742 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 885 $self->{_ww} = AE::io $self->{fh}, 1, $cb
743 if length $self->{wbuf}; 886 if length $self->{wbuf};
744 }; 887 };
745} 888}
746 889
747our %WH; 890our %WH;
748 891
892# deprecated
749sub register_write_type($$) { 893sub register_write_type($$) {
750 $WH{$_[0]} = $_[1]; 894 $WH{$_[0]} = $_[1];
751} 895}
752 896
753sub push_write { 897sub push_write {
754 my $self = shift; 898 my $self = shift;
755 899
756 if (@_ > 1) { 900 if (@_ > 1) {
757 my $type = shift; 901 my $type = shift;
758 902
903 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
759 @_ = ($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")
760 ->($self, @_); 905 ->($self, @_);
761 } 906 }
762 907
763 if ($self->{tls}) { 908 if ($self->{tls}) {
764 $self->{_tls_wbuf} .= $_[0]; 909 $self->{_tls_wbuf} .= $_[0];
… …
769 } 914 }
770} 915}
771 916
772=item $handle->push_write (type => @args) 917=item $handle->push_write (type => @args)
773 918
774Instead of formatting your data yourself, you can also let this module do 919Instead of formatting your data yourself, you can also let this module
775the 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).
776 924
777Predefined 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
778drop by and tell us): 926drop by and tell us):
779 927
780=over 4 928=over 4
… …
837Other languages could read single lines terminated by a newline and pass 985Other languages could read single lines terminated by a newline and pass
838this line into their JSON decoder of choice. 986this line into their JSON decoder of choice.
839 987
840=cut 988=cut
841 989
990sub json_coder() {
991 eval { require JSON::XS; JSON::XS->new->utf8 }
992 || do { require JSON; JSON->new->utf8 }
993}
994
842register_write_type json => sub { 995register_write_type json => sub {
843 my ($self, $ref) = @_; 996 my ($self, $ref) = @_;
844 997
845 require JSON; 998 my $json = $self->{json} ||= json_coder;
846 999
847 $self->{json} ? $self->{json}->encode ($ref) 1000 $json->encode ($ref)
848 : JSON::encode_json ($ref)
849}; 1001};
850 1002
851=item storable => $reference 1003=item storable => $reference
852 1004
853Freezes the given reference using L<Storable> and writes it to the 1005Freezes the given reference using L<Storable> and writes it to the
… …
888 1040
889 delete $self->{low_water_mark}; 1041 delete $self->{low_water_mark};
890 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1042 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
891} 1043}
892 1044
893=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1045=item custom write types - Package::anyevent_write_type $handle, @args
894 1046
895This 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
896Whenever 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
897reference with the handle object and the remaining arguments. 1054the handle object and the remaining arguments.
898 1055
899The 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
900be 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.
901 1059
902Note 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
903global, 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 }
904 1076
905=cut 1077=cut
906 1078
907############################################################################# 1079#############################################################################
908 1080
… …
1127 my $cb = pop; 1299 my $cb = pop;
1128 1300
1129 if (@_) { 1301 if (@_) {
1130 my $type = shift; 1302 my $type = shift;
1131 1303
1304 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1132 $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")
1133 ->($self, $cb, @_); 1306 ->($self, $cb, @_);
1134 } 1307 }
1135 1308
1136 push @{ $self->{_queue} }, $cb; 1309 push @{ $self->{_queue} }, $cb;
1137 $self->_drain_rbuf; 1310 $self->_drain_rbuf;
… …
1146 1319
1147 $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")
1148 ->($self, $cb, @_); 1321 ->($self, $cb, @_);
1149 } 1322 }
1150 1323
1151
1152 unshift @{ $self->{_queue} }, $cb; 1324 unshift @{ $self->{_queue} }, $cb;
1153 $self->_drain_rbuf; 1325 $self->_drain_rbuf;
1154} 1326}
1155 1327
1156=item $handle->push_read (type => @args, $cb) 1328=item $handle->push_read (type => @args, $cb)
1157 1329
1158=item $handle->unshift_read (type => @args, $cb) 1330=item $handle->unshift_read (type => @args, $cb)
1159 1331
1160Instead 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
1161between a number of predefined parsing formats, for chunks of data, lines 1333between a number of predefined parsing formats, for chunks of data, lines
1162etc. 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).
1163 1337
1164Predefined 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
1165drop by and tell us): 1339drop by and tell us):
1166 1340
1167=over 4 1341=over 4
… …
1407=cut 1581=cut
1408 1582
1409register_read_type json => sub { 1583register_read_type json => sub {
1410 my ($self, $cb) = @_; 1584 my ($self, $cb) = @_;
1411 1585
1412 my $json = $self->{json} ||= 1586 my $json = $self->{json} ||= json_coder;
1413 eval { require JSON::XS; JSON::XS->new->utf8 }
1414 || do { require JSON; JSON->new->utf8 };
1415 1587
1416 my $data; 1588 my $data;
1417 my $rbuf = \$self->{rbuf}; 1589 my $rbuf = \$self->{rbuf};
1418 1590
1419 sub { 1591 sub {
… …
1488 } 1660 }
1489}; 1661};
1490 1662
1491=back 1663=back
1492 1664
1493=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1665=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1494 1666
1495This 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).
1496 1672
1497Whenever 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
1498reference with the handle object, the callback and the remaining 1674handle object, the original callback and the remaining arguments.
1499arguments.
1500 1675
1501The code reference is supposed to return a callback (usually a closure) 1676The function is supposed to return a callback (usually a closure) that
1502that 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.
1503 1680
1504It should invoke the passed callback when it is done reading (remember to 1681It should invoke the original callback when it is done reading (remember
1505pass 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).
1506 1684
1507Note that this is a function, and all types registered this way will be
1508global, so try to use unique names.
1509
1510For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1685For examples, see the source of this module (F<perldoc -m
1511search for C<register_read_type>)). 1686AnyEvent::Handle>, search for C<register_read_type>)).
1512 1687
1513=item $handle->stop_read 1688=item $handle->stop_read
1514 1689
1515=item $handle->start_read 1690=item $handle->start_read
1516 1691
… …
1539 my ($self) = @_; 1714 my ($self) = @_;
1540 1715
1541 unless ($self->{_rw} || $self->{_eof}) { 1716 unless ($self->{_rw} || $self->{_eof}) {
1542 Scalar::Util::weaken $self; 1717 Scalar::Util::weaken $self;
1543 1718
1544 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1719 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1545 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1720 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1546 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;
1547 1722
1548 if ($len > 0) { 1723 if ($len > 0) {
1549 $self->{_activity} = AnyEvent->now; 1724 $self->{_activity} = $self->{_ractivity} = AE::now;
1550 1725
1551 if ($self->{tls}) { 1726 if ($self->{tls}) {
1552 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1727 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1553 1728
1554 &_dotls ($self); 1729 &_dotls ($self);
… …
1562 $self->_drain_rbuf; 1737 $self->_drain_rbuf;
1563 1738
1564 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1739 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1565 return $self->_error ($!, 1); 1740 return $self->_error ($!, 1);
1566 } 1741 }
1567 }); 1742 };
1568 } 1743 }
1569} 1744}
1570 1745
1571our $ERROR_SYSCALL; 1746our $ERROR_SYSCALL;
1572our $ERROR_WANT_READ; 1747our $ERROR_WANT_READ;
… …
1693 require Net::SSLeay; 1868 require Net::SSLeay;
1694 1869
1695 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1870 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1696 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1871 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1697 1872
1698 $tls = $self->{tls}; 1873 $tls = delete $self->{tls};
1699 $ctx = $self->{tls_ctx}; 1874 $ctx = $self->{tls_ctx};
1700 1875
1701 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
1702 1877
1703 if ("HASH" eq ref $ctx) { 1878 if ("HASH" eq ref $ctx) {
… …
1790 my $fh = delete $self->{fh}; 1965 my $fh = delete $self->{fh};
1791 my $wbuf = delete $self->{wbuf}; 1966 my $wbuf = delete $self->{wbuf};
1792 1967
1793 my @linger; 1968 my @linger;
1794 1969
1795 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1970 push @linger, AE::io $fh, 1, sub {
1796 my $len = syswrite $fh, $wbuf, length $wbuf; 1971 my $len = syswrite $fh, $wbuf, length $wbuf;
1797 1972
1798 if ($len > 0) { 1973 if ($len > 0) {
1799 substr $wbuf, 0, $len, ""; 1974 substr $wbuf, 0, $len, "";
1800 } else { 1975 } else {
1801 @linger = (); # end 1976 @linger = (); # end
1802 } 1977 }
1803 }); 1978 };
1804 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1979 push @linger, AE::timer $linger, 0, sub {
1805 @linger = (); 1980 @linger = ();
1806 }); 1981 };
1807 } 1982 }
1808} 1983}
1809 1984
1810=item $handle->destroy 1985=item $handle->destroy
1811 1986

Diff Legend

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