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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.159 by root, Fri Jul 24 12:35:58 2009 UTC vs.
Revision 1.184 by root, Thu Sep 3 13:14:38 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
64	=head1 METHODS	65	=head1 METHODS
65		66
66	=over 4	67	=over 4
67		68
68	=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...	69	=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...
…		…
83	Try to connect to the specified host and service (port), using	84	Try to connect to the specified host and service (port), using
84	C<AnyEvent::Socket::tcp_connect>. The C<$host> additionally becomes the	85	C<AnyEvent::Socket::tcp_connect>. The C<$host> additionally becomes the
85	default C<peername>.	86	default C<peername>.
86		87
87	You have to specify either this parameter, or C<fh>, above.	88	You have to specify either this parameter, or C<fh>, above.
		89
		90	It is possible to push requests on the read and write queues, and modify
		91	properties of the stream, even while AnyEvent::Handle is connecting.
88		92
89	When this parameter is specified, then the C<on_prepare>,	93	When this parameter is specified, then the C<on_prepare>,
90	C<on_connect_error> and C<on_connect> callbacks will be called under the	94	C<on_connect_error> and C<on_connect> callbacks will be called under the
91	appropriate circumstances:	95	appropriate circumstances:
92		96
…		…
97	This (rarely used) callback is called before a new connection is	101	This (rarely used) callback is called before a new connection is
98	attempted, but after the file handle has been created. It could be used to	102	attempted, but after the file handle has been created. It could be used to
99	prepare the file handle with parameters required for the actual connect	103	prepare the file handle with parameters required for the actual connect
100	(as opposed to settings that can be changed when the connection is already	104	(as opposed to settings that can be changed when the connection is already
101	established).	105	established).
		106
		107	The return value of this callback should be the connect timeout value in
		108	seconds (or C<0>, or C<undef>, or the empty list, to indicate the default
		109	timeout is to be used).
102		110
103	=item on_connect => $cb->($handle, $host, $port, $retry->())	111	=item on_connect => $cb->($handle, $host, $port, $retry->())
104		112
105	This callback is called when a connection has been successfully established.	113	This callback is called when a connection has been successfully established.
106		114
…		…
209	memory and push it into the queue, but instead only read more data from	217	memory and push it into the queue, but instead only read more data from
210	the file when the write queue becomes empty.	218	the file when the write queue becomes empty.
211		219
212	=item timeout => $fractional_seconds	220	=item timeout => $fractional_seconds
213		221
		222	=item rtimeout => $fractional_seconds
		223
		224	=item wtimeout => $fractional_seconds
		225
214	If non-zero, then this enables an "inactivity" timeout: whenever this many	226	If non-zero, then these enables an "inactivity" timeout: whenever this
215	seconds pass without a successful read or write on the underlying file	227	many seconds pass without a successful read or write on the underlying
216	handle, the C<on_timeout> callback will be invoked (and if that one is	228	file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
217	missing, a non-fatal C<ETIMEDOUT> error will be raised).	229	will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
		230	error will be raised).
		231
		232	There are three variants of the timeouts that work fully independent
		233	of each other, for both read and write, just read, and just write:
		234	C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
		235	C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
		236	C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
