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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.58 by root, Wed Jun 4 22:51:15 2008 UTC vs.
Revision 1.67 by root, Fri Jun 6 15:33:10 2008 UTC

…		…
14		14
15	AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent	15	AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16		16
17	=cut	17	=cut
18		18
19	our $VERSION = 4.13;	19	our $VERSION = 4.15;
20		20
21	=head1 SYNOPSIS	21	=head1 SYNOPSIS
22		22
23	use AnyEvent;	23	use AnyEvent;
24	use AnyEvent::Handle;	24	use AnyEvent::Handle;
…		…
105	C<croak>.	105	C<croak>.
106		106
107	=item on_read => $cb->($handle)	107	=item on_read => $cb->($handle)
108		108
109	This sets the default read callback, which is called when data arrives	109	This sets the default read callback, which is called when data arrives
110	and no read request is in the queue.	110	and no read request is in the queue (unlike read queue callbacks, this
		111	callback will only be called when at least one octet of data is in the
		112	read buffer).
111		113
112	To access (and remove data from) the read buffer, use the C<< ->rbuf >>	114	To access (and remove data from) the read buffer, use the C<< ->rbuf >>
113	method or access the C<$handle->{rbuf}> member directly.	115	method or access the C<$handle->{rbuf}> member directly.
114		116
115	When an EOF condition is detected then AnyEvent::Handle will first try to	117	When an EOF condition is detected then AnyEvent::Handle will first try to
…		…
165		167
166	Sets the amount of bytes (default: C<0>) that make up an "empty" write	168	Sets the amount of bytes (default: C<0>) that make up an "empty" write
167	buffer: If the write reaches this size or gets even samller it is	169	buffer: If the write reaches this size or gets even samller it is
168	considered empty.	170	considered empty.
169		171
		172	=item linger => <seconds>
		173
		174	If non-zero (default: C<3600>), then the destructor of the
		175	AnyEvent::Handle object will check wether there is still outstanding write
		176	data and will install a watcher that will write out this data. No errors
		177	will be reported (this mostly matches how the operating system treats
		178	outstanding data at socket close time).
		179
		180	This will not work for partial TLS data that could not yet been
		181	encoded. This data will be lost.
		182
170	=item tls => "accept" \| "connect" \| Net::SSLeay::SSL object	183	=item tls => "accept" \| "connect" \| Net::SSLeay::SSL object
171		184
172	When this parameter is given, it enables TLS (SSL) mode, that means it	185	When this parameter is given, it enables TLS (SSL) mode, that means it
173	will start making tls handshake and will transparently encrypt/decrypt	186	will start making tls handshake and will transparently encrypt/decrypt
174	data.	187	data.
…		…
228		241
229	$self->{_activity} = AnyEvent->now;	242	$self->{_activity} = AnyEvent->now;
230	$self->_timeout;	243	$self->_timeout;
231		244
232	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};	245	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
233	$self->on_read (delete $self->{on_read} ) if $self->{on_read};	246
		247	$self->start_read
		248	if $self->{on_read};
234		249
235	$self	250	$self
236	}	251	}
237		252
238	sub _shutdown {	253	sub _shutdown {
…		…
476	my ($self, $string) = @_;	491	my ($self, $string) = @_;
477		492
478	sprintf "%d:%s,", (length $string), $string	493	sprintf "%d:%s,", (length $string), $string
479	};	494	};
480		495
		496	=item packstring => $format, $data
		497
		498	An octet string prefixed with an encoded length. The encoding C<$format>
		499	uses the same format as a Perl C<pack> format, but must specify a single
		500	integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
		501	optional C<!>, C<< < >> or C<< > >> modifier).
		502
		503	=cut
		504
		505	register_write_type packstring => sub {
		506	my ($self, $format, $string) = @_;
		507
		508	pack "$format/a*", $string
		509	};
		510
