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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.160 by root, Fri Jul 24 22:47:04 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...
…		…
100	This (rarely used) callback is called before a new connection is	116	This (rarely used) callback is called before a new connection is
101	attempted, but after the file handle has been created. It could be used to	117	attempted, but after the file handle has been created. It could be used to
102	prepare the file handle with parameters required for the actual connect	118	prepare the file handle with parameters required for the actual connect
103	(as opposed to settings that can be changed when the connection is already	119	(as opposed to settings that can be changed when the connection is already
104	established).	120	established).
		121
		122	The return value of this callback should be the connect timeout value in
		123	seconds (or C<0>, or C<undef>, or the empty list, to indicate the default
		124	timeout is to be used).
105		125
106	=item on_connect => $cb->($handle, $host, $port, $retry->())	126	=item on_connect => $cb->($handle, $host, $port, $retry->())
107		127
108	This callback is called when a connection has been successfully established.	128	This callback is called when a connection has been successfully established.
109		129
…		…
212	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
213	the file when the write queue becomes empty.	233	the file when the write queue becomes empty.
214		234
215	=item timeout => $fractional_seconds	235	=item timeout => $fractional_seconds
216		236
		237	=item rtimeout => $fractional_seconds
		238
		239	=item wtimeout => $fractional_seconds
		240