218		237
219	Note that timeout processing is also active when you currently do not have	238	Note that timeout processing is also active when you currently do not have
220	any outstanding read or write requests: If you plan to keep the connection	239	any outstanding read or write requests: If you plan to keep the connection
221	idle then you should disable the timout temporarily or ignore the timeout	240	idle then you should disable the timout temporarily or ignore the timeout
222	in the C<on_timeout> callback, in which case AnyEvent::Handle will simply	241	in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
…		…
266	accomplishd by setting this option to a true value.	285	accomplishd by setting this option to a true value.
267		286
268	The default is your opertaing system's default behaviour (most likely	287	The default is your opertaing system's default behaviour (most likely
269	enabled), this option explicitly enables or disables it, if possible.	288	enabled), this option explicitly enables or disables it, if possible.
270		289
		290	=item keepalive => <boolean>
		291
		292	Enables (default disable) the SO_KEEPALIVE option on the stream socket:
		293	normally, TCP connections have no time-out once established, so TCP
		294	conenctions, once established, can stay alive forever even when the other
		295	side has long gone. TCP keepalives are a cheap way to take down long-lived
		296	TCP connections whent he other side becomes unreachable. While the default
		297	is OS-dependent, TCP keepalives usually kick in after around two hours,
		298	and, if the other side doesn't reply, take down the TCP connection some 10
		299	to 15 minutes later.
		300
		301	It is harmless to specify this option for file handles that do not support
		302	keepalives, and enabling it on connections that are potentially long-lived
		303	is usually a good idea.
		304
		305	=item oobinline => <boolean>
		306
		307	BSD majorly fucked up the implementation of TCP urgent data. The result
		308	is that almost no OS implements TCP according to the specs, and every OS
		309	implements it slightly differently.
		310
		311	If you want to handle TCP urgent data, then setting this flag (the default
		312	is enabled) gives you the most portable way of getting urgent data, by
		313	putting it into the stream.
		314
		315	Since BSD emulation of OOB data on top of TCP's urgent data can have
		316	security implications, AnyEvent::Handle sets this flag automatically
		317	unless explicitly specified. Note that setting this flag after
		318	establishing a connection I<may> be a bit too late (data loss could
		319	already have occured on BSD systems), but at least it will protect you
		320	from most attacks.
		321
271	=item read_size => <bytes>	322	=item read_size => <bytes>
272		323
273	The default read block size (the amount of bytes this module will	324	The default read block size (the amount of bytes this module will
274	try to read during each loop iteration, which affects memory	325	try to read during each loop iteration, which affects memory
275	requirements). Default: C<8192>.	326	requirements). Default: C<8192>.
…		…
431	delete $self->{_skip_drain_rbuf};	482	delete $self->{_skip_drain_rbuf};
432	$self->_start;	483	$self->_start;
433		484
434	$self->{on_connect}	485	$self->{on_connect}
435	and $self->{on_connect}($self, $host, $port, sub {	486	and $self->{on_connect}($self, $host, $port, sub {
436	delete @$self{qw(fh _tw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};	487	delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
437	$self->{_skip_drain_rbuf} = 1;	488	$self->{_skip_drain_rbuf} = 1;
438	&$retry;	489	&$retry;
439	});	490	});
440		491
441	} else {	492	} else {
442	if ($self->{on_connect_error}) {	493	if ($self->{on_connect_error}) {
443	$self->{on_connect_error}($self, "$!");	494	$self->{on_connect_error}($self, "$!");
444	$self->destroy;	495	$self->destroy;
445	} else {	496	} else {
446	$self->fatal ($!, 1);	497	$self->_error ($!, 1);
447	}	498	}
448	}	499	}
449	},	500	},
450	sub {	501	sub {
451	local $self->{fh} = $_[0];	502	local $self->{fh} = $_[0];
452		503
		504	$self->{on_prepare}
453	$self->{on_prepare}->($self)	505	? $self->{on_prepare}->($self)
454	if $self->{on_prepare};	506	: ()
455	}	507	}
456	);	508	);
457	}	509	}
458		510
459	} else {	511	} else {
…		…
466	sub _start {	518	sub _start {
467	my ($self) = @_;	519	my ($self) = @_;
468		520
469	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;	521	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
470		522
		523	$self->{_activity} =
		524	$self->{_ractivity} =
471	$self->{_activity} = AnyEvent->now;	525	$self->{_wactivity} = AE::now;
472	$self->_timeout;
473		526
		527	$self->timeout (delete $self->{timeout} ) if $self->{timeout};
		528	$self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout};
		529	$self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout};
		530
474	$self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};	531	$self->no_delay (delete $self->{no_delay} ) if exists $self->{no_delay} && $self->{no_delay};
		532	$self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive} && $self->{keepalive};
475		533
		534	$self->oobinline (exists $self->{oobinline} ? delete $self->{oobinline} : 1);
		535