481	=item json => $array_or_hashref	511	=item json => $array_or_hashref
482		512
483	Encodes the given hash or array reference into a JSON object. Unless you	513	Encodes the given hash or array reference into a JSON object. Unless you
484	provide your own JSON object, this means it will be encoded to JSON text	514	provide your own JSON object, this means it will be encoded to JSON text
485	in UTF-8.	515	in UTF-8.
…		…
517		547
518	$self->{json} ? $self->{json}->encode ($ref)	548	$self->{json} ? $self->{json}->encode ($ref)
519	: JSON::encode_json ($ref)	549	: JSON::encode_json ($ref)
520	};	550	};
521		551
		552	=item storable => $reference
		553
		554	Freezes the given reference using L<Storable> and writes it to the
		555	handle. Uses the C<nfreeze> format.
		556
		557	=cut
		558
		559	register_write_type storable => sub {
		560	my ($self, $ref) = @_;
		561
		562	require Storable;
		563
		564	pack "w/a*", Storable::nfreeze ($ref)
		565	};
		566
522	=back	567	=back
523		568
524	=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)	569	=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525		570
526	This function (not method) lets you add your own types to C<push_write>.	571	This function (not method) lets you add your own types to C<push_write>.
…		…
553	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want	598	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
554	or not.	599	or not.
555		600
556	In the more complex case, you want to queue multiple callbacks. In this	601	In the more complex case, you want to queue multiple callbacks. In this
557	case, AnyEvent::Handle will call the first queued callback each time new	602	case, AnyEvent::Handle will call the first queued callback each time new
558	data arrives and removes it when it has done its job (see C<push_read>,	603	data arrives (also the first time it is queued) and removes it when it has
559	below).	604	done its job (see C<push_read>, below).
560		605
561	This way you can, for example, push three line-reads, followed by reading	606	This way you can, for example, push three line-reads, followed by reading
562	a chunk of data, and AnyEvent::Handle will execute them in order.	607	a chunk of data, and AnyEvent::Handle will execute them in order.
563		608
564	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by	609	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
…		…
616	=cut	661	=cut
617		662
618	sub _drain_rbuf {	663	sub _drain_rbuf {
619	my ($self) = @_;	664	my ($self) = @_;
620		665
		666	local $self->{_in_drain} = 1;
		667
621	if (	668	if (
622	defined $self->{rbuf_max}	669	defined $self->{rbuf_max}
623	&& $self->{rbuf_max} < length $self->{rbuf}	670	&& $self->{rbuf_max} < length $self->{rbuf}
624	) {	671	) {
625	return $self->_error (&Errno::ENOSPC, 1);	672	return $self->_error (&Errno::ENOSPC, 1);
626	}	673	}
627		674
628	return if $self->{in_drain};	675	while () {
629	local $self->{in_drain} = 1;
630
631	while (my $len = length $self->{rbuf}) {
632	no strict 'refs';	676	no strict 'refs';
		677
		678	my $len = length $self->{rbuf};
		679
633	if (my $cb = shift @{ $self->{_queue} }) {	680	if (my $cb = shift @{ $self->{_queue} }) {
634	unless ($cb->($self)) {	681	unless ($cb->($self)) {
635	if ($self->{_eof}) {	682	if ($self->{_eof}) {
636	# no progress can be made (not enough data and no data forthcoming)	683	# no progress can be made (not enough data and no data forthcoming)
637	return $self->_error (&Errno::EPIPE, 1);	684	$self->_error (&Errno::EPIPE, 1), last;
638	}	685	}
639		686
640	unshift @{ $self->{_queue} }, $cb;	687	unshift @{ $self->{_queue} }, $cb;
641	last;	688	last;
642	}	689	}
643	} elsif ($self->{on_read}) {	690	} elsif ($self->{on_read}) {
		691	last unless $len;
		692
644	$self->{on_read}($self);	693	$self->{on_read}($self);
645		694
646	if (	695	if (
647	$len == length $self->{rbuf} # if no data has been consumed	696	$len == length $self->{rbuf} # if no data has been consumed
648	&& !@{ $self->{_queue} } # and the queue is still empty	697	&& !@{ $self->{_queue} } # and the queue is still empty
649	&& $self->{on_read} # but we still have on_read	698	&& $self->{on_read} # but we still have on_read
650	) {	699	) {
651	# no further data will arrive	700	# no further data will arrive
652	# so no progress can be made	701	# so no progress can be made
653	return $self->_error (&Errno::EPIPE, 1)	702	$self->_error (&Errno::EPIPE, 1), last
654	if $self->{_eof};	703	if $self->{_eof};
655		704
656	last; # more data might arrive	705	last; # more data might arrive
657	}	706	}
658	} else {	707	} else {
…		…
682		731
683	sub on_read {	732	sub on_read {
684	my ($self, $cb) = @_;	733	my ($self, $cb) = @_;
685		734
686	$self->{on_read} = $cb;	735	$self->{on_read} = $cb;
687	$self->_drain_rbuf if $cb;	736	$self->_drain_rbuf if $cb && !$self->{_in_drain};
688	}	737	}
689		738
690	=item $handle->rbuf	739	=item $handle->rbuf
691		740
692	Returns the read buffer (as a modifiable lvalue).	741	Returns the read buffer (as a modifiable lvalue).
…		…
741	$cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")	790	$cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
742	->($self, $cb, @_);	791	->($self, $cb, @_);
743	}	792	}
744		793
745	push @{ $self->{_queue} }, $cb;	794	push @{ $self->{_queue} }, $cb;
746	$self->_drain_rbuf;	795	$self->_drain_rbuf unless $self->{_in_drain};
747	}	796	}
748		797
749	sub unshift_read {	798	sub unshift_read {
750	my $self = shift;	799	my $self = shift;
751	my $cb = pop;	800	my $cb = pop;
…		…
757	->($self, $cb, @_);	806	->($self, $cb, @_);
758	}	807	}
759		808
760		809
761	unshift @{ $self->{_queue} }, $cb;	810	unshift @{ $self->{_queue} }, $cb;
762	$self->_drain_rbuf;	811	$self->_drain_rbuf unless $self->{_in_drain};
763	}	812	}
764		813
765	=item $handle->push_read (type => @args, $cb)	814	=item $handle->push_read (type => @args, $cb)
766		815
767	=item $handle->unshift_read (type => @args, $cb)	816	=item $handle->unshift_read (type => @args, $cb)
…		…
852		901
853	sub unshift_read_line {	902	sub unshift_read_line {
854	my $self = shift;	903	my $self = shift;
855	$self->unshift_read (line => @_);	904	$self->unshift_read (line => @_);
856	}	905	}
857
858	=item netstring => $cb->($handle, $string)
859
860	A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
861
862	Throws an error with C<$!> set to EBADMSG on format violations.
863
864	=cut
865
866	register_read_type netstring => sub {
867	my ($self, $cb) = @_;
868
869	sub {
870	unless ($_[0]{rbuf} =~ s/^(0\|[1-9][0-9]*)://) {
871	if ($_[0]{rbuf} =~ /[^0-9]/) {
872	$self->_error (&Errno::EBADMSG);
873	}
874	return;
875	}
876
877	my $len = $1;
878
879	$self->unshift_read (chunk => $len, sub {
880	my $string = $_[1];
881	$_[0]->unshift_read (chunk => 1, sub {
882	if ($_[1] eq ",") {
883	$cb->($_[0], $string);
884	} else {
885	$self->_error (&Errno::EBADMSG);
886	}
887	});
888	});
889
890	1
891	}
892	};
893		906
894	=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)	907	=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
895		908
896	Makes a regex match against the regex object C<$accept> and returns	909	Makes a regex match against the regex object C<$accept> and returns
897	everything up to and including the match.	910	everything up to and including the match.
…		…
959		972
960	()	973	()
961	}	974	}
962	};	975	};
963		976
		977	=item netstring => $cb->($handle, $string)
		978
		979	A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
		980
		981	Throws an error with C<$!> set to EBADMSG on format violations.
		982
		983	=cut
		984
		985	register_read_type netstring => sub {
		986	my ($self, $cb) = @_;
		987
		988	sub {
		989	unless ($_[0]{rbuf} =~ s/^(0\|[1-9][0-9]*)://) {
		990	if ($_[0]{rbuf} =~ /[^0-9]/) {
		991	$self->_error (&Errno::EBADMSG);
		992	}
		993	return;
		994	}
		995
		996	my $len = $1;
		997
		998	$self->unshift_read (chunk => $len, sub {
		999	my $string = $_[1];
		1000	$_[0]->unshift_read (chunk => 1, sub {
		1001	if ($_[1] eq ",") {
		1002	$cb->($_[0], $string);
		1003	} else {
		1004	$self->_error (&Errno::EBADMSG);
		1005	}
		1006	});
		1007	});
		1008
		1009	1
		1010	}
		1011	};
		1012
		1013	=item packstring => $format, $cb->($handle, $string)
		1014
		1015	An octet string prefixed with an encoded length. The encoding C<$format>
		1016	uses the same format as a Perl C<pack> format, but must specify a single
		1017	integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
		1018	optional C<!>, C<< < >> or C<< > >> modifier).
		1019
		1020	DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
		1021
		1022	Example: read a block of data prefixed by its length in BER-encoded
		1023	format (very efficient).
		1024
		1025	$handle->push_read (packstring => "w", sub {
		1026	my ($handle, $data) = @_;
		1027	});
		1028
		1029	=cut
		1030
		1031	register_read_type packstring => sub {
		1032	my ($self, $cb, $format) = @_;
		1033
		1034	sub {
		1035	# when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
		1036	defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
		1037	or return;
		1038
		1039	# remove prefix
		1040	substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
		1041
		1042	# read rest
		1043	$_[0]->unshift_read (chunk => $len, $cb);
		1044
		1045	1
		1046	}
		1047	};
		1048
