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

Diff Legend

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