476	$self->starttls (delete $self->{tls}, delete $self->{tls_ctx})	536	$self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
477	if $self->{tls};	537	if $self->{tls};
478		538
479	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};	539	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
480		540
481	$self->start_read	541	$self->start_read
482	if $self->{on_read} \|\| @{ $self->{_queue} };	542	if $self->{on_read} \|\| @{ $self->{_queue} };
483	}
484		543
485	#sub _shutdown {	544	$self->_drain_wbuf;
486	# my ($self) = @_;	545	}
487	#
488	# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
489	# $self->{_eof} = 1; # tell starttls et. al to stop trying
490	#
491	# &_freetls;
492	#}
493		546
494	sub _error {	547	sub _error {
495	my ($self, $errno, $fatal, $message) = @_;	548	my ($self, $errno, $fatal, $message) = @_;
496		549
497	$! = $errno;	550	$! = $errno;
…		…
534	$_[0]{on_eof} = $_[1];	587	$_[0]{on_eof} = $_[1];
535	}	588	}
536		589
537	=item $handle->on_timeout ($cb)	590	=item $handle->on_timeout ($cb)
538		591
539	Replace the current C<on_timeout> callback, or disables the callback (but	592	=item $handle->on_rtimeout ($cb)
540	not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
541	argument and method.
542		593
543	=cut	594	=item $handle->on_wtimeout ($cb)
544		595
545	sub on_timeout {	596	Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
546	$_[0]{on_timeout} = $_[1];	597	callback, or disables the callback (but not the timeout) if C<$cb> =
547	}	598	C<undef>. See the C<timeout> constructor argument and method.
		599
		600	=cut
		601
		602	# see below
548		603
549	=item $handle->autocork ($boolean)	604	=item $handle->autocork ($boolean)
550		605
551	Enables or disables the current autocork behaviour (see C<autocork>	606	Enables or disables the current autocork behaviour (see C<autocork>
552	constructor argument). Changes will only take effect on the next write.	607	constructor argument). Changes will only take effect on the next write.
…		…
567	sub no_delay {	622	sub no_delay {
568	$_[0]{no_delay} = $_[1];	623	$_[0]{no_delay} = $_[1];
569		624
570	eval {	625	eval {
571	local $SIG{__DIE__};	626	local $SIG{__DIE__};
572	setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1]	627	setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
573	if $_[0]{fh};	628	if $_[0]{fh};
574	};	629	};
575	}	630	}
576		631
		632	=item $handle->keepalive ($boolean)
		633
		634	Enables or disables the C<keepalive> setting (see constructor argument of
		635	the same name for details).
		636
		637	=cut
		638
		639	sub keepalive {
		640	$_[0]{keepalive} = $_[1];
		641
		642	eval {
		643	local $SIG{__DIE__};
		644	setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
		645	if $_[0]{fh};
		646	};
		647	}
		648
		649	=item $handle->oobinline ($boolean)
		650
		651	Enables or disables the C<oobinline> setting (see constructor argument of
		652	the same name for details).
		653
		654	=cut
		655
		656	sub oobinline {
		657	$_[0]{oobinline} = $_[1];
		658
		659	eval {
		660	local $SIG{__DIE__};
		661	setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
		662	if $_[0]{fh};
		663	};
		664	}
		665
		666	=item $handle->keepalive ($boolean)
		667
		668	Enables or disables the C<keepalive> setting (see constructor argument of
		669	the same name for details).
		670
		671	=cut
		672
		673	sub keepalive {
		674	$_[0]{keepalive} = $_[1];
		675
		676	eval {
		677	local $SIG{__DIE__};
		678	setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
		679	if $_[0]{fh};
		680	};
		681	}
		682
577	=item $handle->on_starttls ($cb)	683	=item $handle->on_starttls ($cb)
578		684
579	Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).	685	Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
580		686
581	=cut	687	=cut
…		…
592		698
593	sub on_starttls {	699	sub on_starttls {
594	$_[0]{on_stoptls} = $_[1];	700	$_[0]{on_stoptls} = $_[1];
595	}	701	}
596		702
		703	=item $handle->rbuf_max ($max_octets)
		704
		705	Configures the C<rbuf_max> setting (C<undef> disables it).
		706
		707	=cut
		708
		709	sub rbuf_max {
		710	$_[0]{rbuf_max} = $_[1];
		711	}
		712
597	#############################################################################	713	#############################################################################
598		714
599	=item $handle->timeout ($seconds)	715	=item $handle->timeout ($seconds)
600		716
		717	=item $handle->rtimeout ($seconds)
		718
		719	=item $handle->wtimeout ($seconds)
		720
