[ViewVC] Diff of: cvs/AnyEvent/lib/AnyEvent/Handle.pm

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.161 by root, Sat Jul 25 06:16:45 2009 UTC vs.
Revision 1.185 by root, Thu Sep 3 19:48:27 2009 UTC

…		…
1	package AnyEvent::Handle;
2
3	use Scalar::Util ();
4	use Carp ();
5	use Errno qw(EAGAIN EINTR);
6
7	use AnyEvent (); BEGIN { AnyEvent::common_sense }
8	use AnyEvent::Util qw(WSAEWOULDBLOCK);
9
10	=head1 NAME	1	=head1 NAME
11		2
12	AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent	3	AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
13
14	=cut
15
16	our $VERSION = 4.86;
17		4
18	=head1 SYNOPSIS	5	=head1 SYNOPSIS
19		6
20	use AnyEvent;	7	use AnyEvent;
21	use AnyEvent::Handle;	8	use AnyEvent::Handle;
59	C<on_error> callback.	46	C<on_error> callback.
60		47
61	All callbacks will be invoked with the handle object as their first	48	All callbacks will be invoked with the handle object as their first
62	argument.	49	argument.
63		50
		51	=cut
		52
		53	package AnyEvent::Handle;
		54
		55	use Scalar::Util ();
		56	use List::Util ();
		57	use Carp ();
		58	use Errno qw(EAGAIN EINTR);
		59
		60	use AnyEvent (); BEGIN { AnyEvent::common_sense }
		61	use AnyEvent::Util qw(WSAEWOULDBLOCK);
		62
		63	our $VERSION = $AnyEvent::VERSION;
		64
		65	sub _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...
…		…
216	memory and push it into the queue, but instead only read more data from	232	memory and push it into the queue, but instead only read more data from
217	the file when the write queue becomes empty.	233	the 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
221	If non-zero, then this enables an "inactivity" timeout: whenever this many	241	If non-zero, then these enables an "inactivity" timeout: whenever this
222	seconds pass without a successful read or write on the underlying file	242	many seconds pass without a successful read or write on the underlying
223	handle, the C<on_timeout> callback will be invoked (and if that one is	243	file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
224	missing, a non-fatal C<ETIMEDOUT> error will be raised).	244	will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
		245	error will be raised).
		246
		247	There are three variants of the timeouts that work fully independent
		248	of each other, for both read and write, just read, and just write:
		249	C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
		250	C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
		251	C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
225		252
226	Note that timeout processing is also active when you currently do not have	253	Note that timeout processing is also active when you currently do not have
227	any outstanding read or write requests: If you plan to keep the connection	254	any outstanding read or write requests: If you plan to keep the connection
228	idle then you should disable the timout temporarily or ignore the timeout	255	idle then you should disable the timout temporarily or ignore the timeout
229	in the C<on_timeout> callback, in which case AnyEvent::Handle will simply	256	in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
…		…
273	accomplishd by setting this option to a true value.	300	accomplishd by setting this option to a true value.
274		301
275	The default is your opertaing system's default behaviour (most likely	302	The default is your opertaing system's default behaviour (most likely
276	enabled), this option explicitly enables or disables it, if possible.	303	enabled), this option explicitly enables or disables it, if possible.
277		304
		305	=item keepalive => <boolean>
		306
		307	Enables (default disable) the SO_KEEPALIVE option on the stream socket:
		308	normally, TCP connections have no time-out once established, so TCP
		309	conenctions, once established, can stay alive forever even when the other
		310	side has long gone. TCP keepalives are a cheap way to take down long-lived
		311	TCP connections whent he other side becomes unreachable. While the default
		312	is OS-dependent, TCP keepalives usually kick in after around two hours,
		313	and, if the other side doesn't reply, take down the TCP connection some 10
		314	to 15 minutes later.
		315
		316	It is harmless to specify this option for file handles that do not support
		317	keepalives, and enabling it on connections that are potentially long-lived
		318	is usually a good idea.
		319
		320	=item oobinline => <boolean>
		321
		322	BSD majorly fucked up the implementation of TCP urgent data. The result
		323	is that almost no OS implements TCP according to the specs, and every OS
		324	implements it slightly differently.
		325
		326	If you want to handle TCP urgent data, then setting this flag (the default
		327	is enabled) gives you the most portable way of getting urgent data, by
		328	putting it into the stream.
		329
		330	Since BSD emulation of OOB data on top of TCP's urgent data can have
		331	security implications, AnyEvent::Handle sets this flag automatically
		332	unless explicitly specified. Note that setting this flag after
		333	establishing a connection I<may> be a bit too late (data loss could
		334	already have occured on BSD systems), but at least it will protect you
		335	from most attacks.
		336