964	=item json => $cb->($handle, $hash_or_arrayref)	1049	=item json => $cb->($handle, $hash_or_arrayref)
965		1050
966	Reads a JSON object or array, decodes it and passes it to the callback.	1051	Reads a JSON object or array, decodes it and passes it to the callback.
967		1052
968	If a C<json> object was passed to the constructor, then that will be used	1053	If a C<json> object was passed to the constructor, then that will be used
…		…
978	the C<json> write type description, above, for an actual example.	1063	the C<json> write type description, above, for an actual example.
979		1064
980	=cut	1065	=cut
981		1066
982	register_read_type json => sub {	1067	register_read_type json => sub {
983	my ($self, $cb, $accept, $reject, $skip) = @_;	1068	my ($self, $cb) = @_;
984		1069
985	require JSON;	1070	require JSON;
986		1071
987	my $data;	1072	my $data;
988	my $rbuf = \$self->{rbuf};	1073	my $rbuf = \$self->{rbuf};
…		…
1003	()	1088	()
1004	}	1089	}
1005	}	1090	}
1006	};	1091	};
1007		1092
		1093	=item storable => $cb->($handle, $ref)
		1094
		1095	Deserialises a L<Storable> frozen representation as written by the
		1096	C<storable> write type (BER-encoded length prefix followed by nfreeze'd
		1097	data).
		1098
		1099	Raises C<EBADMSG> error if the data could not be decoded.
		1100
		1101	=cut
		1102
		1103	register_read_type storable => sub {
		1104	my ($self, $cb) = @_;
		1105
		1106	require Storable;
		1107
		1108	sub {
		1109	# when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
		1110	defined (my $len = eval { unpack "w", $_[0]->{rbuf} })
		1111	or return;
		1112
		1113	# remove prefix
		1114	substr $_[0]->{rbuf}, 0, (length pack "w", $len), "";
		1115
		1116	# read rest
		1117	$_[0]->unshift_read (chunk => $len, sub {
		1118	if (my $ref = eval { Storable::thaw ($_[1]) }) {
		1119	$cb->($_[0], $ref);
		1120	} else {
		1121	$self->_error (&Errno::EBADMSG);
		1122	}
		1123	});
		1124	}
		1125	};
		1126