217	If non-zero, then this enables an "inactivity" timeout: whenever this many	241	If non-zero, then these enables an "inactivity" timeout: whenever this
218	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
219	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
220	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>.
221		252
222	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
223	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
224	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
225	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
…		…
269	accomplishd by setting this option to a true value.	300	accomplishd by setting this option to a true value.
270		301
271	The default is your opertaing system's default behaviour (most likely	302	The default is your opertaing system's default behaviour (most likely
272	enabled), this option explicitly enables or disables it, if possible.	303	enabled), this option explicitly enables or disables it, if possible.
273		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
274	=item read_size => <bytes>	337	=item read_size => <bytes>
275		338
276	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
277	try to read during each loop iteration, which affects memory	340	try to read during each loop iteration, which affects memory
278	requirements). Default: C<8192>.	341	requirements). Default: C<8192>.
…		…
434	delete $self->{_skip_drain_rbuf};	497	delete $self->{_skip_drain_rbuf};
435	$self->_start;	498	$self->_start;
436		499
437	$self->{on_connect}	500	$self->{on_connect}
438	and $self->{on_connect}($self, $host, $port, sub {	501	and $self->{on_connect}($self, $host, $port, sub {
439	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)};
440	$self->{_skip_drain_rbuf} = 1;	503	$self->{_skip_drain_rbuf} = 1;
441	&$retry;	504	&$retry;
442	});	505	});
443		506
444	} else {	507	} else {
445	if ($self->{on_connect_error}) {	508	if ($self->{on_connect_error}) {
446	$self->{on_connect_error}($self, "$!");	509	$self->{on_connect_error}($self, "$!");
447	$self->destroy;	510	$self->destroy;
448	} else {	511	} else {
449	$self->fatal ($!, 1);	512	$self->_error ($!, 1);
450	}	513	}
451	}	514	}
452	},	515	},
453	sub {	516	sub {
454	local $self->{fh} = $_[0];	517	local $self->{fh} = $_[0];
455		518
		519	$self->{on_prepare}
456	$self->{on_prepare}->($self)	520	? $self->{on_prepare}->($self)
457	if $self->{on_prepare};	521	: ()
458	}	522	}
459	);	523	);
460	}	524	}
461		525
462	} else {	526	} else {
…		…
469	sub _start {	533	sub _start {
470	my ($self) = @_;	534	my ($self) = @_;
471		535
472	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;	536	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
473		537
		538	$self->{_activity} =
		539	$self->{_ractivity} =
474	$self->{_activity} = AnyEvent->now;	540	$self->{_wactivity} = AE::now;
475	$self->_timeout;
476		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
477	$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};
478		548
		549	$self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
		550
479	$self->starttls (delete $self->{tls}, delete $self->{tls_ctx})	551	$self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
480	if $self->{tls};	552	if $self->{tls};
481		553
482	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};	554	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
483		555
484	$self->start_read	556	$self->start_read
485	if $self->{on_read} \|\| @{ $self->{_queue} };	557	if $self->{on_read} \|\| @{ $self->{_queue} };
486		558
487	$self->_drain_wbuf;	559	$self->_drain_wbuf;
488	}	560	}
489
490	#sub _shutdown {
491	# my ($self) = @_;
492	#
493	# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
494	# $self->{_eof} = 1; # tell starttls et. al to stop trying
495	#
496	# &_freetls;
497	#}
498		561
499	sub _error {	562	sub _error {
500	my ($self, $errno, $fatal, $message) = @_;	563	my ($self, $errno, $fatal, $message) = @_;
501		564
502	$! = $errno;	565	$! = $errno;
…		…
539	$_[0]{on_eof} = $_[1];	602	$_[0]{on_eof} = $_[1];
540	}	603	}
541		604
542	=item $handle->on_timeout ($cb)	605	=item $handle->on_timeout ($cb)
543		606
544	Replace the current C<on_timeout> callback, or disables the callback (but	607	=item $handle->on_rtimeout ($cb)
545	not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
546	argument and method.
547		608
548	=cut	609	=item $handle->on_wtimeout ($cb)
549		610
550	sub on_timeout {	611	Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
551	$_[0]{on_timeout} = $_[1];	612	callback, or disables the callback (but not the timeout) if C<$cb> =
552	}	613	C<undef>. See the C<timeout> constructor argument and method.
		614
		615	=cut
		616
		617	# see below
553		618
554	=item $handle->autocork ($boolean)	619	=item $handle->autocork ($boolean)
555		620
556	Enables or disables the current autocork behaviour (see C<autocork>	621	Enables or disables the current autocork behaviour (see C<autocork>
557	constructor argument). Changes will only take effect on the next write.	622	constructor argument). Changes will only take effect on the next write.
…		…
572	sub no_delay {	637	sub no_delay {
573	$_[0]{no_delay} = $_[1];	638	$_[0]{no_delay} = $_[1];
574		639
575	eval {	640	eval {
576	local $SIG{__DIE__};	641	local $SIG{__DIE__};
577	setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1]	642	setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
578	if $_[0]{fh};	643	if $_[0]{fh};
579	};	644	};
580	}	645	}
581		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
582	=item $handle->on_starttls ($cb)	698	=item $handle->on_starttls ($cb)
583		699
584	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).
585		701
586	=cut	702	=cut
…		…
597		713
598	sub on_starttls {	714	sub on_starttls {
599	$_[0]{on_stoptls} = $_[1];	715	$_[0]{on_stoptls} = $_[1];
600	}	716	}
601		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
602	#############################################################################	728	#############################################################################
603		729
604	=item $handle->timeout ($seconds)	730	=item $handle->timeout ($seconds)
605		731
		732	=item $handle->rtimeout ($seconds)
		733
		734	=item $handle->wtimeout ($seconds)
		735
606	Configures (or disables) the inactivity timeout.	736	Configures (or disables) the inactivity timeout.
607		737
608	=cut	738	=item $handle->timeout_reset
609		739
610	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 {
611	my ($self, $timeout) = @_;	762	my ($self, $new_value) = @_;
612		763
613	$self->{timeout} = $timeout;	764	$self->{$timeout} = $new_value;
614	$self->_timeout;	765	delete $self->{$tw}; &$cb;
615	}	766	};
616		767
		768	*{"${dir}timeout_reset"} = sub {
		769	$_[0]{$activity} = AE::now;
		770	};
		771
		772	# main workhorse:
617	# reset the timeout watcher, as neccessary	773	# reset the timeout watcher, as neccessary
618	# also check for time-outs	774	# also check for time-outs
619	sub _timeout {	775	$cb = sub {
620	my ($self) = @_;	776	my ($self) = @_;
621		777
622	if ($self->{timeout} && $self->{fh}) {	778	if ($self->{$timeout} && $self->{fh}) {
623	my $NOW = AnyEvent->now;	779	my $NOW = AE::now;
624		780
625	# when would the timeout trigger?	781	# when would the timeout trigger?
626	my $after = $self->{_activity} + $self->{timeout} - $NOW;	782	my $after = $self->{$activity} + $self->{$timeout} - $NOW;
627		783
628	# now or in the past already?	784	# now or in the past already?
629	if ($after <= 0) {	785	if ($after <= 0) {
630	$self->{_activity} = $NOW;	786	$self->{$activity} = $NOW;
631		787
632	if ($self->{on_timeout}) {	788	if ($self->{$on_timeout}) {
633	$self->{on_timeout}($self);	789	$self->{$on_timeout}($self);
634	} else {	790	} else {
635	$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};
636	}	799	}
637		800
638	# callback could have changed timeout value, optimise	801	Scalar::Util::weaken $self;
639	return unless $self->{timeout};	802	return unless $self; # ->error could have destroyed $self
640		803
641	# calculate new after	804	$self->{$tw} \|\|= AE::timer $after, 0, sub {
642	$after = $self->{timeout};	805	delete $self->{$tw};
		806	$cb->($self);
		807	};
		808	} else {
		809	delete $self->{$tw};
643	}	810	}
644
645	Scalar::Util::weaken $self;
646	return unless $self; # ->error could have destroyed $self
647
648	$self->{_tw} \|\|= AnyEvent->timer (after => $after, cb => sub {
649	delete $self->{_tw};
650	$self->_timeout;
651	});
652	} else {
653	delete $self->{_tw};
654	}	811	}
655	}	812	}
656		813
657	#############################################################################	814	#############################################################################
658		815
…		…
706	my $len = syswrite $self->{fh}, $self->{wbuf};	863	my $len = syswrite $self->{fh}, $self->{wbuf};
707		864
708	if (defined $len) {	865	if (defined $len) {
709	substr $self->{wbuf}, 0, $len, "";	866	substr $self->{wbuf}, 0, $len, "";
710		867
711	$self->{_activity} = AnyEvent->now;	868	$self->{_activity} = $self->{_wactivity} = AE::now;
712		869
713	$self->{on_drain}($self)	870	$self->{on_drain}($self)
714	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})
715	&& $self->{on_drain};	872	&& $self->{on_drain};
716		873
…		…
722		879
723	# try to write data immediately	880	# try to write data immediately
724	$cb->() unless $self->{autocork};	881	$cb->() unless $self->{autocork};
725		882
726	# if still data left in wbuf, we need to poll	883	# if still data left in wbuf, we need to poll
727	$self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)	884	$self->{_ww} = AE::io $self->{fh}, 1, $cb
728	if length $self->{wbuf};	885	if length $self->{wbuf};
729	};	886	};
730	}	887	}
731		888
732	our %WH;	889	our %WH;
733		890
		891	# deprecated
734	sub register_write_type($$) {	892	sub register_write_type($$) {
735	$WH{$_[0]} = $_[1];	893	$WH{$_[0]} = $_[1];
736	}	894	}
737		895
738	sub push_write {	896	sub push_write {
739	my $self = shift;	897	my $self = shift;
740		898
741	if (@_ > 1) {	899	if (@_ > 1) {
742	my $type = shift;	900	my $type = shift;
743		901
		902	@_ = ($WH{$type} \|\|= _load_func "$type\::anyevent_write_type"
744	@_ = ($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")
745	->($self, @_);	904	->($self, @_);
746	}	905	}
747		906
748	if ($self->{tls}) {	907	if ($self->{tls}) {
749	$self->{_tls_wbuf} .= $_[0];	908	$self->{_tls_wbuf} .= $_[0];
…		…
754	}	913	}
755	}	914	}
756		915
757	=item $handle->push_write (type => @args)	916	=item $handle->push_write (type => @args)
758		917
759	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
760	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).
761		923
762	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
763	drop by and tell us):	925	drop by and tell us):
764		926
765	=over 4	927	=over 4
…		…
822	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
823	this line into their JSON decoder of choice.	985	this line into their JSON decoder of choice.
824		986
825	=cut	987	=cut
826		988
		989	sub json_coder() {
		990	eval { require JSON::XS; JSON::XS->new->utf8 }
		991	\|\| do { require JSON; JSON->new->utf8 }
		992	}
		993
