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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.5 by elmex, Mon Apr 28 08:01:05 2008 UTC vs.
Revision 1.38 by root, Mon May 26 21:28:33 2008 UTC

1	package AnyEvent::Handle;	1	package AnyEvent::Handle;
2		2
3	use warnings;	3	no warnings;
4	use strict;	4	use strict;
5		5
6	use AnyEvent;	6	use AnyEvent ();
7	use IO::Handle;	7	use AnyEvent::Util qw(WSAWOULDBLOCK);
		8	use Scalar::Util ();
		9	use Carp ();
		10	use Fcntl ();
8	use Errno qw/EAGAIN EINTR/;	11	use Errno qw/EAGAIN EINTR/;
9		12
10	=head1 NAME	13	=head1 NAME
11		14
12	AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent	15	AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
13		16
14	=head1 VERSION
15
16	Version 0.01
17
18	=cut	17	=cut
19		18
20	our $VERSION = '0.01';	19	our $VERSION = '0.04';
21		20
22	=head1 SYNOPSIS	21	=head1 SYNOPSIS
23		22
24	use AnyEvent;	23	use AnyEvent;
25	use AnyEvent::Handle;	24	use AnyEvent::Handle;
26		25
27	my $cv = AnyEvent->condvar;	26	my $cv = AnyEvent->condvar;
28		27
29	my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);	28	my $handle =
30
31	$ae_fh->on_eof (sub { $cv->broadcast });
32
33	$ae_fh->readlines (sub {
34	my ($ae_fh, @lines) = @_;
35	for (@lines) {
36	chomp;
37	print "Line: $_";
38	}
39	});
40
41	# or use the constructor to pass the callback:
42
43	my $ae_fh2 =
44	AnyEvent::Handle->new (	29	AnyEvent::Handle->new (
45	fh => \*STDIN,	30	fh => \*STDIN,
46	on_eof => sub {	31	on_eof => sub {
47	$cv->broadcast;	32	$cv->broadcast;
48	},	33	},
49	on_readline => sub {	34	);
		35
		36	# send some request line
		37	$handle->push_write ("getinfo\015\012");
		38
		39	# read the response line
		40	$handle->push_read (line => sub {
50	my ($ae_fh, @lines) = @_;	41	my ($handle, $line) = @_;
51	for (@lines) {	42	warn "read line <$line>\n";
52	chomp;	43	$cv->send;
53	print "Line: $_";	44	});
		45
		46	$cv->recv;
		47
		48	=head1 DESCRIPTION
		49
		50	This module is a helper module to make it easier to do event-based I/O on
		51	filehandles. For utility functions for doing non-blocking connects and accepts
		52	on sockets see L<AnyEvent::Util>.
		53
		54	In the following, when the documentation refers to of "bytes" then this
		55	means characters. As sysread and syswrite are used for all I/O, their
		56	treatment of characters applies to this module as well.
		57
		58	All callbacks will be invoked with the handle object as their first
		59	argument.
		60
		61	=head1 METHODS
		62
		63	=over 4
		64
		65	=item B<new (%args)>
		66
		67	The constructor supports these arguments (all as key => value pairs).
		68
		69	=over 4
		70
		71	=item fh => $filehandle [MANDATORY]
		72
		73	The filehandle this L<AnyEvent::Handle> object will operate on.
		74
		75	NOTE: The filehandle will be set to non-blocking (using
		76	AnyEvent::Util::fh_nonblocking).
		77
		78	=item on_eof => $cb->($self)
		79
		80	Set the callback to be called on EOF.
		81
		82	While not mandatory, it is highly recommended to set an eof callback,
		83	otherwise you might end up with a closed socket while you are still
		84	waiting for data.
		85
		86	=item on_error => $cb->($self)
		87
		88	This is the fatal error callback, that is called when, well, a fatal error
		89	occurs, such as not being able to resolve the hostname, failure to connect
		90	or a read error.
		91
		92	The object will not be in a usable state when this callback has been
		93	called.
		94
		95	On callback entrance, the value of C<$!> contains the operating system
		96	error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>).
		97
		98	The callback should throw an exception. If it returns, then
		99	AnyEvent::Handle will C<croak> for you.
		100
		101	While not mandatory, it is I<highly> recommended to set this callback, as
		102	you will not be notified of errors otherwise. The default simply calls
		103	die.
		104
		105	=item on_read => $cb->($self)
		106
		107	This sets the default read callback, which is called when data arrives
		108	and no read request is in the queue.
		109
		110	To access (and remove data from) the read buffer, use the C<< ->rbuf >>
		111	method or access the C<$self->{rbuf}> member directly.
		112
		113	When an EOF condition is detected then AnyEvent::Handle will first try to
		114	feed all the remaining data to the queued callbacks and C<on_read> before
		115	calling the C<on_eof> callback. If no progress can be made, then a fatal
		116	error will be raised (with C<$!> set to C<EPIPE>).
		117
		118	=item on_drain => $cb->()
		119
		120	This sets the callback that is called when the write buffer becomes empty
		121	(or when the callback is set and the buffer is empty already).
		122
		123	To append to the write buffer, use the C<< ->push_write >> method.
		124
		125	=item rbuf_max => <bytes>
		126
		127	If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
		128	when the read buffer ever (strictly) exceeds this size. This is useful to
		129	avoid denial-of-service attacks.
		130
		131	For example, a server accepting connections from untrusted sources should
		132	be configured to accept only so-and-so much data that it cannot act on
		133	(for example, when expecting a line, an attacker could send an unlimited
		134	amount of data without a callback ever being called as long as the line
		135	isn't finished).
		136
		137	=item read_size => <bytes>
		138
		139	The default read block size (the amount of bytes this module will try to read
		140	on each [loop iteration). Default: C<4096>.
		141
		142	=item low_water_mark => <bytes>
		143
		144	Sets the amount of bytes (default: C<0>) that make up an "empty" write
		145	buffer: If the write reaches this size or gets even samller it is
		146	considered empty.
		147
		148	=item tls => "accept" \| "connect" \| Net::SSLeay::SSL object
		149
		150	When this parameter is given, it enables TLS (SSL) mode, that means it
		151	will start making tls handshake and will transparently encrypt/decrypt
		152	data.
		153
		154	TLS mode requires Net::SSLeay to be installed (it will be loaded
		155	automatically when you try to create a TLS handle).
		156
		157	For the TLS server side, use C<accept>, and for the TLS client side of a
		158	connection, use C<connect> mode.
		159
		160	You can also provide your own TLS connection object, but you have
		161	to make sure that you call either C<Net::SSLeay::set_connect_state>
		162	or C<Net::SSLeay::set_accept_state> on it before you pass it to
		163	AnyEvent::Handle.
		164
		165	See the C<starttls> method if you need to start TLs negotiation later.
		166
		167	=item tls_ctx => $ssl_ctx
		168
		169	Use the given Net::SSLeay::CTX object to create the new TLS connection
		170	(unless a connection object was specified directly). If this parameter is
		171	missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
		172
		173	=item filter_r => $cb
		174
		175	=item filter_w => $cb
		176
		177	These exist, but are undocumented at this time.
		178
		179	=back
		180
		181	=cut
		182
		183	sub new {
		184	my $class = shift;
		185
		186	my $self = bless { @_ }, $class;
		187
		188	$self->{fh} or Carp::croak "mandatory argument fh is missing";
		189
		190	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
		191
		192	if ($self->{tls}) {
		193	require Net::SSLeay;
		194	$self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
		195	}
		196
		197	$self->on_eof (delete $self->{on_eof} ) if $self->{on_eof};
		198	$self->on_error (delete $self->{on_error}) if $self->{on_error};
		199	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
		200	$self->on_read (delete $self->{on_read} ) if $self->{on_read};
		201
		202	$self->start_read;
		203
		204	$self
		205	}
		206
		207	sub _shutdown {
		208	my ($self) = @_;
		209
		210	delete $self->{_rw};
		211	delete $self->{_ww};
		212	delete $self->{fh};
		213	}
		214
		215	sub error {
		216	my ($self) = @_;
		217
		218	{
		219	local $!;
		220	$self->_shutdown;
		221	}
		222
		223	$self->{on_error}($self)
		224	if $self->{on_error};
		225
		226	Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
		227	}
		228
		229	=item $fh = $handle->fh
		230
		231	This method returns the file handle of the L<AnyEvent::Handle> object.
		232
		233	=cut
		234
		235	sub fh { $_[0]{fh} }
		236
		237	=item $handle->on_error ($cb)
		238
		239	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
		240
		241	=cut
		242
		243	sub on_error {
		244	$_[0]{on_error} = $_[1];
		245	}
		246
		247	=item $handle->on_eof ($cb)
		248
		249	Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
		250
		251	=cut
		252
		253	sub on_eof {
		254	$_[0]{on_eof} = $_[1];
		255	}
		256
		257	#############################################################################
		258
		259	=back
		260
		261	=head2 WRITE QUEUE
		262
		263	AnyEvent::Handle manages two queues per handle, one for writing and one
		264	for reading.
		265
		266	The write queue is very simple: you can add data to its end, and
		267	AnyEvent::Handle will automatically try to get rid of it for you.
		268
		269	When data could be written and the write buffer is shorter then the low
		270	water mark, the C<on_drain> callback will be invoked.
		271
		272	=over 4
		273
		274	=item $handle->on_drain ($cb)
		275
		276	Sets the C<on_drain> callback or clears it (see the description of
		277	C<on_drain> in the constructor).
		278
		279	=cut
		280
		281	sub on_drain {
		282	my ($self, $cb) = @_;
		283
		284	$self->{on_drain} = $cb;
		285
		286	$cb->($self)
		287	if $cb && $self->{low_water_mark} >= length $self->{wbuf};
		288	}
		289
		290	=item $handle->push_write ($data)
		291
		292	Queues the given scalar to be written. You can push as much data as you
		293	want (only limited by the available memory), as C<AnyEvent::Handle>
		294	buffers it independently of the kernel.
		295
		296	=cut
		297
		298	sub _drain_wbuf {
		299	my ($self) = @_;
		300
		301	if (!$self->{_ww} && length $self->{wbuf}) {
		302
		303	Scalar::Util::weaken $self;
		304
		305	my $cb = sub {
		306	my $len = syswrite $self->{fh}, $self->{wbuf};
		307
		308	if ($len >= 0) {
		309	substr $self->{wbuf}, 0, $len, "";
		310
		311	$self->{on_drain}($self)
		312	if $self->{low_water_mark} >= length $self->{wbuf}
		313	&& $self->{on_drain};
		314
		315	delete $self->{_ww} unless length $self->{wbuf};
		316	} elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
		317	$self->error;
		318	}
		319	};
		320
		321	# try to write data immediately
		322	$cb->();
		323
		324	# if still data left in wbuf, we need to poll
		325	$self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
		326	if length $self->{wbuf};
		327	};
		328	}
		329
		330	our %WH;
		331
		332	sub register_write_type($$) {
		333	$WH{$_[0]} = $_[1];
		334	}
		335
		336	sub push_write {
		337	my $self = shift;
		338
		339	if (@_ > 1) {
		340	my $type = shift;
		341
		342	@_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
		343	->($self, @_);
		344	}
		345
		346	if ($self->{filter_w}) {
		347	$self->{filter_w}->($self, \$_[0]);
		348	} else {
		349	$self->{wbuf} .= $_[0];
		350	$self->_drain_wbuf;
		351	}
		352	}
		353
		354	=item $handle->push_write (type => @args)
		355
		356	=item $handle->unshift_write (type => @args)
		357
		358	Instead of formatting your data yourself, you can also let this module do
		359	the job by specifying a type and type-specific arguments.
		360
		361	Predefined types are (if you have ideas for additional types, feel free to
		362	drop by and tell us):
		363
		364	=over 4
		365
		366	=item netstring => $string
		367
		368	Formats the given value as netstring
		369	(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
		370
		371	=back
		372
		373	=cut
		374
		375	register_write_type netstring => sub {
		376	my ($self, $string) = @_;
		377
		378	sprintf "%d:%s,", (length $string), $string
		379	};
		380
		381	=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
		382
		383	This function (not method) lets you add your own types to C<push_write>.
		384	Whenever the given C<type> is used, C<push_write> will invoke the code
		385	reference with the handle object and the remaining arguments.
		386
		387	The code reference is supposed to return a single octet string that will
		388	be appended to the write buffer.
		389
		390	Note that this is a function, and all types registered this way will be
		391	global, so try to use unique names.
		392
		393	=cut
		394
		395	#############################################################################
		396
		397	=back
		398
		399	=head2 READ QUEUE
		400
		401	AnyEvent::Handle manages two queues per handle, one for writing and one
		402	for reading.
		403
		404	The read queue is more complex than the write queue. It can be used in two
		405	ways, the "simple" way, using only C<on_read> and the "complex" way, using
		406	a queue.
		407
		408	In the simple case, you just install an C<on_read> callback and whenever
		409	new data arrives, it will be called. You can then remove some data (if
		410	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
		411	or not.
		412
		413	In the more complex case, you want to queue multiple callbacks. In this
		414	case, AnyEvent::Handle will call the first queued callback each time new
		415	data arrives and removes it when it has done its job (see C<push_read>,
		416	below).
		417
		418	This way you can, for example, push three line-reads, followed by reading
		419	a chunk of data, and AnyEvent::Handle will execute them in order.
		420
		421	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
		422	the specified number of bytes which give an XML datagram.
		423
		424	# in the default state, expect some header bytes
		425	$handle->on_read (sub {
		426	# some data is here, now queue the length-header-read (4 octets)
		427	shift->unshift_read_chunk (4, sub {
		428	# header arrived, decode
		429	my $len = unpack "N", $_[1];
		430
		431	# now read the payload
		432	shift->unshift_read_chunk ($len, sub {
		433	my $xml = $_[1];
		434	# handle xml
		435	});
		436	});
		437	});
		438
		439	Example 2: Implement a client for a protocol that replies either with
		440	"OK" and another line or "ERROR" for one request, and 64 bytes for the
		441	second request. Due tot he availability of a full queue, we can just
		442	pipeline sending both requests and manipulate the queue as necessary in
		443	the callbacks:
		444
		445	# request one
		446	$handle->push_write ("request 1\015\012");
		447
		448	# we expect "ERROR" or "OK" as response, so push a line read
		449	$handle->push_read_line (sub {
		450	# if we got an "OK", we have to _prepend_ another line,
		451	# so it will be read before the second request reads its 64 bytes
		452	# which are already in the queue when this callback is called
		453	# we don't do this in case we got an error
		454	if ($_[1] eq "OK") {
		455	$_[0]->unshift_read_line (sub {
		456	my $response = $_[1];
		457	...
		458	});
		459	}
		460	});
		461
		462	# request two
		463	$handle->push_write ("request 2\015\012");
		464
		465	# simply read 64 bytes, always
		466	$handle->push_read_chunk (64, sub {
		467	my $response = $_[1];
		468	...
		469	});
		470
		471	=over 4
		472
		473	=cut
		474
		475	sub _drain_rbuf {
		476	my ($self) = @_;
		477
		478	if (
		479	defined $self->{rbuf_max}
		480	&& $self->{rbuf_max} < length $self->{rbuf}
		481	) {
		482	$! = &Errno::ENOSPC;
		483	$self->error;
		484	}
		485
		486	return if $self->{in_drain};
		487	local $self->{in_drain} = 1;
		488
		489	while (my $len = length $self->{rbuf}) {
		490	no strict 'refs';
		491	if (my $cb = shift @{ $self->{_queue} }) {
		492	unless ($cb->($self)) {
		493	if ($self->{_eof}) {
		494	# no progress can be made (not enough data and no data forthcoming)
		495	$! = &Errno::EPIPE;
		496	$self->error;
54	}	497	}
		498
		499	unshift @{ $self->{_queue} }, $cb;
		500	return;
55	}	501	}
56	);
57
58	$cv->wait;
59
60	=head1 DESCRIPTION
61
62	This module is a helper module to make it easier to do non-blocking I/O
63	on filehandles (and sockets, see L<AnyEvent::Socket>).
64
65	The event loop is provided by L<AnyEvent>.
66
67	=head1 METHODS
68
69	=over 4
70
71	=item B<new (%args)>
72
73	The constructor has these arguments:
74
75	=over 4
76
77	=item fh => $filehandle
78
79	The filehandle this L<AnyEvent::Handle> object will operate on.
80
81	NOTE: The filehandle will be set to non-blocking.
82
83	=item read_block_size => $size
84
85	The default read block size use for reads via the C<on_read>
86	method.
87
88	=item on_read => $cb
89
90	=item on_eof => $cb
91
92	=item on_error => $cb
93
94	These are shortcuts, that will call the corresponding method and set the callback to C<$cb>.
95
96	=item on_readline => $cb
97
98	The C<readlines> method is called with the default seperator and C<$cb> as callback
99	for you.
100
101	=back
102
103	=cut
104
105	sub new {
106	my $this = shift;
107	my $class = ref($this) \|\| $this;
108	my $self = {
109	read_block_size => 4096,
110	rbuf => '',
111	@_
112	};
113	bless $self, $class;
114
115	$self->{fh}->blocking (0) if $self->{fh};
116
117	if ($self->{on_read}) {
118	$self->on_read ($self->{on_read});
119
120	} elsif ($self->{on_readline}) {	502	} elsif ($self->{on_read}) {
121	$self->readlines ($self->{on_readline});	503	$self->{on_read}($self);
122		504
		505	if (
		506	$self->{_eof} # if no further data will arrive
		507	&& $len == length $self->{rbuf} # and no data has been consumed
		508	&& !@{ $self->{_queue} } # and the queue is still empty
		509	&& $self->{on_read} # and we still want to read data
		510	) {
		511	# then no progress can be made
		512	$! = &Errno::EPIPE;
		513	$self->error;
		514	}
		515	} else {
		516	# read side becomes idle
		517	delete $self->{_rw};
		518	return;
		519	}
		520	}
		521
123	} elsif ($self->{on_eof}) {	522	if ($self->{_eof}) {
124	$self->on_eof ($self->{on_eof});	523	$self->_shutdown;
125		524	$self->{on_eof}($self)
126	} elsif ($self->{on_error}) {	525	if $self->{on_eof};
127	$self->on_eof ($self->{on_error});
128	}	526	}
129
130	return $self
131	}	527	}
132		528
133	=item B<fh>	529	=item $handle->on_read ($cb)
134		530
135	This method returns the filehandle of the L<AnyEvent::Handle> object.	531	This replaces the currently set C<on_read> callback, or clears it (when
136		532	the new callback is C<undef>). See the description of C<on_read> in the
137	=cut	533	constructor.
138
139	sub fh { $_[0]->{fh} }
140
141	=item B<on_read ($callback)>
142
143	This method installs a C<$callback> that will be called
144	when new data arrived. You can access the read buffer via the C<rbuf>
145	method (see below).
146
147	The first argument of the C<$callback> will be the L<AnyEvent::Handle> object.
148		534
149	=cut	535	=cut
150		536
151	sub on_read {	537	sub on_read {
152	my ($self, $cb) = @_;	538	my ($self, $cb) = @_;
		539
153	$self->{on_read} = $cb;	540	$self->{on_read} = $cb;
		541	}
154		542
155	unless (defined $self->{on_read}) {	543	=item $handle->rbuf
156	delete $self->{on_read_w};	544
157	return;	545	Returns the read buffer (as a modifiable lvalue).
		546
		547	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
		548	you want.
		549
		550	NOTE: The read buffer should only be used or modified if the C<on_read>,
		551	C<push_read> or C<unshift_read> methods are used. The other read methods
		552	automatically manage the read buffer.
		553
		554	=cut
		555
		556	sub rbuf : lvalue {
		557	$_[0]{rbuf}
		558	}
		559
		560	=item $handle->push_read ($cb)
		561
		562	=item $handle->unshift_read ($cb)
		563
		564	Append the given callback to the end of the queue (C<push_read>) or
		565	prepend it (C<unshift_read>).
		566
		567	The callback is called each time some additional read data arrives.
		568
		569	It must check whether enough data is in the read buffer already.
		570
		571	If not enough data is available, it must return the empty list or a false
		572	value, in which case it will be called repeatedly until enough data is
		573	available (or an error condition is detected).
		574
		575	If enough data was available, then the callback must remove all data it is
		576	interested in (which can be none at all) and return a true value. After returning
		577	true, it will be removed from the queue.
		578
		579	=cut
		580
		581	our %RH;
		582
		583	sub register_read_type($$) {
		584	$RH{$_[0]} = $_[1];
		585	}
		586
		587	sub push_read {
		588	my $self = shift;
		589	my $cb = pop;
		590
		591	if (@_) {
		592	my $type = shift;
		593
		594	$cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
		595	->($self, $cb, @_);
		596	}
		597
		598	push @{ $self->{_queue} }, $cb;
		599	$self->_drain_rbuf;
		600	}
		601
		602	sub unshift_read {
		603	my $self = shift;
		604	my $cb = pop;
		605
		606	if (@_) {
		607	my $type = shift;
		608
		609	$cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
		610	->($self, $cb, @_);
		611	}
		612
		613
		614	unshift @{ $self->{_queue} }, $cb;
		615	$self->_drain_rbuf;
		616	}
		617
		618	=item $handle->push_read (type => @args, $cb)
		619
		620	=item $handle->unshift_read (type => @args, $cb)
		621
		622	Instead of providing a callback that parses the data itself you can chose
		623	between a number of predefined parsing formats, for chunks of data, lines
		624	etc.
		625
		626	Predefined types are (if you have ideas for additional types, feel free to
		627	drop by and tell us):
		628
		629	=over 4
		630
		631	=item chunk => $octets, $cb->($self, $data)
		632
		633	Invoke the callback only once C<$octets> bytes have been read. Pass the
		634	data read to the callback. The callback will never be called with less
		635	data.
		636
		637	Example: read 2 bytes.
		638
		639	$handle->push_read (chunk => 2, sub {
		640	warn "yay ", unpack "H*", $_[1];
158	}	641	});
159		642
160	$self->{on_read_w} =	643	=cut
161	AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub {	644
162	#d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n";	645	register_read_type chunk => sub {
163	my $rbuf_len = length $self->{rbuf};	646	my ($self, $cb, $len) = @_;
164	my $l;	647
165	if (defined $self->{read_size}) {	648	sub {
166	$l = sysread $self->{fh}, $self->{rbuf},	649	$len <= length $_[0]{rbuf} or return;
167	($self->{read_size} - $rbuf_len), $rbuf_len;	650	$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
168	} else {	651	1
169	$l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len;	652	}
		653	};
		654
		655	# compatibility with older API
		656	sub push_read_chunk {
		657	$_[0]->push_read (chunk => $_[1], $_[2]);
		658	}
		659
		660	sub unshift_read_chunk {
		661	$_[0]->unshift_read (chunk => $_[1], $_[2]);
		662	}
		663
		664	=item line => [$eol, ]$cb->($self, $line, $eol)
		665
		666	The callback will be called only once a full line (including the end of
		667	line marker, C<$eol>) has been read. This line (excluding the end of line
		668	marker) will be passed to the callback as second argument (C<$line>), and
		669	the end of line marker as the third argument (C<$eol>).
		670
		671	The end of line marker, C<$eol>, can be either a string, in which case it
		672	will be interpreted as a fixed record end marker, or it can be a regex
		673	object (e.g. created by C<qr>), in which case it is interpreted as a
		674	regular expression.
		675
		676	The end of line marker argument C<$eol> is optional, if it is missing (NOT
		677	undef), then C<qr\|\015?\012\|> is used (which is good for most internet
		678	protocols).
		679
		680	Partial lines at the end of the stream will never be returned, as they are
		681	not marked by the end of line marker.
		682
		683	=cut
		684
		685	register_read_type line => sub {
		686	my ($self, $cb, $eol) = @_;
		687
		688	$eol = qr\|(\015?\012)\| if @_ < 3;
		689	$eol = quotemeta $eol unless ref $eol;
		690	$eol = qr\|^(.*?)($eol)\|s;
		691
		692	sub {
		693	$_[0]{rbuf} =~ s/$eol// or return;
		694
		695	$cb->($_[0], $1, $2);
		696	1
		697	}
		698	};
		699
		700	# compatibility with older API
		701	sub push_read_line {
		702	my $self = shift;
		703	$self->push_read (line => @_);
		704	}
		705
		706	sub unshift_read_line {
		707	my $self = shift;
		708	$self->unshift_read (line => @_);
		709	}
		710
		711	=item netstring => $cb->($string)
		712
		713	A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
		714
		715	Throws an error with C<$!> set to EBADMSG on format violations.
		716
		717	=cut
		718
		719	register_read_type netstring => sub {
		720	my ($self, $cb) = @_;
		721
		722	sub {
		723	unless ($_[0]{rbuf} =~ s/^(0\|[1-9][0-9]*)://) {
		724	if ($_[0]{rbuf} =~ /[^0-9]/) {
		725	$! = &Errno::EBADMSG;
		726	$self->error;
170	}	727	}
171	#d# warn "READL $l [$self->{rbuf}]\n";	728	return;
		729	}
172		730
173	if (not defined $l) {	731	my $len = $1;
174	return if $! == EAGAIN \|\| $! == EINTR;
175	$self->{on_error}->($self) if $self->{on_error};
176	delete $self->{on_read_w};
177		732
178	} elsif ($l == 0) {	733	$self->unshift_read (chunk => $len, sub {
179	$self->{on_eof}->($self) if $self->{on_eof};	734	my $string = $_[1];
180	delete $self->{on_read_w};	735	$_[0]->unshift_read (chunk => 1, sub {
181		736	if ($_[1] eq ",") {
		737	$cb->($_[0], $string);
182	} else {	738	} else {
183	$self->{on_read}->($self);	739	$! = &Errno::EBADMSG;
		740	$self->error;
		741	}
		742	});
		743	});
		744
		745	1
		746	}
		747	};
		748
		749	=item regex => $accept[, $reject[, $skip], $cb->($data)
		750
		751	Makes a regex match against the regex object C<$accept> and returns
		752	everything up to and including the match.
		753
		754	Example: read a single line terminated by '\n'.
		755
		756	$handle->push_read (regex => qr<\n>, sub { ... });
		757
		758	If C<$reject> is given and not undef, then it determines when the data is
		759	to be rejected: it is matched against the data when the C<$accept> regex
		760	does not match and generates an C<EBADMSG> error when it matches. This is
		761	useful to quickly reject wrong data (to avoid waiting for a timeout or a
		762	receive buffer overflow).
		763
		764	Example: expect a single decimal number followed by whitespace, reject
		765	anything else (not the use of an anchor).
		766
		767	$handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
		768
		769	If C<$skip> is given and not C<undef>, then it will be matched against
		770	the receive buffer when neither C<$accept> nor C<$reject> match,
		771	and everything preceding and including the match will be accepted
		772	unconditionally. This is useful to skip large amounts of data that you
		773	know cannot be matched, so that the C<$accept> or C<$reject> regex do not
		774	have to start matching from the beginning. This is purely an optimisation
		775	and is usually worth only when you expect more than a few kilobytes.
		776
		777	Example: expect a http header, which ends at C<\015\012\015\012>. Since we
		778	expect the header to be very large (it isn't in practise, but...), we use
		779	a skip regex to skip initial portions. The skip regex is tricky in that
		780	it only accepts something not ending in either \015 or \012, as these are
		781	required for the accept regex.
		782
		783	$handle->push_read (regex =>
		784	qr<\015\012\015\012>,
		785	undef, # no reject
		786	qr<^.*[^\015\012]>,
		787	sub { ... });
		788
		789	=cut
		790
		791	register_read_type regex => sub {
		792	my ($self, $cb, $accept, $reject, $skip) = @_;
		793
		794	my $data;
		795	my $rbuf = \$self->{rbuf};
		796
		797	sub {
		798	# accept
		799	if ($$rbuf =~ $accept) {
		800	$data .= substr $$rbuf, 0, $+[0], "";
		801	$cb->($self, $data);
		802	return 1;
		803	}
		804
		805	# reject
		806	if ($reject && $$rbuf =~ $reject) {
		807	$! = &Errno::EBADMSG;
		808	$self->error;
		809	}
		810
		811	# skip
		812	if ($skip && $$rbuf =~ $skip) {
		813	$data .= substr $$rbuf, 0, $+[0], "";
		814	}
		815
		816	()
		817	}
		818	};
		819
		820	=back
		821
		822	=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
		823
		824	This function (not method) lets you add your own types to C<push_read>.
		825
		826	Whenever the given C<type> is used, C<push_read> will invoke the code
		827	reference with the handle object, the callback and the remaining
		828	arguments.
		829
		830	The code reference is supposed to return a callback (usually a closure)
		831	that works as a plain read callback (see C<< ->push_read ($cb) >>).
		832
		833	It should invoke the passed callback when it is done reading (remember to
		834	pass C<$self> as first argument as all other callbacks do that).
		835
		836	Note that this is a function, and all types registered this way will be
		837	global, so try to use unique names.
		838
		839	For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
		840	search for C<register_read_type>)).
		841
		842	=item $handle->stop_read
		843
		844	=item $handle->start_read
		845
		846	In rare cases you actually do not want to read anything from the
		847	socket. In this case you can call C<stop_read>. Neither C<on_read> no
		848	any queued callbacks will be executed then. To start reading again, call
		849	C<start_read>.
		850
		851	=cut
		852
		853	sub stop_read {
		854	my ($self) = @_;
		855
		856	delete $self->{_rw};
		857	}
		858
		859	sub start_read {
		860	my ($self) = @_;
		861
		862	unless ($self->{_rw} \|\| $self->{_eof}) {
		863	Scalar::Util::weaken $self;
		864
		865	$self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
		866	my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
		867	my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} \|\| 8192, length $$rbuf;
		868
		869	if ($len > 0) {
		870	$self->{filter_r}
		871	? $self->{filter_r}->($self, $rbuf)
		872	: $self->_drain_rbuf;
		873
		874	} elsif (defined $len) {
		875	delete $self->{_rw};
		876	$self->{_eof} = 1;
		877	$self->_drain_rbuf;
		878
		879	} elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
		880	return $self->error;
184	}	881	}
185	});	882	});
		883	}