601	Configures (or disables) the inactivity timeout.	721	Configures (or disables) the inactivity timeout.
602		722
603	=cut	723	=item $handle->timeout_reset
604		724
605	sub timeout {	725	=item $handle->rtimeout_reset
		726
		727	=item $handle->wtimeout_reset
		728
		729	Reset the activity timeout, as if data was received or sent.
		730
		731	These methods are cheap to call.
		732
		733	=cut
		734
		735	for my $dir ("", "r", "w") {
		736	my $timeout = "${dir}timeout";
		737	my $tw = "_${dir}tw";
		738	my $on_timeout = "on_${dir}timeout";
		739	my $activity = "_${dir}activity";
		740	my $cb;
		741
		742	*$on_timeout = sub {
		743	$_[0]{$on_timeout} = $_[1];
		744	};
		745
		746	*$timeout = sub {
606	my ($self, $timeout) = @_;	747	my ($self, $new_value) = @_;
607		748
608	$self->{timeout} = $timeout;	749	$self->{$timeout} = $new_value;
609	$self->_timeout;	750	delete $self->{$tw}; &$cb;
610	}	751	};
611		752
		753	*{"${dir}timeout_reset"} = sub {
		754	$_[0]{$activity} = AE::now;
		755	};
		756
		757	# main workhorse:
612	# reset the timeout watcher, as neccessary	758	# reset the timeout watcher, as neccessary
613	# also check for time-outs	759	# also check for time-outs
614	sub _timeout {	760	$cb = sub {
615	my ($self) = @_;	761	my ($self) = @_;
616		762
617	if ($self->{timeout} && $self->{fh}) {	763	if ($self->{$timeout} && $self->{fh}) {
618	my $NOW = AnyEvent->now;	764	my $NOW = AE::now;
619		765
620	# when would the timeout trigger?	766	# when would the timeout trigger?
621	my $after = $self->{_activity} + $self->{timeout} - $NOW;	767	my $after = $self->{$activity} + $self->{$timeout} - $NOW;
622		768
623	# now or in the past already?	769	# now or in the past already?
624	if ($after <= 0) {	770	if ($after <= 0) {
625	$self->{_activity} = $NOW;	771	$self->{$activity} = $NOW;
626		772
627	if ($self->{on_timeout}) {	773	if ($self->{$on_timeout}) {
628	$self->{on_timeout}($self);	774	$self->{$on_timeout}($self);
629	} else {	775	} else {
630	$self->_error (Errno::ETIMEDOUT);	776	$self->_error (Errno::ETIMEDOUT);
		777	}
		778
		779	# callback could have changed timeout value, optimise
		780	return unless $self->{$timeout};
		781
		782	# calculate new after
		783	$after = $self->{$timeout};
631	}	784	}
632		785
633	# callback could have changed timeout value, optimise	786	Scalar::Util::weaken $self;
634	return unless $self->{timeout};	787	return unless $self; # ->error could have destroyed $self
635		788
636	# calculate new after	789	$self->{$tw} \|\|= AE::timer $after, 0, sub {
637	$after = $self->{timeout};	790	delete $self->{$tw};
		791	$cb->($self);
		792	};
		793	} else {
		794	delete $self->{$tw};
638	}	795	}
639
640	Scalar::Util::weaken $self;
641	return unless $self; # ->error could have destroyed $self
642
643	$self->{_tw} \|\|= AnyEvent->timer (after => $after, cb => sub {
644	delete $self->{_tw};
645	$self->_timeout;
646	});
647	} else {
648	delete $self->{_tw};
649	}	796	}
650	}	797	}
651		798
652	#############################################################################	799	#############################################################################
653		800
…		…
701	my $len = syswrite $self->{fh}, $self->{wbuf};	848	my $len = syswrite $self->{fh}, $self->{wbuf};
702		849
703	if (defined $len) {	850	if (defined $len) {
704	substr $self->{wbuf}, 0, $len, "";	851	substr $self->{wbuf}, 0, $len, "";
705		852
706	$self->{_activity} = AnyEvent->now;	853	$self->{_activity} = $self->{_wactivity} = AE::now;
707		854
708	$self->{on_drain}($self)	855	$self->{on_drain}($self)
709	if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})	856	if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
710	&& $self->{on_drain};	857	&& $self->{on_drain};
711		858
…		…
717		864
718	# try to write data immediately	865	# try to write data immediately
719	$cb->() unless $self->{autocork};	866	$cb->() unless $self->{autocork};
720		867
721	# if still data left in wbuf, we need to poll	868	# if still data left in wbuf, we need to poll
722	$self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)	869	$self->{_ww} = AE::io $self->{fh}, 1, $cb
723	if length $self->{wbuf};	870	if length $self->{wbuf};
724	};	871	};
725	}	872	}
726		873
727	our %WH;	874	our %WH;
…		…
740	->($self, @_);	887	->($self, @_);
741	}	888	}
742		889
743	if ($self->{tls}) {	890	if ($self->{tls}) {
744	$self->{_tls_wbuf} .= $_[0];	891	$self->{_tls_wbuf} .= $_[0];
745		892	&_dotls ($self) if $self->{fh};
746	&_dotls ($self);
747	} else {	893	} else {
748	$self->{wbuf} .= $_[0];	894	$self->{wbuf} .= $_[0];
749	$self->_drain_wbuf if $self->{fh};	895	$self->_drain_wbuf if $self->{fh};
750	}	896	}
751	}	897	}
752		898
753	=item $handle->push_write (type => @args)	899	=item $handle->push_write (type => @args)
…		…
818	Other languages could read single lines terminated by a newline and pass	964	Other languages could read single lines terminated by a newline and pass
819	this line into their JSON decoder of choice.	965	this line into their JSON decoder of choice.
820		966
821	=cut	967	=cut
822		968
		969	sub json_coder() {
		970	eval { require JSON::XS; JSON::XS->new->utf8 }
		971	\|\| do { require JSON; JSON->new->utf8 }
		972	}
		973