827	register_write_type json => sub {	994	register_write_type json => sub {
828	my ($self, $ref) = @_;	995	my ($self, $ref) = @_;
829		996
830	require JSON;	997	my $json = $self->{json} \|\|= json_coder;
831		998
832	$self->{json} ? $self->{json}->encode ($ref)	999	$json->encode ($ref)
833	: JSON::encode_json ($ref)
834	};	1000	};
835		1001
836	=item storable => $reference	1002	=item storable => $reference
837		1003
838	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
…		…
873		1039
874	delete $self->{low_water_mark};	1040	delete $self->{low_water_mark};
875	$self->on_drain (sub { shutdown $_[0]{fh}, 1 });	1041	$self->on_drain (sub { shutdown $_[0]{fh}, 1 });
876	}	1042	}
877		1043
878	=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)	1044	=item custom write types - Package::anyevent_write_type $handle, @args
879		1045
880	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
881	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
882	reference with the handle object and the remaining arguments.	1053	the handle object and the remaining arguments.
883		1054
884	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
885	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.
886		1058
887	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
888	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	}
889		1075
890	=cut	1076	=cut
891		1077
892	#############################################################################	1078	#############################################################################
893		1079
…		…
976		1162
977	sub _drain_rbuf {	1163	sub _drain_rbuf {
978	my ($self) = @_;	1164	my ($self) = @_;
979		1165
980	# avoid recursion	1166	# avoid recursion
981	return if exists $self->{_skip_drain_rbuf};	1167	return if $self->{_skip_drain_rbuf};
982	local $self->{_skip_drain_rbuf} = 1;	1168	local $self->{_skip_drain_rbuf} = 1;
983
984	if (
985	defined $self->{rbuf_max}
986	&& $self->{rbuf_max} < length $self->{rbuf}
987	) {
988	$self->_error (Errno::ENOSPC, 1), return;
989	}
990		1169
991	while () {	1170	while () {
992	# 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
993	# 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.
994	$self->{rbuf} .= delete $self->{_tls_rbuf} if exists $self->{_tls_rbuf};	1173	$self->{rbuf} .= delete $self->{_tls_rbuf}
		1174	if exists $self->{_tls_rbuf};
995		1175
996	my $len = length $self->{rbuf};	1176	my $len = length $self->{rbuf};
997		1177
998	if (my $cb = shift @{ $self->{_queue} }) {	1178	if (my $cb = shift @{ $self->{_queue} }) {
999	unless ($cb->($self)) {	1179	unless ($cb->($self)) {
1000	if ($self->{_eof}) {	1180	# no progress can be made
1001	# no progress can be made (not enough data and no data forthcoming)	1181	# (not enough data and no data forthcoming)
1002	$self->_error (Errno::EPIPE, 1), return;	1182	$self->_error (Errno::EPIPE, 1), return
1003	}	1183	if $self->{_eof};
1004		1184
1005	unshift @{ $self->{_queue} }, $cb;	1185	unshift @{ $self->{_queue} }, $cb;
1006	last;	1186	last;
1007	}	1187	}
1008	} elsif ($self->{on_read}) {	1188	} elsif ($self->{on_read}) {
…		…
1028	last;	1208	last;
1029	}	1209	}
1030	}	1210	}
1031		1211
1032	if ($self->{_eof}) {	1212	if ($self->{_eof}) {
1033	if ($self->{on_eof}) {	1213	$self->{on_eof}
1034	$self->{on_eof}($self)	1214	? $self->{on_eof}($self)
1035	} else {
1036	$self->_error (0, 1, "Unexpected end-of-file");	1215	: $self->_error (0, 1, "Unexpected end-of-file");
1037	}	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;
1038	}	1225	}
1039		1226
1040	# may need to restart read watcher	1227	# may need to restart read watcher
1041	unless ($self->{_rw}) {	1228	unless ($self->{_rw}) {
1042	$self->start_read	1229	$self->start_read
…		…
1111	my $cb = pop;	1298	my $cb = pop;
1112		1299
1113	if (@_) {	1300	if (@_) {
1114	my $type = shift;	1301	my $type = shift;
1115		1302
		1303	$cb = ($RH{$type} \|\|= _load_func "$type\::anyevent_read_type"
1116	$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")
1117	->($self, $cb, @_);	1305	->($self, $cb, @_);
1118	}	1306	}
1119		1307
1120	push @{ $self->{_queue} }, $cb;	1308	push @{ $self->{_queue} }, $cb;
1121	$self->_drain_rbuf;	1309	$self->_drain_rbuf;
…		…
1130		1318
1131	$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")
1132	->($self, $cb, @_);	1320	->($self, $cb, @_);
1133	}	1321	}
1134		1322
1135
1136	unshift @{ $self->{_queue} }, $cb;	1323	unshift @{ $self->{_queue} }, $cb;
1137	$self->_drain_rbuf;	1324	$self->_drain_rbuf;
1138	}	1325	}
1139		1326
1140	=item $handle->push_read (type => @args, $cb)	1327	=item $handle->push_read (type => @args, $cb)
1141		1328
1142	=item $handle->unshift_read (type => @args, $cb)	1329	=item $handle->unshift_read (type => @args, $cb)
1143		1330
1144	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
1145	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
1146	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).
1147		1336
1148	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
1149	drop by and tell us):	1338	drop by and tell us):
1150		1339
1151	=over 4	1340	=over 4
…		…
1391	=cut	1580	=cut
1392		1581
1393	register_read_type json => sub {	1582	register_read_type json => sub {
1394	my ($self, $cb) = @_;	1583	my ($self, $cb) = @_;
1395		1584
1396	my $json = $self->{json} \|\|=	1585	my $json = $self->{json} \|\|= json_coder;
1397	eval { require JSON::XS; JSON::XS->new->utf8 }
1398	\|\| do { require JSON; JSON->new->utf8 };
1399		1586
1400	my $data;	1587	my $data;
1401	my $rbuf = \$self->{rbuf};	1588	my $rbuf = \$self->{rbuf};
1402		1589
1403	sub {	1590	sub {
…		…
1472	}	1659	}
1473	};	1660	};
1474		1661
1475	=back	1662	=back
1476		1663
1477	=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)	1664	=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1478		1665
1479	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).
1480		1671
1481	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
1482	reference with the handle object, the callback and the remaining	1673	handle object, the original callback and the remaining arguments.
1483	arguments.
1484		1674
1485	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
1486	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.
1487		1679
1488	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
1489	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).
1490		1683
1491	Note that this is a function, and all types registered this way will be
1492	global, so try to use unique names.
1493
1494	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
1495	search for C<register_read_type>)).	1685	AnyEvent::Handle>, search for C<register_read_type>)).
1496		1686
1497	=item $handle->stop_read	1687	=item $handle->stop_read
1498		1688
1499	=item $handle->start_read	1689	=item $handle->start_read
1500		1690
…		…
1523	my ($self) = @_;	1713	my ($self) = @_;
1524		1714
1525	unless ($self->{_rw} \|\| $self->{_eof}) {	1715	unless ($self->{_rw} \|\| $self->{_eof}) {
1526	Scalar::Util::weaken $self;	1716	Scalar::Util::weaken $self;
1527		1717
1528	$self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {	1718	$self->{_rw} = AE::io $self->{fh}, 0, sub {
1529	my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});	1719	my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1530	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;
1531		1721
1532	if ($len > 0) {	1722	if ($len > 0) {
1533	$self->{_activity} = AnyEvent->now;	1723	$self->{_activity} = $self->{_ractivity} = AE::now;
1534		1724
1535	if ($self->{tls}) {	1725	if ($self->{tls}) {
1536	Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);	1726	Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1537		1727
1538	&_dotls ($self);	1728	&_dotls ($self);
…		…
1546	$self->_drain_rbuf;	1736	$self->_drain_rbuf;
1547		1737
1548	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {	1738	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1549	return $self->_error ($!, 1);	1739	return $self->_error ($!, 1);
1550	}	1740	}
1551	});	1741	};
1552	}	1742	}
1553	}	1743	}
1554		1744
1555	our $ERROR_SYSCALL;	1745	our $ERROR_SYSCALL;
1556	our $ERROR_WANT_READ;	1746	our $ERROR_WANT_READ;
…		…
1677	require Net::SSLeay;	1867	require Net::SSLeay;
1678		1868
1679	$ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();	1869	$ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1680	$ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();	1870	$ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1681		1871
1682	$tls = $self->{tls};	1872	$tls = delete $self->{tls};
1683	$ctx = $self->{tls_ctx};	1873	$ctx = $self->{tls_ctx};
1684		1874
1685	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
1686		1876
1687	if ("HASH" eq ref $ctx) {	1877	if ("HASH" eq ref $ctx) {
…		…
1716	Net::SSLeay::CTX_set_mode ($tls, 1\|2);	1906	Net::SSLeay::CTX_set_mode ($tls, 1\|2);
1717		1907
1718	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());	1908	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1719	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());	1909	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1720		1910
		1911	Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
		1912