186	}	884	}
187		885
188	=item B<on_error ($callback)>	886	sub _dotls {
189
190	Whenever a read or write operation resulted in an error the C<$callback>
191	will be called.
192
193	The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself.
194	The error is given as errno in C<$!>.
195
196	=cut
197
198	sub on_error {
199	$_[0]->{on_error} = $_[1];
200	}
201
202	=item B<on_eof ($callback)>
203
204	Installs the C<$callback> that will be called when the end of file is
205	encountered in a read operation this C<$callback> will be called. The first
206	argument will be the L<AnyEvent::Handle> object itself.
207
208	=cut
209
210	sub on_eof {
211	$_[0]->{on_eof} = $_[1];
212	}
213
214	=item B<rbuf>
215
216	Returns a reference to the read buffer.
217
218	NOTE: The read buffer should only be used or modified if the C<on_read>
219	method is used directly. The C<read> and C<readlines> methods will provide
220	the read data to their callbacks.
221
222	=cut
223
224	sub rbuf : lvalue {
225	$_[0]->{rbuf}
226	}
227
228	=item B<read ($len, $callback)>
229
230	Will read exactly C<$len> bytes from the filehandle and call the C<$callback>
231	if done so. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
232	object itself and the second argument the read data.
233
234	NOTE: This method will override any callbacks installed via the C<on_read> method.
235
236	=cut
237
238	sub read {
239	my ($self, $len, $cb) = @_;	887	my ($self) = @_;
240		888
241	$self->{read_cb} = $cb;	889	if (length $self->{_tls_wbuf}) {
242	my $old_blk_size = $self->{read_block_size};	890	while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
243	$self->{read_block_size} = $len;	891	substr $self->{_tls_wbuf}, 0, $len, "";
244
245	$self->on_read (sub {
246	#d# warn "OFOFO $len \|\| ".length($_[0]->{rbuf})."\|\|\n";
247
248	if ($len == length $_[0]->{rbuf}) {
249	$_[0]->{read_block_size} = $old_blk_size;
250	$_[0]->on_read (undef);
251	$_[0]->{read_cb}->($_[0], (substr $self->{rbuf}, 0, $len, ''));
252	}	892	}
253	});	893	}
254	}
255		894
256	=item B<readlines ($callback)>	895	if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
257		896	$self->{wbuf} .= $buf;
258	=item B<readlines ($sep, $callback)>	897	$self->_drain_wbuf;
259
260	This method will read lines from the filehandle, seperated by C<$sep> or C<"\n">
261	if C<$sep> is not provided. C<$sep> will be used as "line" seperator.
262
263	The C<$callback> will be called when at least one
264	line could be read. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
265	object itself and the rest of the arguments will be the read lines.
266
267	NOTE: This method will override any callbacks installed via the C<on_read> method.
268
269	=cut
270
271	sub readlines {
272	my ($self, $sep, $cb) = @_;
273
274	if (ref $sep) {
275	$cb = $sep;
276	$sep = "\n";
277
278	} elsif (not defined $sep) {
279	$sep = "\n";
280	}	898	}
281		899
282	my $sep_len = length $sep;	900	while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
		901	$self->{rbuf} .= $buf;
		902	$self->_drain_rbuf;
		903	}
