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.185 by root, Thu Sep 3 19:48:27 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;
… …
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...
… …
216memory and push it into the queue, but instead only read more data from 232memory and push it into the queue, but instead only read more data from
217the file when the write queue becomes empty. 233the file when the write queue becomes empty.
218 234
219=item timeout => $fractional_seconds 235=item timeout => $fractional_seconds
220 236
237=item rtimeout => $fractional_seconds
238
239=item wtimeout => $fractional_seconds
240
221If non-zero, then this enables an "inactivity" timeout: whenever this many 241If non-zero, then these enables an "inactivity" timeout: whenever this
222seconds pass without a successful read or write on the underlying file 242many 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 243file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
224missing, a non-fatal C<ETIMEDOUT> error will be raised). 244will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
245error will be raised).
246
247There are three variants of the timeouts that work fully independent
248of each other, for both read and write, just read, and just write:
249C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
250C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
251C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
225 252
226Note that timeout processing is also active when you currently do not have 253Note 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 254any outstanding read or write requests: If you plan to keep the connection
228idle then you should disable the timout temporarily or ignore the timeout 255idle then you should disable the timout temporarily or ignore the timeout
229in the C<on_timeout> callback, in which case AnyEvent::Handle will simply 256in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
… …
273accomplishd by setting this option to a true value. 300accomplishd by setting this option to a true value.
274 301
275The default is your opertaing system's default behaviour (most likely 302The default is your opertaing system's default behaviour (most likely
276enabled), this option explicitly enables or disables it, if possible. 303enabled), this option explicitly enables or disables it, if possible.
277 304
305=item keepalive => <boolean>
306
307Enables (default disable) the SO_KEEPALIVE option on the stream socket:
308normally, TCP connections have no time-out once established, so TCP
309conenctions, once established, can stay alive forever even when the other
310side has long gone. TCP keepalives are a cheap way to take down long-lived
311TCP connections whent he other side becomes unreachable. While the default
312is OS-dependent, TCP keepalives usually kick in after around two hours,
313and, if the other side doesn't reply, take down the TCP connection some 10
314to 15 minutes later.
315
316It is harmless to specify this option for file handles that do not support
317keepalives, and enabling it on connections that are potentially long-lived
318is usually a good idea.
319
320=item oobinline => <boolean>
321
322BSD majorly fucked up the implementation of TCP urgent data. The result
323is that almost no OS implements TCP according to the specs, and every OS
324implements it slightly differently.
325
326If you want to handle TCP urgent data, then setting this flag (the default
327is enabled) gives you the most portable way of getting urgent data, by
328putting it into the stream.
329
330Since BSD emulation of OOB data on top of TCP's urgent data can have
331security implications, AnyEvent::Handle sets this flag automatically
332unless explicitly specified. Note that setting this flag after
333establishing a connection I<may> be a bit too late (data loss could
334already have occured on BSD systems), but at least it will protect you
335from most attacks.
336
278=item read_size => <bytes> 337=item read_size => <bytes>
279 338
280The default read block size (the amount of bytes this module will 339The default read block size (the amount of bytes this module will
281try to read during each loop iteration, which affects memory 340try to read during each loop iteration, which affects memory
282requirements). Default: C<8192>. 341requirements). Default: C<8192>.
… …
438 delete $self->{_skip_drain_rbuf}; 497 delete $self->{_skip_drain_rbuf};
439 $self->_start; 498 $self->_start;
440 499
441 $self->{on_connect} 500 $self->{on_connect}
442 and $self->{on_connect}($self, $host, $port, sub { 501 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)}; 502 delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
444 $self->{_skip_drain_rbuf} = 1; 503 $self->{_skip_drain_rbuf} = 1;
445 &$retry; 504 &$retry;
446 }); 505 });
447 506
448 } else { 507 } else {
… …
474sub _start { 533sub _start {
475 my ($self) = @_; 534 my ($self) = @_;
476 535
477 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 536 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
478 537
538 $self->{_activity} =
539 $self->{_ractivity} =
479 $self->{_activity} = AnyEvent->now; 540 $self->{_wactivity} = AE::now;
480 $self->_timeout;
481 541
542 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
543 $self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout};
544 $self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout};
545
482 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay}; 546 $self->no_delay (delete $self->{no_delay} ) if exists $self->{no_delay} && $self->{no_delay};
547 $self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive} && $self->{keepalive};
483 548
549 $self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
550
484 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 551 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
485 if $self->{tls}; 552 if $self->{tls};
486 553
487 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 554 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
488 555
489 $self->start_read 556 $self->start_read
490 if $self->{on_read} || @{ $self->{_queue} }; 557 if $self->{on_read} || @{ $self->{_queue} };
491 558
492 $self->_drain_wbuf; 559 $self->_drain_wbuf;
493} 560}
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 561
504sub _error { 562sub _error {
505 my ($self, $errno, $fatal, $message) = @_; 563 my ($self, $errno, $fatal, $message) = @_;
506 564
507 $! = $errno; 565 $! = $errno;
… …
544 $_[0]{on_eof} = $_[1]; 602 $_[0]{on_eof} = $_[1];
545} 603}
546 604
547=item $handle->on_timeout ($cb) 605=item $handle->on_timeout ($cb)
548 606
549Replace the current C<on_timeout> callback, or disables the callback (but 607=item $handle->on_rtimeout ($cb)
550not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551argument and method.
552 608
553=cut 609=item $handle->on_wtimeout ($cb)
554 610
555sub on_timeout { 611Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556 $_[0]{on_timeout} = $_[1]; 612callback, or disables the callback (but not the timeout) if C<$cb> =
557} 613C<undef>. See the C<timeout> constructor argument and method.
614
615=cut
616
617# see below
558 618
559=item $handle->autocork ($boolean) 619=item $handle->autocork ($boolean)
560 620
561Enables or disables the current autocork behaviour (see C<autocork> 621Enables or disables the current autocork behaviour (see C<autocork>
562constructor argument). Changes will only take effect on the next write. 622constructor argument). Changes will only take effect on the next write.
… …
577sub no_delay { 637sub no_delay {
578 $_[0]{no_delay} = $_[1]; 638 $_[0]{no_delay} = $_[1];
579 639
580 eval { 640 eval {
581 local $SIG{__DIE__}; 641 local $SIG{__DIE__};
582 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1] 642 setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
583 if $_[0]{fh}; 643 if $_[0]{fh};
584 }; 644 };
585} 645}
586 646
647=item $handle->keepalive ($boolean)
648
649Enables or disables the C<keepalive> setting (see constructor argument of
650the same name for details).
651
652=cut
653
654sub keepalive {
655 $_[0]{keepalive} = $_[1];
656
657 eval {
658 local $SIG{__DIE__};
659 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
660 if $_[0]{fh};
661 };
662}
663
664=item $handle->oobinline ($boolean)
665
666Enables or disables the C<oobinline> setting (see constructor argument of
667the same name for details).
668
669=cut
670
671sub oobinline {
672 $_[0]{oobinline} = $_[1];
673
674 eval {
675 local $SIG{__DIE__};
676 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
677 if $_[0]{fh};
678 };
679}
680
681=item $handle->keepalive ($boolean)
682
683Enables or disables the C<keepalive> setting (see constructor argument of
684the same name for details).
685
686=cut
687
688sub keepalive {
689 $_[0]{keepalive} = $_[1];
690
691 eval {
692 local $SIG{__DIE__};
693 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
694 if $_[0]{fh};
695 };
696}
697
587=item $handle->on_starttls ($cb) 698=item $handle->on_starttls ($cb)
588 699
589Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument). 700Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
590 701
591=cut 702=cut
… …
616 727
617############################################################################# 728#############################################################################
618 729
619=item $handle->timeout ($seconds) 730=item $handle->timeout ($seconds)
620 731
732=item $handle->rtimeout ($seconds)
733
734=item $handle->wtimeout ($seconds)
735
621Configures (or disables) the inactivity timeout. 736Configures (or disables) the inactivity timeout.
622 737
623=cut 738=item $handle->timeout_reset
624 739
625sub timeout { 740=item $handle->rtimeout_reset
741
742=item $handle->wtimeout_reset
743
744Reset the activity timeout, as if data was received or sent.
745
746These methods are cheap to call.
747
748=cut
749
750for my $dir ("", "r", "w") {
751 my $timeout = "${dir}timeout";
752 my $tw = "_${dir}tw";
753 my $on_timeout = "on_${dir}timeout";
754 my $activity = "_${dir}activity";
755 my $cb;
756
757 *$on_timeout = sub {
758 $_[0]{$on_timeout} = $_[1];
759 };
760
761 *$timeout = sub {
626 my ($self, $timeout) = @_; 762 my ($self, $new_value) = @_;
627 763
628 $self->{timeout} = $timeout; 764 $self->{$timeout} = $new_value;
629 delete $self->{_tw}; 765 delete $self->{$tw}; &$cb;
630 $self->_timeout; 766 };
631}
632 767
768 *{"${dir}timeout_reset"} = sub {
769 $_[0]{$activity} = AE::now;
770 };
771
772 # main workhorse:
633# reset the timeout watcher, as neccessary 773 # reset the timeout watcher, as neccessary
634# also check for time-outs 774 # also check for time-outs
635sub _timeout { 775 $cb = sub {
636 my ($self) = @_; 776 my ($self) = @_;
637 777
638 if ($self->{timeout} && $self->{fh}) { 778 if ($self->{$timeout} && $self->{fh}) {
639 my $NOW = AnyEvent->now; 779 my $NOW = AE::now;
640 780
641 # when would the timeout trigger? 781 # when would the timeout trigger?
642 my $after = $self->{_activity} + $self->{timeout} - $NOW; 782 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
643 783
644 # now or in the past already? 784 # now or in the past already?
645 if ($after <= 0) { 785 if ($after <= 0) {
646 $self->{_activity} = $NOW; 786 $self->{$activity} = $NOW;
647 787
648 if ($self->{on_timeout}) { 788 if ($self->{$on_timeout}) {
649 $self->{on_timeout}($self); 789 $self->{$on_timeout}($self);
650 } else { 790 } else {
651 $self->_error (Errno::ETIMEDOUT); 791 $self->_error (Errno::ETIMEDOUT);
792 }
793
794 # callback could have changed timeout value, optimise
795 return unless $self->{$timeout};
796
797 # calculate new after
798 $after = $self->{$timeout};
652 } 799 }
653 800
654 # callback could have changed timeout value, optimise 801 Scalar::Util::weaken $self;
655 return unless $self->{timeout}; 802 return unless $self; # ->error could have destroyed $self
656 803
657 # calculate new after 804 $self->{$tw} ||= AE::timer $after, 0, sub {
658 $after = $self->{timeout}; 805 delete $self->{$tw};
806 $cb->($self);
807 };
808 } else {
809 delete $self->{$tw};
659 } 810 }
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 } 811 }
671} 812}
672 813
673############################################################################# 814#############################################################################
674 815
… …
722 my $len = syswrite $self->{fh}, $self->{wbuf}; 863 my $len = syswrite $self->{fh}, $self->{wbuf};
723 864
724 if (defined $len) { 865 if (defined $len) {
725 substr $self->{wbuf}, 0, $len, ""; 866 substr $self->{wbuf}, 0, $len, "";
726 867
727 $self->{_activity} = AnyEvent->now; 868 $self->{_activity} = $self->{_wactivity} = AE::now;
728 869
729 $self->{on_drain}($self) 870 $self->{on_drain}($self)
730 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 871 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
731 && $self->{on_drain}; 872 && $self->{on_drain};
732 873
… …
738 879
739 # try to write data immediately 880 # try to write data immediately
740 $cb->() unless $self->{autocork}; 881 $cb->() unless $self->{autocork};
741 882
742 # if still data left in wbuf, we need to poll 883 # if still data left in wbuf, we need to poll
743 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 884 $self->{_ww} = AE::io $self->{fh}, 1, $cb
744 if length $self->{wbuf}; 885 if length $self->{wbuf};
745 }; 886 };
746} 887}
747 888
748our %WH; 889our %WH;
749 890
891# deprecated
750sub register_write_type($$) { 892sub register_write_type($$) {
751 $WH{$_[0]} = $_[1]; 893 $WH{$_[0]} = $_[1];
752} 894}
753 895
754sub push_write { 896sub push_write {
755 my $self = shift; 897 my $self = shift;
756 898
757 if (@_ > 1) { 899 if (@_ > 1) {
758 my $type = shift; 900 my $type = shift;
759 901
902 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
760 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 903 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_write")
761 ->($self, @_); 904 ->($self, @_);
762 } 905 }
763 906
764 if ($self->{tls}) { 907 if ($self->{tls}) {
765 $self->{_tls_wbuf} .= $_[0]; 908 $self->{_tls_wbuf} .= $_[0];
… …
770 } 913 }
771} 914}
772 915
773=item $handle->push_write (type => @args) 916=item $handle->push_write (type => @args)
774 917
775Instead of formatting your data yourself, you can also let this module do 918Instead of formatting your data yourself, you can also let this module
776the job by specifying a type and type-specific arguments. 919do the job by specifying a type and type-specific arguments. You
920can also specify the (fully qualified) name of a package, in which
921case AnyEvent tries to load the package and then expects to find the
922C<anyevent_read_type> function inside (see "custom write types", below).
777 923
778Predefined types are (if you have ideas for additional types, feel free to 924Predefined types are (if you have ideas for additional types, feel free to
779drop by and tell us): 925drop by and tell us):
780 926
781=over 4 927=over 4
… …
838Other languages could read single lines terminated by a newline and pass 984Other languages could read single lines terminated by a newline and pass
839this line into their JSON decoder of choice. 985this line into their JSON decoder of choice.
840 986
841=cut 987=cut
842 988
989sub json_coder() {
990 eval { require JSON::XS; JSON::XS->new->utf8 }
991 || do { require JSON; JSON->new->utf8 }
992}
993
843register_write_type json => sub { 994register_write_type json => sub {
844 my ($self, $ref) = @_; 995 my ($self, $ref) = @_;
845 996
846 require JSON; 997 my $json = $self->{json} ||= json_coder;
847 998
848 $self->{json} ? $self->{json}->encode ($ref) 999 $json->encode ($ref)
849 : JSON::encode_json ($ref)
850}; 1000};
851 1001
852=item storable => $reference 1002=item storable => $reference
853 1003
854Freezes the given reference using L<Storable> and writes it to the 1004Freezes the given reference using L<Storable> and writes it to the
… …
889 1039
890 delete $self->{low_water_mark}; 1040 delete $self->{low_water_mark};
891 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1041 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
892} 1042}
893 1043
894=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1044=item custom write types - Package::anyevent_write_type $handle, @args
895 1045
896This function (not method) lets you add your own types to C<push_write>. 1046Instead of one of the predefined types, you can also specify the name of
1047a package. AnyEvent will try to load the package and then expects to find
1048a function named C<anyevent_write_type> inside. If it isn't found, it
1049progressively tries to load the parent package until it either finds the
1050function (good) or runs out of packages (bad).
1051
897Whenever the given C<type> is used, C<push_write> will invoke the code 1052Whenever the given C<type> is used, C<push_write> will the function with
898reference with the handle object and the remaining arguments. 1053the handle object and the remaining arguments.
899 1054
900The code reference is supposed to return a single octet string that will 1055The function is supposed to return a single octet string that will be
901be appended to the write buffer. 1056appended to the write buffer, so you cna mentally treat this function as a
1057"arguments to on-the-wire-format" converter.
902 1058
903Note that this is a function, and all types registered this way will be 1059Example: implement a custom write type C<join> that joins the remaining
904global, so try to use unique names. 1060arguments using the first one.
1061
1062 $handle->push_write (My::Type => " ", 1,2,3);
1063
1064 # uses the following package, which can be defined in the "My::Type" or in
1065 # the "My" modules to be auto-loaded, or just about anywhere when the
1066 # My::Type::anyevent_write_type is defined before invoking it.
1067
1068 package My::Type;
1069
1070 sub anyevent_write_type {
1071 my ($handle, $delim, @args) = @_;
1072
1073 join $delim, @args
1074 }
905 1075
906=cut 1076=cut
907 1077
908############################################################################# 1078#############################################################################
909 1079
… …
1128 my $cb = pop; 1298 my $cb = pop;
1129 1299
1130 if (@_) { 1300 if (@_) {
1131 my $type = shift; 1301 my $type = shift;
1132 1302
1303 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1133 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 1304 or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_read")
1134 ->($self, $cb, @_); 1305 ->($self, $cb, @_);
1135 } 1306 }
1136 1307
1137 push @{ $self->{_queue} }, $cb; 1308 push @{ $self->{_queue} }, $cb;
1138 $self->_drain_rbuf; 1309 $self->_drain_rbuf;
… …
1147 1318
1148 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1319 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1149 ->($self, $cb, @_); 1320 ->($self, $cb, @_);
1150 } 1321 }
1151 1322
1152
1153 unshift @{ $self->{_queue} }, $cb; 1323 unshift @{ $self->{_queue} }, $cb;
1154 $self->_drain_rbuf; 1324 $self->_drain_rbuf;
1155} 1325}
1156 1326
1157=item $handle->push_read (type => @args, $cb) 1327=item $handle->push_read (type => @args, $cb)
1158 1328
1159=item $handle->unshift_read (type => @args, $cb) 1329=item $handle->unshift_read (type => @args, $cb)
1160 1330
1161Instead of providing a callback that parses the data itself you can chose 1331Instead of providing a callback that parses the data itself you can chose
1162between a number of predefined parsing formats, for chunks of data, lines 1332between a number of predefined parsing formats, for chunks of data, lines
1163etc. 1333etc. You can also specify the (fully qualified) name of a package, in
1334which case AnyEvent tries to load the package and then expects to find the
1335C<anyevent_read_type> function inside (see "custom read types", below).
1164 1336
1165Predefined types are (if you have ideas for additional types, feel free to 1337Predefined types are (if you have ideas for additional types, feel free to
1166drop by and tell us): 1338drop by and tell us):
1167 1339
1168=over 4 1340=over 4
… …
1408=cut 1580=cut
1409 1581
1410register_read_type json => sub { 1582register_read_type json => sub {
1411 my ($self, $cb) = @_; 1583 my ($self, $cb) = @_;
1412 1584
1413 my $json = $self->{json} ||= 1585 my $json = $self->{json} ||= json_coder;
1414 eval { require JSON::XS; JSON::XS->new->utf8 }
1415 || do { require JSON; JSON->new->utf8 };
1416 1586
1417 my $data; 1587 my $data;
1418 my $rbuf = \$self->{rbuf}; 1588 my $rbuf = \$self->{rbuf};
1419 1589
1420 sub { 1590 sub {
… …
1489 } 1659 }
1490}; 1660};
1491 1661
1492=back 1662=back
1493 1663
1494=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1664=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1495 1665
1496This function (not method) lets you add your own types to C<push_read>. 1666Instead of one of the predefined types, you can also specify the name
1667of a package. AnyEvent will try to load the package and then expects to
1668find a function named C<anyevent_read_type> inside. If it isn't found, it
1669progressively tries to load the parent package until it either finds the
1670function (good) or runs out of packages (bad).
1497 1671
1498Whenever the given C<type> is used, C<push_read> will invoke the code 1672Whenever this type is used, C<push_read> will invoke the function with the
1499reference with the handle object, the callback and the remaining 1673handle object, the original callback and the remaining arguments.
1500arguments.
1501 1674
1502The code reference is supposed to return a callback (usually a closure) 1675The function is supposed to return a callback (usually a closure) that
1503that works as a plain read callback (see C<< ->push_read ($cb) >>). 1676works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
1677mentally treat the function as a "configurable read type to read callback"
1678converter.
1504 1679
1505It should invoke the passed callback when it is done reading (remember to 1680It should invoke the original callback when it is done reading (remember
1506pass C<$handle> as first argument as all other callbacks do that). 1681to pass C<$handle> as first argument as all other callbacks do that,
1682although there is no strict requirement on this).
1507 1683
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>, 1684For examples, see the source of this module (F<perldoc -m
1512search for C<register_read_type>)). 1685AnyEvent::Handle>, search for C<register_read_type>)).
1513 1686
1514=item $handle->stop_read 1687=item $handle->stop_read
1515 1688
1516=item $handle->start_read 1689=item $handle->start_read
1517 1690
… …
1540 my ($self) = @_; 1713 my ($self) = @_;
1541 1714
1542 unless ($self->{_rw} || $self->{_eof}) { 1715 unless ($self->{_rw} || $self->{_eof}) {
1543 Scalar::Util::weaken $self; 1716 Scalar::Util::weaken $self;
1544 1717
1545 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1718 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1546 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1719 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1547 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1720 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1548 1721
1549 if ($len > 0) { 1722 if ($len > 0) {
1550 $self->{_activity} = AnyEvent->now; 1723 $self->{_activity} = $self->{_ractivity} = AE::now;
1551 1724
1552 if ($self->{tls}) { 1725 if ($self->{tls}) {
1553 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1726 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1554 1727
1555 &_dotls ($self); 1728 &_dotls ($self);
… …
1563 $self->_drain_rbuf; 1736 $self->_drain_rbuf;
1564 1737
1565 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1738 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1566 return $self->_error ($!, 1); 1739 return $self->_error ($!, 1);
1567 } 1740 }
1568 }); 1741 };
1569 } 1742 }
1570} 1743}
1571 1744
1572our $ERROR_SYSCALL; 1745our $ERROR_SYSCALL;
1573our $ERROR_WANT_READ; 1746our $ERROR_WANT_READ;
… …
1694 require Net::SSLeay; 1867 require Net::SSLeay;
1695 1868
1696 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1869 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1697 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1870 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1698 1871
1699 $tls = $self->{tls}; 1872 $tls = delete $self->{tls};
1700 $ctx = $self->{tls_ctx}; 1873 $ctx = $self->{tls_ctx};
1701 1874
1702 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1875 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1703 1876
1704 if ("HASH" eq ref $ctx) { 1877 if ("HASH" eq ref $ctx) {
… …
1791 my $fh = delete $self->{fh}; 1964 my $fh = delete $self->{fh};
1792 my $wbuf = delete $self->{wbuf}; 1965 my $wbuf = delete $self->{wbuf};
1793 1966
1794 my @linger; 1967 my @linger;
1795 1968
1796 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1969 push @linger, AE::io $fh, 1, sub {
1797 my $len = syswrite $fh, $wbuf, length $wbuf; 1970 my $len = syswrite $fh, $wbuf, length $wbuf;
1798 1971
1799 if ($len > 0) { 1972 if ($len > 0) {
1800 substr $wbuf, 0, $len, ""; 1973 substr $wbuf, 0, $len, "";
1801 } else { 1974 } else {
1802 @linger = (); # end 1975 @linger = (); # end
1803 } 1976 }
1804 }); 1977 };
1805 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1978 push @linger, AE::timer $linger, 0, sub {
1806 @linger = (); 1979 @linger = ();
1807 }); 1980 };
1808 } 1981 }
1809} 1982}
1810 1983
1811=item $handle->destroy 1984=item $handle->destroy
1812 1985

Diff Legend

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