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

Diff Legend

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