283		904
284	$self->{on_readline} = $cb;	905	my $err = Net::SSLeay::get_error ($self->{tls}, -1);
285		906
286	$self->on_read (sub {	907	if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
287	my @lines;	908	if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
288	my $rb = \$_[0]->{rbuf};	909	$self->error;
289	my $pos;	910	} elsif ($err == Net::SSLeay::ERROR_SSL ()) {
290	while (($pos = index ($$rb, $sep)) >= 0) {	911	$! = &Errno::EIO;
291	push @lines, substr $$rb, 0, $pos + $sep_len, '';	912	$self->error;
292	}	913	}
293	$self->{on_readline}->($_[0], @lines);	914
		915	# all others are fine for our purposes
		916	}
		917	}
		918
		919	=item $handle->starttls ($tls[, $tls_ctx])
		920
		921	Instead of starting TLS negotiation immediately when the AnyEvent::Handle
		922	object is created, you can also do that at a later time by calling
		923	C<starttls>.
		924
		925	The first argument is the same as the C<tls> constructor argument (either
		926	C<"connect">, C<"accept"> or an existing Net::SSLeay object).
		927
		928	The second argument is the optional C<Net::SSLeay::CTX> object that is
		929	used when AnyEvent::Handle has to create its own TLS connection object.
		930
		931	The TLS connection object will end up in C<< $handle->{tls} >> after this
		932	call and can be used or changed to your liking. Note that the handshake
		933	might have already started when this function returns.
		934
		935	=cut
		936
		937	# TODO: maybe document...
		938	sub starttls {
		939	my ($self, $ssl, $ctx) = @_;
		940
		941	$self->stoptls;
		942
		943	if ($ssl eq "accept") {
		944	$ssl = Net::SSLeay::new ($ctx \|\| TLS_CTX ());
		945	Net::SSLeay::set_accept_state ($ssl);
		946	} elsif ($ssl eq "connect") {
		947	$ssl = Net::SSLeay::new ($ctx \|\| TLS_CTX ());
		948	Net::SSLeay::set_connect_state ($ssl);
		949	}
		950
		951	$self->{tls} = $ssl;
		952
		953	# basically, this is deep magic (because SSL_read should have the same issues)
		954	# but the openssl maintainers basically said: "trust us, it just works".
		955	# (unfortunately, we have to hardcode constants because the abysmally misdesigned
		956	# and mismaintained ssleay-module doesn't even offer them).
		957	# http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
		958	Net::SSLeay::CTX_set_mode ($self->{tls},
		959	(eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } \|\| 1)
		960	\| (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } \|\| 2));
		961
		962	$self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
		963	$self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
		964
		965	Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
		966
		967	$self->{filter_w} = sub {
		968	$_[0]{_tls_wbuf} .= ${$_[1]};
		969	&_dotls;
294	});	970	};
		971	$self->{filter_r} = sub {
		972	Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
		973	&_dotls;
		974	};