1008	=back	1127	=back
1009		1128
1010	=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)	1129	=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1011		1130
1012	This function (not method) lets you add your own types to C<push_read>.	1131	This function (not method) lets you add your own types to C<push_read>.
…		…
1062	if ($len > 0) {	1181	if ($len > 0) {
1063	$self->{_activity} = AnyEvent->now;	1182	$self->{_activity} = AnyEvent->now;
1064		1183
1065	$self->{filter_r}	1184	$self->{filter_r}
1066	? $self->{filter_r}($self, $rbuf)	1185	? $self->{filter_r}($self, $rbuf)
1067	: $self->_drain_rbuf;	1186	: $self->{_in_drain} \|\| $self->_drain_rbuf;
1068		1187
1069	} elsif (defined $len) {	1188	} elsif (defined $len) {
1070	delete $self->{_rw};	1189	delete $self->{_rw};
1071	$self->{_eof} = 1;	1190	$self->{_eof} = 1;
1072	$self->_drain_rbuf;	1191	$self->_drain_rbuf unless $self->{_in_drain};
1073		1192
1074	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {	1193	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1075	return $self->_error ($!, 1);	1194	return $self->_error ($!, 1);
1076	}	1195	}
1077	});	1196	});
…		…
1095	}	1214	}
1096		1215
1097	while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {	1216	while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1098	if (length $buf) {	1217	if (length $buf) {
1099	$self->{rbuf} .= $buf;	1218	$self->{rbuf} .= $buf;
1100	$self->_drain_rbuf;	1219	$self->_drain_rbuf unless $self->{_in_drain};
1101	} else {	1220	} else {
1102	# let's treat SSL-eof as we treat normal EOF	1221	# let's treat SSL-eof as we treat normal EOF
1103	$self->{_eof} = 1;	1222	$self->{_eof} = 1;
1104	$self->_shutdown;	1223	$self->_shutdown;
1105	return;	1224	return;
…		…
1197		1316
1198	sub DESTROY {	1317	sub DESTROY {
1199	my $self = shift;	1318	my $self = shift;
1200		1319
1201	$self->stoptls;	1320	$self->stoptls;
		1321
		1322	my $linger = exists $self->{linger} ? $self->{linger} : 3600;
		1323
		1324	if ($linger && length $self->{wbuf}) {
		1325	my $fh = delete $self->{fh};
		1326	my $wbuf = delete $self->{wbuf};
		1327
		1328	my @linger;
		1329
		1330	push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
		1331	my $len = syswrite $fh, $wbuf, length $wbuf;
		1332
		1333	if ($len > 0) {
		1334	substr $wbuf, 0, $len, "";
		1335	} else {
		1336	@linger = (); # end
		1337	}
		1338	});
		1339	push @linger, AnyEvent->timer (after => $linger, cb => sub {
		1340	@linger = ();
		1341	});
		1342	}
1202	}	1343	}
1203		1344
1204	=item AnyEvent::Handle::TLS_CTX	1345	=item AnyEvent::Handle::TLS_CTX
1205		1346
1206	This function creates and returns the Net::SSLeay::CTX object used by	1347	This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents): Revision 1.58 by root, Wed Jun 4 22:51:15 2008 UTC vs. Revision 1.67 by root, Fri Jun 6 15:33:10 2008 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.58 by root, Wed Jun 4 22:51:15 2008 UTC vs.
Revision 1.67 by root, Fri Jun 6 15:33:10 2008 UTC