1721	Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});	1913	Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1722		1914
1723	$self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }	1915	$self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1724	if $self->{on_starttls};	1916	if $self->{on_starttls};
1725		1917
…		…
1754	my ($self) = @_;	1946	my ($self) = @_;
1755		1947
1756	return unless $self->{tls};	1948	return unless $self->{tls};
1757		1949
1758	$self->{tls_ctx}->_put_session (delete $self->{tls})	1950	$self->{tls_ctx}->_put_session (delete $self->{tls})
1759	if ref $self->{tls};	1951	if $self->{tls} > 0;
1760		1952
1761	delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};	1953	delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1762	}	1954	}
1763		1955
1764	sub DESTROY {	1956	sub DESTROY {
…		…
1772	my $fh = delete $self->{fh};	1964	my $fh = delete $self->{fh};
1773	my $wbuf = delete $self->{wbuf};	1965	my $wbuf = delete $self->{wbuf};
1774		1966
1775	my @linger;	1967	my @linger;
1776		1968
1777	push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {	1969	push @linger, AE::io $fh, 1, sub {
1778	my $len = syswrite $fh, $wbuf, length $wbuf;	1970	my $len = syswrite $fh, $wbuf, length $wbuf;
1779		1971
1780	if ($len > 0) {	1972	if ($len > 0) {
1781	substr $wbuf, 0, $len, "";	1973	substr $wbuf, 0, $len, "";
1782	} else {	1974	} else {
1783	@linger = (); # end	1975	@linger = (); # end
1784	}	1976	}
1785	});	1977	};
1786	push @linger, AnyEvent->timer (after => $linger, cb => sub {	1978	push @linger, AE::timer $linger, 0, sub {
1787	@linger = ();	1979	@linger = ();
1788	});	1980	};
1789	}	1981	}
1790	}	1982	}
1791		1983
1792	=item $handle->destroy	1984	=item $handle->destroy
1793		1985
1794	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
1795	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
1796	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).
1797		1991
1798	Normally, you can just "forget" any references to an AnyEvent::Handle	1992	Normally, you can just "forget" any references to an AnyEvent::Handle
1799	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
1800	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
1801	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
…		…
1815	sub destroy {	2009	sub destroy {
1816	my ($self) = @_;	2010	my ($self) = @_;
1817		2011
1818	$self->DESTROY;	2012	$self->DESTROY;
1819	%$self = ();	2013	%$self = ();
		2014	bless $self, "AnyEvent::Handle::destroyed";
		2015	}
		2016
		2017	sub AnyEvent::Handle::destroyed::AUTOLOAD {
		2018	#nop
1820	}	2019	}
1821		2020
1822	=item AnyEvent::Handle::TLS_CTX	2021	=item AnyEvent::Handle::TLS_CTX
1823		2022
1824	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.160 by root, Fri Jul 24 22:47:04 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.160 by root, Fri Jul 24 22:47:04 2009 UTC vs.
Revision 1.185 by root, Thu Sep 3 19:48:27 2009 UTC