823	register_write_type json => sub {	974	register_write_type json => sub {
824	my ($self, $ref) = @_;	975	my ($self, $ref) = @_;
825		976
826	require JSON;	977	my $json = $self->{json} \|\|= json_coder;
827		978
828	$self->{json} ? $self->{json}->encode ($ref)	979	$json->encode ($ref)
829	: JSON::encode_json ($ref)
830	};	980	};
831		981
832	=item storable => $reference	982	=item storable => $reference
833		983
834	Freezes the given reference using L<Storable> and writes it to the	984	Freezes the given reference using L<Storable> and writes it to the
…		…
972		1122
973	sub _drain_rbuf {	1123	sub _drain_rbuf {
974	my ($self) = @_;	1124	my ($self) = @_;
975		1125
976	# avoid recursion	1126	# avoid recursion
977	return if exists $self->{_skip_drain_rbuf};	1127	return if $self->{_skip_drain_rbuf};
978	local $self->{_skip_drain_rbuf} = 1;	1128	local $self->{_skip_drain_rbuf} = 1;
979
980	if (
981	defined $self->{rbuf_max}
982	&& $self->{rbuf_max} < length $self->{rbuf}
983	) {
984	$self->_error (Errno::ENOSPC, 1), return;
985	}
986		1129
987	while () {	1130	while () {
988	# we need to use a separate tls read buffer, as we must not receive data while	1131	# we need to use a separate tls read buffer, as we must not receive data while
989	# we are draining the buffer, and this can only happen with TLS.	1132	# we are draining the buffer, and this can only happen with TLS.
990	$self->{rbuf} .= delete $self->{_tls_rbuf} if exists $self->{_tls_rbuf};	1133	$self->{rbuf} .= delete $self->{_tls_rbuf}
		1134	if exists $self->{_tls_rbuf};
991		1135
992	my $len = length $self->{rbuf};	1136	my $len = length $self->{rbuf};
993		1137
994	if (my $cb = shift @{ $self->{_queue} }) {	1138	if (my $cb = shift @{ $self->{_queue} }) {
995	unless ($cb->($self)) {	1139	unless ($cb->($self)) {
996	if ($self->{_eof}) {	1140	# no progress can be made
997	# no progress can be made (not enough data and no data forthcoming)	1141	# (not enough data and no data forthcoming)
998	$self->_error (Errno::EPIPE, 1), return;	1142	$self->_error (Errno::EPIPE, 1), return
999	}	1143	if $self->{_eof};
1000		1144
1001	unshift @{ $self->{_queue} }, $cb;	1145	unshift @{ $self->{_queue} }, $cb;
1002	last;	1146	last;
1003	}	1147	}
1004	} elsif ($self->{on_read}) {	1148	} elsif ($self->{on_read}) {
…		…
1024	last;	1168	last;
1025	}	1169	}
1026	}	1170	}
1027		1171
1028	if ($self->{_eof}) {	1172	if ($self->{_eof}) {
1029	if ($self->{on_eof}) {	1173	$self->{on_eof}
1030	$self->{on_eof}($self)	1174	? $self->{on_eof}($self)
1031	} else {
1032	$self->_error (0, 1, "Unexpected end-of-file");	1175	: $self->_error (0, 1, "Unexpected end-of-file");
1033	}	1176
		1177	return;
		1178	}
		1179
		1180	if (
		1181	defined $self->{rbuf_max}
		1182	&& $self->{rbuf_max} < length $self->{rbuf}
		1183	) {
		1184	$self->_error (Errno::ENOSPC, 1), return;
1034	}	1185	}
1035		1186
1036	# may need to restart read watcher	1187	# may need to restart read watcher
1037	unless ($self->{_rw}) {	1188	unless ($self->{_rw}) {
1038	$self->start_read	1189	$self->start_read
…		…
1125	my $type = shift;	1276	my $type = shift;
1126		1277
1127	$cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")	1278	$cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1128	->($self, $cb, @_);	1279	->($self, $cb, @_);
1129	}	1280	}
1130
1131		1281
1132	unshift @{ $self->{_queue} }, $cb;	1282	unshift @{ $self->{_queue} }, $cb;
1133	$self->_drain_rbuf;	1283	$self->_drain_rbuf;
1134	}	1284	}
1135		1285
…		…
1387	=cut	1537	=cut
1388		1538
1389	register_read_type json => sub {	1539	register_read_type json => sub {
1390	my ($self, $cb) = @_;	1540	my ($self, $cb) = @_;
1391		1541
1392	my $json = $self->{json} \|\|=	1542	my $json = $self->{json} \|\|= json_coder;
1393	eval { require JSON::XS; JSON::XS->new->utf8 }
1394	\|\| do { require JSON; JSON->new->utf8 };
1395		1543
1396	my $data;	1544	my $data;
1397	my $rbuf = \$self->{rbuf};	1545	my $rbuf = \$self->{rbuf};
1398		1546
1399	sub {	1547	sub {
…		…
1519	my ($self) = @_;	1667	my ($self) = @_;
1520		1668
1521	unless ($self->{_rw} \|\| $self->{_eof}) {	1669	unless ($self->{_rw} \|\| $self->{_eof}) {
1522	Scalar::Util::weaken $self;	1670	Scalar::Util::weaken $self;
1523		1671
1524	$self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {	1672	$self->{_rw} = AE::io $self->{fh}, 0, sub {
1525	my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});	1673	my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1526	my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} \|\| 8192, length $$rbuf;	1674	my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} \|\| 8192, length $$rbuf;
1527		1675
1528	if ($len > 0) {	1676	if ($len > 0) {
1529	$self->{_activity} = AnyEvent->now;	1677	$self->{_activity} = $self->{_ractivity} = AE::now;
1530		1678
1531	if ($self->{tls}) {	1679	if ($self->{tls}) {
1532	Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);	1680	Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1533		1681
1534	&_dotls ($self);	1682	&_dotls ($self);
…		…
1542	$self->_drain_rbuf;	1690	$self->_drain_rbuf;
1543		1691
1544	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {	1692	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1545	return $self->_error ($!, 1);	1693	return $self->_error ($!, 1);
1546	}	1694	}
1547	});	1695	};
1548	}	1696	}
1549	}	1697	}
1550		1698
1551	our $ERROR_SYSCALL;	1699	our $ERROR_SYSCALL;
1552	our $ERROR_WANT_READ;	1700	our $ERROR_WANT_READ;
…		…
1649	The TLS connection object will end up in C<< $handle->{tls} >>, the TLS	1797	The TLS connection object will end up in C<< $handle->{tls} >>, the TLS
1650	context in C<< $handle->{tls_ctx} >> after this call and can be used or	1798	context in C<< $handle->{tls_ctx} >> after this call and can be used or
1651	changed to your liking. Note that the handshake might have already started	1799	changed to your liking. Note that the handshake might have already started
1652	when this function returns.	1800	when this function returns.
1653		1801
1654	If it an error to start a TLS handshake more than once per	1802	Due to bugs in OpenSSL, it might or might not be possible to do multiple
1655	AnyEvent::Handle object (this is due to bugs in OpenSSL).	1803	handshakes on the same stream. Best do not attempt to use the stream after
		1804	stopping TLS.
