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.165 by root, Mon Jul 27 22:49:23 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.87;
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
… …
602 713
603sub on_starttls { 714sub on_starttls {
604 $_[0]{on_stoptls} = $_[1]; 715 $_[0]{on_stoptls} = $_[1];
605} 716}
606 717
718=item $handle->rbuf_max ($max_octets)
719
720Configures the C<rbuf_max> setting (C<undef> disables it).
721
722=cut
723
724sub rbuf_max {
725 $_[0]{rbuf_max} = $_[1];
726}
727
607############################################################################# 728#############################################################################
608 729
609=item $handle->timeout ($seconds) 730=item $handle->timeout ($seconds)
610 731
732=item $handle->rtimeout ($seconds)
733
734=item $handle->wtimeout ($seconds)
735
611Configures (or disables) the inactivity timeout. 736Configures (or disables) the inactivity timeout.
612 737
613=cut 738=item $handle->timeout_reset
614 739
615sub 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 {
616 my ($self, $timeout) = @_; 762 my ($self, $new_value) = @_;
617 763
618 $self->{timeout} = $timeout; 764 $self->{$timeout} = $new_value;
619 $self->_timeout; 765 delete $self->{$tw}; &$cb;
620} 766 };
621 767
768 *{"${dir}timeout_reset"} = sub {
769 $_[0]{$activity} = AE::now;
770 };
771
772 # main workhorse:
622# reset the timeout watcher, as neccessary 773 # reset the timeout watcher, as neccessary
623# also check for time-outs 774 # also check for time-outs
624sub _timeout { 775 $cb = sub {
625 my ($self) = @_; 776 my ($self) = @_;
626 777
627 if ($self->{timeout} && $self->{fh}) { 778 if ($self->{$timeout} && $self->{fh}) {
628 my $NOW = AnyEvent->now; 779 my $NOW = AE::now;
629 780
630 # when would the timeout trigger? 781 # when would the timeout trigger?
631 my $after = $self->{_activity} + $self->{timeout} - $NOW; 782 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
632 783
633 # now or in the past already? 784 # now or in the past already?
634 if ($after <= 0) { 785 if ($after <= 0) {
635 $self->{_activity} = $NOW; 786 $self->{$activity} = $NOW;
636 787
637 if ($self->{on_timeout}) { 788 if ($self->{$on_timeout}) {
638 $self->{on_timeout}($self); 789 $self->{$on_timeout}($self);
639 } else { 790 } else {
640 $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};
641 } 799 }
642 800
643 # callback could have changed timeout value, optimise 801 Scalar::Util::weaken $self;
644 return unless $self->{timeout}; 802 return unless $self; # ->error could have destroyed $self
645 803
646 # calculate new after 804 $self->{$tw} ||= AE::timer $after, 0, sub {
647 $after = $self->{timeout}; 805 delete $self->{$tw};
806 $cb->($self);
807 };
808 } else {
809 delete $self->{$tw};
648 } 810 }
649
650 Scalar::Util::weaken $self;
651 return unless $self; # ->error could have destroyed $self
652
653 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
654 delete $self->{_tw};
655 $self->_timeout;
656 });
657 } else {
658 delete $self->{_tw};
659 } 811 }
660} 812}
661 813
662############################################################################# 814#############################################################################
663 815
… …
711 my $len = syswrite $self->{fh}, $self->{wbuf}; 863 my $len = syswrite $self->{fh}, $self->{wbuf};
712 864
713 if (defined $len) { 865 if (defined $len) {
714 substr $self->{wbuf}, 0, $len, ""; 866 substr $self->{wbuf}, 0, $len, "";
715 867
716 $self->{_activity} = AnyEvent->now; 868 $self->{_activity} = $self->{_wactivity} = AE::now;
717 869
718 $self->{on_drain}($self) 870 $self->{on_drain}($self)
719 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})
720 && $self->{on_drain}; 872 && $self->{on_drain};
721 873
… …
727 879
728 # try to write data immediately 880 # try to write data immediately
729 $cb->() unless $self->{autocork}; 881 $cb->() unless $self->{autocork};
730 882
731 # if still data left in wbuf, we need to poll 883 # if still data left in wbuf, we need to poll
732 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 884 $self->{_ww} = AE::io $self->{fh}, 1, $cb
733 if length $self->{wbuf}; 885 if length $self->{wbuf};
734 }; 886 };
735} 887}
736 888
737our %WH; 889our %WH;
738 890
891# deprecated
739sub register_write_type($$) { 892sub register_write_type($$) {
740 $WH{$_[0]} = $_[1]; 893 $WH{$_[0]} = $_[1];
741} 894}
742 895
743sub push_write { 896sub push_write {
744 my $self = shift; 897 my $self = shift;
745 898
746 if (@_ > 1) { 899 if (@_ > 1) {
747 my $type = shift; 900 my $type = shift;
748 901
902 @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
749 @_ = ($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")
750 ->($self, @_); 904 ->($self, @_);
751 } 905 }
752 906
753 if ($self->{tls}) { 907 if ($self->{tls}) {
754 $self->{_tls_wbuf} .= $_[0]; 908 $self->{_tls_wbuf} .= $_[0];
… …
759 } 913 }
760} 914}
761 915
762=item $handle->push_write (type => @args) 916=item $handle->push_write (type => @args)
763 917
764Instead of formatting your data yourself, you can also let this module do 918Instead of formatting your data yourself, you can also let this module
765the 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).
766 923
767Predefined 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
768drop by and tell us): 925drop by and tell us):
769 926
770=over 4 927=over 4
… …
827Other languages could read single lines terminated by a newline and pass 984Other languages could read single lines terminated by a newline and pass
828this line into their JSON decoder of choice. 985this line into their JSON decoder of choice.
829 986
830=cut 987=cut
831 988
989sub json_coder() {
990 eval { require JSON::XS; JSON::XS->new->utf8 }
991 || do { require JSON; JSON->new->utf8 }
992}
993
832register_write_type json => sub { 994register_write_type json => sub {
833 my ($self, $ref) = @_; 995 my ($self, $ref) = @_;
834 996
835 require JSON; 997 my $json = $self->{json} ||= json_coder;
836 998
837 $self->{json} ? $self->{json}->encode ($ref) 999 $json->encode ($ref)
838 : JSON::encode_json ($ref)
839}; 1000};
840 1001
841=item storable => $reference 1002=item storable => $reference
842 1003
843Freezes the given reference using L<Storable> and writes it to the 1004Freezes the given reference using L<Storable> and writes it to the
… …
878 1039
879 delete $self->{low_water_mark}; 1040 delete $self->{low_water_mark};
880 $self->on_drain (sub { shutdown $_[0]{fh}, 1 }); 1041 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
881} 1042}
882 1043
883=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 1044=item custom write types - Package::anyevent_write_type $handle, @args
884 1045
885This 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
886Whenever 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
887reference with the handle object and the remaining arguments. 1053the handle object and the remaining arguments.
888 1054
889The 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
890be 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.
891 1058
892Note 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
893global, 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 }
894 1075
895=cut 1076=cut
896 1077
897############################################################################# 1078#############################################################################
898 1079
… …
981 1162
982sub _drain_rbuf { 1163sub _drain_rbuf {
983 my ($self) = @_; 1164 my ($self) = @_;
984 1165
985 # avoid recursion 1166 # avoid recursion
986 return if exists $self->{_skip_drain_rbuf}; 1167 return if $self->{_skip_drain_rbuf};
987 local $self->{_skip_drain_rbuf} = 1; 1168 local $self->{_skip_drain_rbuf} = 1;
988
989 if (
990 defined $self->{rbuf_max}
991 && $self->{rbuf_max} < length $self->{rbuf}
992 ) {
993 $self->_error (Errno::ENOSPC, 1), return;
994 }
995 1169
996 while () { 1170 while () {
997 # we need to use a separate tls read buffer, as we must not receive data while 1171 # we need to use a separate tls read buffer, as we must not receive data while
998 # we are draining the buffer, and this can only happen with TLS. 1172 # we are draining the buffer, and this can only happen with TLS.
999 $self->{rbuf} .= delete $self->{_tls_rbuf} 1173 $self->{rbuf} .= delete $self->{_tls_rbuf}
… …
1041 : $self->_error (0, 1, "Unexpected end-of-file"); 1215 : $self->_error (0, 1, "Unexpected end-of-file");
1042 1216
1043 return; 1217 return;
1044 } 1218 }
1045 1219
1220 if (
1221 defined $self->{rbuf_max}
1222 && $self->{rbuf_max} < length $self->{rbuf}
1223 ) {
1224 $self->_error (Errno::ENOSPC, 1), return;
1225 }
1226
1046 # may need to restart read watcher 1227 # may need to restart read watcher
1047 unless ($self->{_rw}) { 1228 unless ($self->{_rw}) {
1048 $self->start_read 1229 $self->start_read
1049 if $self->{on_read} || @{ $self->{_queue} }; 1230 if $self->{on_read} || @{ $self->{_queue} };
1050 } 1231 }
… …
1117 my $cb = pop; 1298 my $cb = pop;
1118 1299
1119 if (@_) { 1300 if (@_) {
1120 my $type = shift; 1301 my $type = shift;
1121 1302
1303 $cb = ($RH{$type} ||= _load_func "$type\::anyevent_read_type"
1122 $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")
1123 ->($self, $cb, @_); 1305 ->($self, $cb, @_);
1124 } 1306 }
1125 1307
1126 push @{ $self->{_queue} }, $cb; 1308 push @{ $self->{_queue} }, $cb;
1127 $self->_drain_rbuf; 1309 $self->_drain_rbuf;
… …
1136 1318
1137 $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")
1138 ->($self, $cb, @_); 1320 ->($self, $cb, @_);
1139 } 1321 }
1140 1322
1141
1142 unshift @{ $self->{_queue} }, $cb; 1323 unshift @{ $self->{_queue} }, $cb;
1143 $self->_drain_rbuf; 1324 $self->_drain_rbuf;
1144} 1325}
1145 1326
1146=item $handle->push_read (type => @args, $cb) 1327=item $handle->push_read (type => @args, $cb)
1147 1328
1148=item $handle->unshift_read (type => @args, $cb) 1329=item $handle->unshift_read (type => @args, $cb)
1149 1330
1150Instead 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
1151between a number of predefined parsing formats, for chunks of data, lines 1332between a number of predefined parsing formats, for chunks of data, lines
1152etc. 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).
1153 1336
1154Predefined 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
1155drop by and tell us): 1338drop by and tell us):
1156 1339
1157=over 4 1340=over 4
… …
1397=cut 1580=cut
1398 1581
1399register_read_type json => sub { 1582register_read_type json => sub {
1400 my ($self, $cb) = @_; 1583 my ($self, $cb) = @_;
1401 1584
1402 my $json = $self->{json} ||= 1585 my $json = $self->{json} ||= json_coder;
1403 eval { require JSON::XS; JSON::XS->new->utf8 }
1404 || do { require JSON; JSON->new->utf8 };
1405 1586
1406 my $data; 1587 my $data;
1407 my $rbuf = \$self->{rbuf}; 1588 my $rbuf = \$self->{rbuf};
1408 1589
1409 sub { 1590 sub {
… …
1478 } 1659 }
1479}; 1660};
1480 1661
1481=back 1662=back
1482 1663
1483=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1664=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1484 1665
1485This 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).
1486 1671
1487Whenever 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
1488reference with the handle object, the callback and the remaining 1673handle object, the original callback and the remaining arguments.
1489arguments.
1490 1674
1491The code reference is supposed to return a callback (usually a closure) 1675The function is supposed to return a callback (usually a closure) that
1492that 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.
1493 1679
1494It should invoke the passed callback when it is done reading (remember to 1680It should invoke the original callback when it is done reading (remember
1495pass 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).
1496 1683
1497Note that this is a function, and all types registered this way will be
1498global, so try to use unique names.
1499
1500For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1684For examples, see the source of this module (F<perldoc -m
1501search for C<register_read_type>)). 1685AnyEvent::Handle>, search for C<register_read_type>)).
1502 1686
1503=item $handle->stop_read 1687=item $handle->stop_read
1504 1688
1505=item $handle->start_read 1689=item $handle->start_read
1506 1690
… …
1529 my ($self) = @_; 1713 my ($self) = @_;
1530 1714
1531 unless ($self->{_rw} || $self->{_eof}) { 1715 unless ($self->{_rw} || $self->{_eof}) {
1532 Scalar::Util::weaken $self; 1716 Scalar::Util::weaken $self;
1533 1717
1534 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1718 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1535 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1719 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1536 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;
1537 1721
1538 if ($len > 0) { 1722 if ($len > 0) {
1539 $self->{_activity} = AnyEvent->now; 1723 $self->{_activity} = $self->{_ractivity} = AE::now;
1540 1724
1541 if ($self->{tls}) { 1725 if ($self->{tls}) {
1542 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1726 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1543 1727
1544 &_dotls ($self); 1728 &_dotls ($self);
… …
1552 $self->_drain_rbuf; 1736 $self->_drain_rbuf;
1553 1737
1554 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1738 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1555 return $self->_error ($!, 1); 1739 return $self->_error ($!, 1);
1556 } 1740 }
1557 }); 1741 };
1558 } 1742 }
1559} 1743}
1560 1744
1561our $ERROR_SYSCALL; 1745our $ERROR_SYSCALL;
1562our $ERROR_WANT_READ; 1746our $ERROR_WANT_READ;
… …
1683 require Net::SSLeay; 1867 require Net::SSLeay;
1684 1868
1685 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1869 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1686 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1870 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1687 1871
1688 $tls = $self->{tls}; 1872 $tls = delete $self->{tls};
1689 $ctx = $self->{tls_ctx}; 1873 $ctx = $self->{tls_ctx};
1690 1874
1691 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
1692 1876
1693 if ("HASH" eq ref $ctx) { 1877 if ("HASH" eq ref $ctx) {
… …
1722 Net::SSLeay::CTX_set_mode ($tls, 1|2); 1906 Net::SSLeay::CTX_set_mode ($tls, 1|2);
1723 1907
1724 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1908 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1725 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1909 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1726 1910
1911 Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
1912
1727 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio}); 1913 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1728 1914
1729 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) } 1915 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1730 if $self->{on_starttls}; 1916 if $self->{on_starttls};
1731 1917
… …
1760 my ($self) = @_; 1946 my ($self) = @_;
1761 1947
1762 return unless $self->{tls}; 1948 return unless $self->{tls};
1763 1949
1764 $self->{tls_ctx}->_put_session (delete $self->{tls}) 1950 $self->{tls_ctx}->_put_session (delete $self->{tls})
1765 if ref $self->{tls}; 1951 if $self->{tls} > 0;
1766 1952
1767 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 1953 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1768} 1954}
1769 1955
1770sub DESTROY { 1956sub DESTROY {
… …
1778 my $fh = delete $self->{fh}; 1964 my $fh = delete $self->{fh};
1779 my $wbuf = delete $self->{wbuf}; 1965 my $wbuf = delete $self->{wbuf};
1780 1966
1781 my @linger; 1967 my @linger;
1782 1968
1783 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1969 push @linger, AE::io $fh, 1, sub {
1784 my $len = syswrite $fh, $wbuf, length $wbuf; 1970 my $len = syswrite $fh, $wbuf, length $wbuf;
1785 1971
1786 if ($len > 0) { 1972 if ($len > 0) {
1787 substr $wbuf, 0, $len, ""; 1973 substr $wbuf, 0, $len, "";
1788 } else { 1974 } else {
1789 @linger = (); # end 1975 @linger = (); # end
1790 } 1976 }
1791 }); 1977 };
1792 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1978 push @linger, AE::timer $linger, 0, sub {
1793 @linger = (); 1979 @linger = ();
1794 }); 1980 };
1795 } 1981 }
1796} 1982}
1797 1983
1798=item $handle->destroy 1984=item $handle->destroy
1799 1985

Diff Legend

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