295	}	975	}
296		976
297	=item B<write ($data)>	977	=item $handle->stoptls
298		978
299	=item B<write ($callback)>	979	Destroys the SSL connection, if any. Partial read or write data will be
		980	lost.
300		981
301	=item B<write ($data, $callback)>
302
303	This method will write C<$data> to the filehandle and call the C<$callback>
304	afterwards. If only C<$callback> is provided it will be called when the
305	write buffer becomes empty the next time (or immediately if it already is empty).
306
307	=cut	982	=cut
308		983
309	sub write {	984	sub stoptls {
310	my ($self, $data, $cb) = @_;
311	if (ref $data) { $cb = $data; undef $data }
312	push @{$self->{write_bufs}}, [$data, $cb];
313	$self->_check_writer;
314	}
315
316	sub _check_writer {
317	my ($self) = @_;	985	my ($self) = @_;
318		986
319	if ($self->{write_w}) {	987	Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
320	unless ($self->{write_cb}) {
321	while (@{$self->{write_bufs}} && not defined $self->{write_bufs}->[0]->[1]) {
322	my $wba = shift @{$self->{write_bufs}};
323	$self->{wbuf} .= $wba->[0];
324	}
325	}
326	return;
327	}
328		988
329	my $wba = shift @{$self->{write_bufs}}	989	delete $self->{_rbio};
330	or return;	990	delete $self->{_wbio};
331		991	delete $self->{_tls_wbuf};
332	unless (defined $wba->[0]) {	992	delete $self->{filter_r};
333	$wba->[1]->($self) if $wba->[1];
334	$self->_check_writer;
335	return;
336	}
337
338	$self->{wbuf} = $wba->[0];
339	$self->{write_cb} = $wba->[1];
340
341	$self->{write_w} =
342	AnyEvent->io (poll => 'w', fh => $self->{fh}, cb => sub {
343	my $l = syswrite $self->{fh}, $self->{wbuf}, length $self->{wbuf};
344
345	if (not defined $l) {
346	return if $! == EAGAIN \|\| $! == EINTR;
347	delete $self->{write_w};	993	delete $self->{filter_w};
348	$self->{on_error}->($self) if $self->{on_error};	994	}
349		995
350	} else {	996	sub DESTROY {
351	substr $self->{wbuf}, 0, $l, '';	997	my $self = shift;
352		998
353	if (length ($self->{wbuf}) == 0) {	999	$self->stoptls;
354	$self->{write_cb}->($self) if $self->{write_cb};	1000	}
355		1001
356	delete $self->{write_w};	1002	=item AnyEvent::Handle::TLS_CTX
357	delete $self->{wbuf};
358	delete $self->{write_cb};
359		1003
360	$self->_check_writer;	1004	This function creates and returns the Net::SSLeay::CTX object used by
361	}	1005	default for TLS mode.
362	}	1006
363	});	1007	The context is created like this:
		1008
		1009	Net::SSLeay::load_error_strings;
		1010	Net::SSLeay::SSLeay_add_ssl_algorithms;
		1011	Net::SSLeay::randomize;
		1012
		1013	my $CTX = Net::SSLeay::CTX_new;
		1014
		1015	Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
		1016
		1017	=cut
		1018
		1019	our $TLS_CTX;
		1020
		1021	sub TLS_CTX() {
		1022	$TLS_CTX \|\| do {
		1023	require Net::SSLeay;
		1024
		1025	Net::SSLeay::load_error_strings ();
		1026	Net::SSLeay::SSLeay_add_ssl_algorithms ();
		1027	Net::SSLeay::randomize ();
		1028
		1029	$TLS_CTX = Net::SSLeay::CTX_new ();
		1030
		1031	Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
		1032
		1033	$TLS_CTX
		1034	}
364	}	1035	}
365		1036
366	=back	1037	=back
367		1038
		1039	=head1 SUBCLASSING AnyEvent::Handle
		1040
		1041	In many cases, you might want to subclass AnyEvent::Handle.
		1042
		1043	To make this easier, a given version of AnyEvent::Handle uses these
		1044	conventions:
		1045
		1046	=over 4
		1047
		1048	=item * all constructor arguments become object members.
		1049
		1050	At least initially, when you pass a C<tls>-argument to the constructor it
		1051	will end up in C<< $handle->{tls} >>. Those members might be changes or
		1052	mutated later on (for example C<tls> will hold the TLS connection object).
		1053
		1054	=item * other object member names are prefixed with an C<_>.
		1055
		1056	All object members not explicitly documented (internal use) are prefixed
		1057	with an underscore character, so the remaining non-C<_>-namespace is free
		1058	for use for subclasses.
		1059
		1060	=item * all members not documented here and not prefixed with an underscore
		1061	are free to use in subclasses.
		1062
		1063	Of course, new versions of AnyEvent::Handle may introduce more "public"
		1064	member variables, but thats just life, at least it is documented.
		1065
		1066	=back
		1067
368	=head1 AUTHOR	1068	=head1 AUTHOR
369		1069
370	Robin Redeker, C<< <elmex at ta-sa.org> >>	1070	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
371		1071
372	=cut	1072	=cut
373		1073
374	1; # End of AnyEvent::Handle	1074	1; # End of AnyEvent::Handle

Diff Legend

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

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents): Revision 1.5 by elmex, Mon Apr 28 08:01:05 2008 UTC vs. Revision 1.38 by root, Mon May 26 21:28:33 2008 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.5 by elmex, Mon Apr 28 08:01:05 2008 UTC vs.
Revision 1.38 by root, Mon May 26 21:28:33 2008 UTC