1656		1805
1657	=cut	1806	=cut
1658		1807
1659	our %TLS_CACHE; #TODO not yet documented, should we?	1808	our %TLS_CACHE; #TODO not yet documented, should we?
1660		1809
1661	sub starttls {	1810	sub starttls {
1662	my ($self, $ssl, $ctx) = @_;	1811	my ($self, $tls, $ctx) = @_;
		1812
		1813	Carp::croak "It is an error to call starttls on an AnyEvent::Handle object while TLS is already active, caught"
		1814	if $self->{tls};
		1815
		1816	$self->{tls} = $tls;
		1817	$self->{tls_ctx} = $ctx if @_ > 2;
		1818
		1819	return unless $self->{fh};
1663		1820
1664	require Net::SSLeay;	1821	require Net::SSLeay;
1665
1666	Carp::croak "it is an error to call starttls more than once on an AnyEvent::Handle object"
1667	if $self->{tls};
1668		1822
1669	$ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();	1823	$ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1670	$ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();	1824	$ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1671		1825
		1826	$tls = delete $self->{tls};
1672	$ctx \|\|= $self->{tls_ctx};	1827	$ctx = $self->{tls_ctx};
1673		1828
1674	local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session	1829	local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1675		1830
1676	if ("HASH" eq ref $ctx) {	1831	if ("HASH" eq ref $ctx) {
1677	require AnyEvent::TLS;	1832	require AnyEvent::TLS;
…		…
1683	$ctx = new AnyEvent::TLS %$ctx;	1838	$ctx = new AnyEvent::TLS %$ctx;
1684	}	1839	}
1685	}	1840	}
1686		1841
1687	$self->{tls_ctx} = $ctx \|\| TLS_CTX ();	1842	$self->{tls_ctx} = $ctx \|\| TLS_CTX ();
1688	$self->{tls} = $ssl = $self->{tls_ctx}->_get_session ($ssl, $self, $self->{peername});	1843	$self->{tls} = $tls = $self->{tls_ctx}->_get_session ($tls, $self, $self->{peername});
1689		1844
1690	# basically, this is deep magic (because SSL_read should have the same issues)	1845	# basically, this is deep magic (because SSL_read should have the same issues)
1691	# but the openssl maintainers basically said: "trust us, it just works".	1846	# but the openssl maintainers basically said: "trust us, it just works".
1692	# (unfortunately, we have to hardcode constants because the abysmally misdesigned	1847	# (unfortunately, we have to hardcode constants because the abysmally misdesigned
1693	# and mismaintained ssleay-module doesn't even offer them).	1848	# and mismaintained ssleay-module doesn't even offer them).
…		…
1700	# and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to	1855	# and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1701	# have identity issues in that area.	1856	# have identity issues in that area.
1702	# Net::SSLeay::CTX_set_mode ($ssl,	1857	# Net::SSLeay::CTX_set_mode ($ssl,
1703	# (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } \|\| 1)	1858	# (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } \|\| 1)
1704	# \| (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } \|\| 2));	1859	# \| (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } \|\| 2));
1705	Net::SSLeay::CTX_set_mode ($ssl, 1\|2);	1860	Net::SSLeay::CTX_set_mode ($tls, 1\|2);
1706		1861
1707	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());	1862	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1708	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());	1863	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1709		1864
		1865	Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
		1866