278	=item read_size => <bytes>	337	=item read_size => <bytes>
279		338
280	The default read block size (the amount of bytes this module will	339	The default read block size (the amount of bytes this module will
281	try to read during each loop iteration, which affects memory	340	try to read during each loop iteration, which affects memory
282	requirements). Default: C<8192>.	341	requirements). 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 {
…		…
474	sub _start {	533	sub _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
504	sub _error {	562	sub _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
549	Replace the current C<on_timeout> callback, or disables the callback (but	607	=item $handle->on_rtimeout ($cb)
550	not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551	argument and method.
552		608
553	=cut	609	=item $handle->on_wtimeout ($cb)
554		610
555	sub on_timeout {	611	Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556	$_[0]{on_timeout} = $_[1];	612	callback, or disables the callback (but not the timeout) if C<$cb> =
557	}	613	C<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
561	Enables or disables the current autocork behaviour (see C<autocork>	621	Enables or disables the current autocork behaviour (see C<autocork>
562	constructor argument). Changes will only take effect on the next write.	622	constructor argument). Changes will only take effect on the next write.
…		…
577	sub no_delay {	637	sub 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
		649	Enables or disables the C<keepalive> setting (see constructor argument of
		650	the same name for details).
		651
		652	=cut
		653
		654	sub 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
		666	Enables or disables the C<oobinline> setting (see constructor argument of
		667	the same name for details).
		668
		669	=cut
		670
		671	sub 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
		683	Enables or disables the C<keepalive> setting (see constructor argument of
		684	the same name for details).
		685
		686	=cut
		687
		688	sub 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
589	Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).	700	Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
590		701
591	=cut	702	=cut
…		…
602		713
603	sub on_starttls {	714	sub on_starttls {
604	$_[0]{on_stoptls} = $_[1];	715	$_[0]{on_stoptls} = $_[1];
605	}	716	}
606		717
		718	=item $handle->rbuf_max ($max_octets)
		719
		720	Configures the C<rbuf_max> setting (C<undef> disables it).
		721
		722	=cut
		723
		724	sub 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
611	Configures (or disables) the inactivity timeout.	736	Configures (or disables) the inactivity timeout.
612		737
613	=cut	738	=item $handle->timeout_reset
614		739
615	sub timeout {	740	=item $handle->rtimeout_reset
		741
		742	=item $handle->wtimeout_reset
		743
		744	Reset the activity timeout, as if data was received or sent.
		745
		746	These methods are cheap to call.
		747
		748	=cut
		749
		750	for 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
624	sub _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
737	our %WH;	889	our %WH;
738		890
		891	# deprecated
739	sub register_write_type($$) {	892	sub register_write_type($$) {
740	$WH{$_[0]} = $_[1];	893	$WH{$_[0]} = $_[1];
741	}	894	}
742		895
743	sub push_write {	896	sub 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
764	Instead of formatting your data yourself, you can also let this module do	918	Instead of formatting your data yourself, you can also let this module
765	the job by specifying a type and type-specific arguments.	919	do the job by specifying a type and type-specific arguments. You
		920	can also specify the (fully qualified) name of a package, in which
		921	case AnyEvent tries to load the package and then expects to find the
		922	C<anyevent_read_type> function inside (see "custom write types", below).
766		923
767	Predefined types are (if you have ideas for additional types, feel free to	924	Predefined types are (if you have ideas for additional types, feel free to
768	drop by and tell us):	925	drop by and tell us):
769		926
770	=over 4	927	=over 4
…		…
827	Other languages could read single lines terminated by a newline and pass	984	Other languages could read single lines terminated by a newline and pass
828	this line into their JSON decoder of choice.	985	this line into their JSON decoder of choice.
829		986
830	=cut	987	=cut
831		988
		989	sub json_coder() {
		990	eval { require JSON::XS; JSON::XS->new->utf8 }
		991	\|\| do { require JSON; JSON->new->utf8 }
		992	}
		993
832	register_write_type json => sub {	994	register_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
843	Freezes the given reference using L<Storable> and writes it to the	1004	Freezes 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
885	This function (not method) lets you add your own types to C<push_write>.	1046	Instead of one of the predefined types, you can also specify the name of
		1047	a package. AnyEvent will try to load the package and then expects to find
		1048	a function named C<anyevent_write_type> inside. If it isn't found, it
		1049	progressively tries to load the parent package until it either finds the
		1050	function (good) or runs out of packages (bad).
		1051
886	Whenever the given C<type> is used, C<push_write> will invoke the code	1052	Whenever the given C<type> is used, C<push_write> will the function with
887	reference with the handle object and the remaining arguments.	1053	the handle object and the remaining arguments.
888		1054
889	The code reference is supposed to return a single octet string that will	1055	The function is supposed to return a single octet string that will be
890	be appended to the write buffer.	1056	appended to the write buffer, so you cna mentally treat this function as a
		1057	"arguments to on-the-wire-format" converter.
891		1058
892	Note that this is a function, and all types registered this way will be	1059	Example: implement a custom write type C<join> that joins the remaining
893	global, so try to use unique names.	1060	arguments 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
982	sub _drain_rbuf {	1163	sub _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} if exists $self->{_tls_rbuf};	1173	$self->{rbuf} .= delete $self->{_tls_rbuf}
		1174	if exists $self->{_tls_rbuf};
1000		1175
1001	my $len = length $self->{rbuf};	1176	my $len = length $self->{rbuf};
1002		1177
1003	if (my $cb = shift @{ $self->{_queue} }) {	1178	if (my $cb = shift @{ $self->{_queue} }) {
1004	unless ($cb->($self)) {	1179	unless ($cb->($self)) {
1005	if ($self->{_eof}) {	1180	# no progress can be made
1006	# no progress can be made (not enough data and no data forthcoming)	1181	# (not enough data and no data forthcoming)
1007	$self->_error (Errno::EPIPE, 1), return;	1182	$self->_error (Errno::EPIPE, 1), return
1008	}	1183	if $self->{_eof};
1009		1184
1010	unshift @{ $self->{_queue} }, $cb;	1185	unshift @{ $self->{_queue} }, $cb;
1011	last;	1186	last;
1012	}	1187	}
1013	} elsif ($self->{on_read}) {	1188	} elsif ($self->{on_read}) {
…		…
1033	last;	1208	last;
1034	}	1209	}
1035	}	1210	}
1036		1211
1037	if ($self->{_eof}) {	1212	if ($self->{_eof}) {
1038	if ($self->{on_eof}) {	1213	$self->{on_eof}
1039	$self->{on_eof}($self)	1214	? $self->{on_eof}($self)
1040	} else {
1041	$self->_error (0, 1, "Unexpected end-of-file");	1215	: $self->_error (0, 1, "Unexpected end-of-file");
1042	}	1216
		1217	return;
		1218	}
		1219
		1220	if (
		1221	defined $self->{rbuf_max}
		1222	&& $self->{rbuf_max} < length $self->{rbuf}
		1223	) {
		1224	$self->_error (Errno::ENOSPC, 1), return;
1043	}	1225	}
1044		1226
1045	# may need to restart read watcher	1227	# may need to restart read watcher
1046	unless ($self->{_rw}) {	1228	unless ($self->{_rw}) {
1047	$self->start_read	1229	$self->start_read
…		…
1116	my $cb = pop;	1298	my $cb = pop;
1117		1299
1118	if (@_) {	1300	if (@_) {
1119	my $type = shift;	1301	my $type = shift;
1120		1302
		1303	$cb = ($RH{$type} \|\|= _load_func "$type\::anyevent_read_type"
1121	$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")
1122	->($self, $cb, @_);	1305	->($self, $cb, @_);
1123	}	1306	}
1124		1307
1125	push @{ $self->{_queue} }, $cb;	1308	push @{ $self->{_queue} }, $cb;
1126	$self->_drain_rbuf;	1309	$self->_drain_rbuf;
…		…
1135		1318
1136	$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")
1137	->($self, $cb, @_);	1320	->($self, $cb, @_);
1138	}	1321	}
1139		1322
1140
1141	unshift @{ $self->{_queue} }, $cb;	1323	unshift @{ $self->{_queue} }, $cb;
1142	$self->_drain_rbuf;	1324	$self->_drain_rbuf;
1143	}	1325	}
1144		1326
1145	=item $handle->push_read (type => @args, $cb)	1327	=item $handle->push_read (type => @args, $cb)
1146		1328
1147	=item $handle->unshift_read (type => @args, $cb)	1329	=item $handle->unshift_read (type => @args, $cb)
1148		1330
1149	Instead of providing a callback that parses the data itself you can chose	1331	Instead of providing a callback that parses the data itself you can chose
1150	between a number of predefined parsing formats, for chunks of data, lines	1332	between a number of predefined parsing formats, for chunks of data, lines
1151	etc.	1333	etc. You can also specify the (fully qualified) name of a package, in
		1334	which case AnyEvent tries to load the package and then expects to find the
		1335	C<anyevent_read_type> function inside (see "custom read types", below).
1152		1336
1153	Predefined types are (if you have ideas for additional types, feel free to	1337	Predefined types are (if you have ideas for additional types, feel free to
1154	drop by and tell us):	1338	drop by and tell us):
1155		1339
1156	=over 4	1340	=over 4
…		…
1396	=cut	1580	=cut
1397		1581
1398	register_read_type json => sub {	1582	register_read_type json => sub {
1399	my ($self, $cb) = @_;	1583	my ($self, $cb) = @_;
1400		1584
1401	my $json = $self->{json} \|\|=	1585	my $json = $self->{json} \|\|= json_coder;
1402	eval { require JSON::XS; JSON::XS->new->utf8 }
1403	\|\| do { require JSON; JSON->new->utf8 };
1404		1586
1405	my $data;	1587	my $data;
1406	my $rbuf = \$self->{rbuf};	1588	my $rbuf = \$self->{rbuf};
1407		1589
1408	sub {	1590	sub {
…		…
1477	}	1659	}
1478	};	1660	};
1479		1661
1480	=back	1662	=back
1481		1663
1482	=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)	1664	=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1483		1665
1484	This function (not method) lets you add your own types to C<push_read>.	1666	Instead of one of the predefined types, you can also specify the name
		1667	of a package. AnyEvent will try to load the package and then expects to
		1668	find a function named C<anyevent_read_type> inside. If it isn't found, it
		1669	progressively tries to load the parent package until it either finds the
		1670	function (good) or runs out of packages (bad).
1485		1671
1486	Whenever the given C<type> is used, C<push_read> will invoke the code	1672	Whenever this type is used, C<push_read> will invoke the function with the
1487	reference with the handle object, the callback and the remaining	1673	handle object, the original callback and the remaining arguments.
1488	arguments.
1489		1674
1490	The code reference is supposed to return a callback (usually a closure)	1675	The function is supposed to return a callback (usually a closure) that
1491	that works as a plain read callback (see C<< ->push_read ($cb) >>).	1676	works as a plain read callback (see C<< ->push_read ($cb) >>), so you can
		1677	mentally treat the function as a "configurable read type to read callback"
		1678	converter.
1492		1679
1493	It should invoke the passed callback when it is done reading (remember to	1680	It should invoke the original callback when it is done reading (remember
1494	pass C<$handle> as first argument as all other callbacks do that).	1681	to pass C<$handle> as first argument as all other callbacks do that,
		1682	although there is no strict requirement on this).
1495		1683
1496	Note that this is a function, and all types registered this way will be
1497	global, so try to use unique names.
1498
1499	For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,	1684	For examples, see the source of this module (F<perldoc -m
1500	search for C<register_read_type>)).	1685	AnyEvent::Handle>, search for C<register_read_type>)).
1501		1686
1502	=item $handle->stop_read	1687	=item $handle->stop_read
1503		1688
1504	=item $handle->start_read	1689	=item $handle->start_read
1505		1690
…		…
1528	my ($self) = @_;	1713	my ($self) = @_;
1529		1714
1530	unless ($self->{_rw} \|\| $self->{_eof}) {	1715	unless ($self->{_rw} \|\| $self->{_eof}) {
1531	Scalar::Util::weaken $self;	1716	Scalar::Util::weaken $self;
1532		1717
1533	$self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {	1718	$self->{_rw} = AE::io $self->{fh}, 0, sub {
1534	my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});	1719	my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1535	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;
1536		1721
1537	if ($len > 0) {	1722	if ($len > 0) {
1538	$self->{_activity} = AnyEvent->now;	1723	$self->{_activity} = $self->{_ractivity} = AE::now;
1539		1724
1540	if ($self->{tls}) {	1725	if ($self->{tls}) {
1541	Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);	1726	Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1542		1727
1543	&_dotls ($self);	1728	&_dotls ($self);
…		…
1551	$self->_drain_rbuf;	1736	$self->_drain_rbuf;
1552		1737
1553	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {	1738	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1554	return $self->_error ($!, 1);	1739	return $self->_error ($!, 1);
1555	}	1740	}
1556	});	1741	};
1557	}	1742	}
1558	}	1743	}
1559		1744
1560	our $ERROR_SYSCALL;	1745	our $ERROR_SYSCALL;
1561	our $ERROR_WANT_READ;	1746	our $ERROR_WANT_READ;
…		…
1682	require Net::SSLeay;	1867	require Net::SSLeay;
1683		1868
1684	$ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();	1869	$ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1685	$ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();	1870	$ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1686		1871
1687	$tls = $self->{tls};	1872	$tls = delete $self->{tls};
1688	$ctx = $self->{tls_ctx};	1873	$ctx = $self->{tls_ctx};
1689		1874
1690	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
1691		1876
1692	if ("HASH" eq ref $ctx) {	1877	if ("HASH" eq ref $ctx) {
…		…
1721	Net::SSLeay::CTX_set_mode ($tls, 1\|2);	1906	Net::SSLeay::CTX_set_mode ($tls, 1\|2);
1722		1907
1723	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());	1908	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1724	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());	1909	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1725		1910
		1911	Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
		1912