1710	Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});	1867	Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1711		1868
1712	$self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }	1869	$self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1713	if $self->{on_starttls};	1870	if $self->{on_starttls};
1714		1871
1715	&_dotls; # need to trigger the initial handshake	1872	&_dotls; # need to trigger the initial handshake
…		…
1718		1875
1719	=item $handle->stoptls	1876	=item $handle->stoptls
1720		1877
1721	Shuts down the SSL connection - this makes a proper EOF handshake by	1878	Shuts down the SSL connection - this makes a proper EOF handshake by
1722	sending a close notify to the other side, but since OpenSSL doesn't	1879	sending a close notify to the other side, but since OpenSSL doesn't
1723	support non-blocking shut downs, it is not possible to re-use the stream	1880	support non-blocking shut downs, it is not guarenteed that you can re-use
1724	afterwards.	1881	the stream afterwards.
1725		1882
1726	=cut	1883	=cut
1727		1884
1728	sub stoptls {	1885	sub stoptls {
1729	my ($self) = @_;	1886	my ($self) = @_;
…		…
1742	sub _freetls {	1899	sub _freetls {
1743	my ($self) = @_;	1900	my ($self) = @_;
1744		1901
1745	return unless $self->{tls};	1902	return unless $self->{tls};
1746		1903
1747	$self->{tls_ctx}->_put_session (delete $self->{tls});	1904	$self->{tls_ctx}->_put_session (delete $self->{tls})
		1905	if $self->{tls} > 0;
1748		1906
1749	delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};	1907	delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1750	}	1908	}
1751		1909
1752	sub DESTROY {	1910	sub DESTROY {
…		…
1760	my $fh = delete $self->{fh};	1918	my $fh = delete $self->{fh};
1761	my $wbuf = delete $self->{wbuf};	1919	my $wbuf = delete $self->{wbuf};
1762		1920
1763	my @linger;	1921	my @linger;
1764		1922
1765	push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {	1923	push @linger, AE::io $fh, 1, sub {
1766	my $len = syswrite $fh, $wbuf, length $wbuf;	1924	my $len = syswrite $fh, $wbuf, length $wbuf;
1767		1925
1768	if ($len > 0) {	1926	if ($len > 0) {
1769	substr $wbuf, 0, $len, "";	1927	substr $wbuf, 0, $len, "";
1770	} else {	1928	} else {
1771	@linger = (); # end	1929	@linger = (); # end
1772	}	1930	}
1773	});	1931	};
1774	push @linger, AnyEvent->timer (after => $linger, cb => sub {	1932	push @linger, AE::timer $linger, 0, sub {
1775	@linger = ();	1933	@linger = ();
1776	});	1934	};
1777	}	1935	}
1778	}	1936	}
1779		1937
1780	=item $handle->destroy	1938	=item $handle->destroy
1781		1939
1782	Shuts down the handle object as much as possible - this call ensures that	1940	Shuts down the handle object as much as possible - this call ensures that
1783	no further callbacks will be invoked and as many resources as possible	1941	no further callbacks will be invoked and as many resources as possible
1784	will be freed. You must not call any methods on the object afterwards.	1942	will be freed. Any method you will call on the handle object after
		1943	destroying it in this way will be silently ignored (and it will return the
		1944	empty list).