1726	Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});	1913	Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1727		1914
1728	$self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }	1915	$self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1729	if $self->{on_starttls};	1916	if $self->{on_starttls};
1730		1917
…		…
1759	my ($self) = @_;	1946	my ($self) = @_;
1760		1947
1761	return unless $self->{tls};	1948	return unless $self->{tls};
1762		1949
1763	$self->{tls_ctx}->_put_session (delete $self->{tls})	1950	$self->{tls_ctx}->_put_session (delete $self->{tls})
1764	if ref $self->{tls};	1951	if $self->{tls} > 0;
1765		1952
1766	delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};	1953	delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1767	}	1954	}
1768		1955
1769	sub DESTROY {	1956	sub DESTROY {
…		…
1777	my $fh = delete $self->{fh};	1964	my $fh = delete $self->{fh};
1778	my $wbuf = delete $self->{wbuf};	1965	my $wbuf = delete $self->{wbuf};
1779		1966
1780	my @linger;	1967	my @linger;
1781		1968
1782	push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {	1969	push @linger, AE::io $fh, 1, sub {
1783	my $len = syswrite $fh, $wbuf, length $wbuf;	1970	my $len = syswrite $fh, $wbuf, length $wbuf;
1784		1971
1785	if ($len > 0) {	1972	if ($len > 0) {
1786	substr $wbuf, 0, $len, "";	1973	substr $wbuf, 0, $len, "";
1787	} else {	1974	} else {
1788	@linger = (); # end	1975	@linger = (); # end
1789	}	1976	}
1790	});	1977	};
1791	push @linger, AnyEvent->timer (after => $linger, cb => sub {	1978	push @linger, AE::timer $linger, 0, sub {
1792	@linger = ();	1979	@linger = ();
1793	});	1980	};
1794	}	1981	}
1795	}	1982	}
1796		1983
1797	=item $handle->destroy	1984	=item $handle->destroy
1798		1985
1799	Shuts down the handle object as much as possible - this call ensures that	1986	Shuts down the handle object as much as possible - this call ensures that
1800	no further callbacks will be invoked and as many resources as possible	1987	no further callbacks will be invoked and as many resources as possible
1801	will be freed. You must not call any methods on the object afterwards.	1988	will be freed. Any method you will call on the handle object after
		1989	destroying it in this way will be silently ignored (and it will return the
		1990	empty list).