1785		1945
1786	Normally, you can just "forget" any references to an AnyEvent::Handle	1946	Normally, you can just "forget" any references to an AnyEvent::Handle
1787	object and it will simply shut down. This works in fatal error and EOF	1947	object and it will simply shut down. This works in fatal error and EOF
1788	callbacks, as well as code outside. It does I<NOT> work in a read or write	1948	callbacks, as well as code outside. It does I<NOT> work in a read or write
1789	callback, so when you want to destroy the AnyEvent::Handle object from	1949	callback, so when you want to destroy the AnyEvent::Handle object from
…		…
1803	sub destroy {	1963	sub destroy {
1804	my ($self) = @_;	1964	my ($self) = @_;
1805		1965
1806	$self->DESTROY;	1966	$self->DESTROY;
1807	%$self = ();	1967	%$self = ();
		1968	bless $self, "AnyEvent::Handle::destroyed";
		1969	}
		1970
		1971	sub AnyEvent::Handle::destroyed::AUTOLOAD {
		1972	#nop
1808	}	1973	}
1809		1974
1810	=item AnyEvent::Handle::TLS_CTX	1975	=item AnyEvent::Handle::TLS_CTX
1811		1976
1812	This function creates and returns the AnyEvent::TLS object used by default	1977	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.159 by root, Fri Jul 24 12:35:58 2009 UTC vs. Revision 1.184 by root, Thu Sep 3 13:14:38 2009 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.159 by root, Fri Jul 24 12:35:58 2009 UTC vs.
Revision 1.184 by root, Thu Sep 3 13:14:38 2009 UTC