1802		1991
1803	Normally, you can just "forget" any references to an AnyEvent::Handle	1992	Normally, you can just "forget" any references to an AnyEvent::Handle
1804	object and it will simply shut down. This works in fatal error and EOF	1993	object and it will simply shut down. This works in fatal error and EOF
1805	callbacks, as well as code outside. It does I<NOT> work in a read or write	1994	callbacks, as well as code outside. It does I<NOT> work in a read or write
1806	callback, so when you want to destroy the AnyEvent::Handle object from	1995	callback, so when you want to destroy the AnyEvent::Handle object from
…		…
1820	sub destroy {	2009	sub destroy {
1821	my ($self) = @_;	2010	my ($self) = @_;
1822		2011
1823	$self->DESTROY;	2012	$self->DESTROY;
1824	%$self = ();	2013	%$self = ();
		2014	bless $self, "AnyEvent::Handle::destroyed";
		2015	}
		2016
		2017	sub AnyEvent::Handle::destroyed::AUTOLOAD {
		2018	#nop
1825	}	2019	}
1826		2020
1827	=item AnyEvent::Handle::TLS_CTX	2021	=item AnyEvent::Handle::TLS_CTX
1828		2022
1829	This function creates and returns the AnyEvent::TLS object used by default	2023	This function creates and returns the AnyEvent::TLS object used by default

Diff Legend

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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents): Revision 1.161 by root, Sat Jul 25 06:16:45 2009 UTC vs. Revision 1.185 by root, Thu Sep 3 19:48:27 2009 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.161 by root, Sat Jul 25 06:16:45 2009 UTC vs.
Revision 1.185 by root, Thu Sep 3 19:48:27 2009 UTC