[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.15 by root, Sat May 17 21:34:15 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 ();
		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 filehandles via AnyEvent
13		16
14	=head1 VERSION	17	This module is experimental.
15		18
16	Version 0.01
17
18	=cut	19	=cut
19		20
20	our $VERSION = '0.01';	21	our $VERSION = '0.04';
21		22
22	=head1 SYNOPSIS	23	=head1 SYNOPSIS
23		24
24	use AnyEvent;	25	use AnyEvent;
25	use AnyEvent::Handle;	26	use AnyEvent::Handle;
26		27
27	my $cv = AnyEvent->condvar;	28	my $cv = AnyEvent->condvar;
28		29
29	my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);	30	my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
30		31
31	$ae_fh->on_eof (sub { $cv->broadcast });	32	#TODO
32
33	$ae_fh->readlines (sub {
34	my ($ae_fh, @lines) = @_;
35	for (@lines) {
36	chomp;
37	print "Line: $_";
38	}
39	});
40		33
41	# or use the constructor to pass the callback:	34	# or use the constructor to pass the callback:
42		35
43	my $ae_fh2 =	36	my $ae_fh2 =
44	AnyEvent::Handle->new (	37	AnyEvent::Handle->new (
45	fh => \*STDIN,	38	fh => \*STDIN,
46	on_eof => sub {	39	on_eof => sub {
47	$cv->broadcast;	40	$cv->broadcast;
48	},	41	},
49	on_readline => sub {	42	#TODO
50	my ($ae_fh, @lines) = @_;	43	);
51	for (@lines) {	44
52	chomp;	45	$cv->wait;
53	print "Line: $_";	46
		47	=head1 DESCRIPTION
		48
		49	This module is a helper module to make it easier to do event-based I/O on
		50	filehandles. For utility functions for doing non-blocking connects and accepts
		51	on sockets see L<AnyEvent::Util>.
		52
		53	In the following, when the documentation refers to of "bytes" then this
		54	means characters. As sysread and syswrite are used for all I/O, their
		55	treatment of characters applies to this module as well.
		56
		57	All callbacks will be invoked with the handle object as their first
		58	argument.
		59
		60	=head1 METHODS
		61
		62	=over 4
		63
		64	=item B<new (%args)>
		65
		66	The constructor supports these arguments (all as key => value pairs).
		67
		68	=over 4
		69
		70	=item fh => $filehandle [MANDATORY]
		71
		72	The filehandle this L<AnyEvent::Handle> object will operate on.
		73
		74	NOTE: The filehandle will be set to non-blocking (using
		75	AnyEvent::Util::fh_nonblocking).
		76
		77	=item on_eof => $cb->($self) [MANDATORY]
		78
		79	Set the callback to be called on EOF.
		80
		81	=item on_error => $cb->($self)
		82
		83	This is the fatal error callback, that is called when, well, a fatal error
		84	ocurs, such as not being able to resolve the hostname, failure to connect
		85	or a read error.
		86
		87	The object will not be in a usable state when this callback has been
		88	called.
		89
		90	On callback entrance, the value of C<$!> contains the operating system
		91	error (or C<ENOSPC> or C<EPIPE>).
		92
		93	While not mandatory, it is I<highly> recommended to set this callback, as
		94	you will not be notified of errors otherwise. The default simply calls
		95	die.
		96
		97	=item on_read => $cb->($self)
		98
		99	This sets the default read callback, which is called when data arrives
		100	and no read request is in the queue.
		101
		102	To access (and remove data from) the read buffer, use the C<< ->rbuf >>
		103	method or acces sthe C<$self->{rbuf}> member directly.
		104
		105	When an EOF condition is detected then AnyEvent::Handle will first try to
		106	feed all the remaining data to the queued callbacks and C<on_read> before
		107	calling the C<on_eof> callback. If no progress can be made, then a fatal
		108	error will be raised (with C<$!> set to C<EPIPE>).
		109
		110	=item on_drain => $cb->()
		111
		112	This sets the callback that is called when the write buffer becomes empty
		113	(or when the callback is set and the buffer is empty already).
		114
		115	To append to the write buffer, use the C<< ->push_write >> method.
		116
		117	=item rbuf_max => <bytes>
		118
		119	If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
		120	when the read buffer ever (strictly) exceeds this size. This is useful to
		121	avoid denial-of-service attacks.
		122
		123	For example, a server accepting connections from untrusted sources should
		124	be configured to accept only so-and-so much data that it cannot act on
		125	(for example, when expecting a line, an attacker could send an unlimited
		126	amount of data without a callback ever being called as long as the line
		127	isn't finished).
		128
		129	=item read_size => <bytes>
		130
		131	The default read block size (the amount of bytes this module will try to read
		132	on each [loop iteration). Default: C<4096>.
		133
		134	=item low_water_mark => <bytes>
		135
		136	Sets the amount of bytes (default: C<0>) that make up an "empty" write
		137	buffer: If the write reaches this size or gets even samller it is
		138	considered empty.
		139
		140	=back
		141
		142	=cut
		143
		144	sub new {
		145	my $class = shift;
		146
		147	my $self = bless { @_ }, $class;
		148
		149	$self->{fh} or Carp::croak "mandatory argument fh is missing";
		150
		151	AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
		152
		153	$self->on_eof ((delete $self->{on_eof} ) or Carp::croak "mandatory argument on_eof is missing");
		154
		155	$self->on_error (delete $self->{on_error}) if $self->{on_error};
		156	$self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
		157	$self->on_read (delete $self->{on_read} ) if $self->{on_read};
		158
		159	$self->start_read;
		160
		161	$self
		162	}
		163
		164	sub _shutdown {
		165	my ($self) = @_;
		166
		167	delete $self->{rw};
		168	delete $self->{ww};
		169	delete $self->{fh};
		170	}
		171
		172	sub error {
		173	my ($self) = @_;
		174
		175	{
		176	local $!;
		177	$self->_shutdown;
		178	}
		179
		180	if ($self->{on_error}) {
		181	$self->{on_error}($self);
		182	} else {
		183	die "AnyEvent::Handle uncaught fatal error: $!";
		184	}
		185	}
		186
		187	=item $fh = $handle->fh
		188
		189	This method returns the filehandle of the L<AnyEvent::Handle> object.
		190
		191	=cut
		192
		193	sub fh { $_[0]->{fh} }
		194
		195	=item $handle->on_error ($cb)
		196
		197	Replace the current C<on_error> callback (see the C<on_error> constructor argument).
		198
		199	=cut
		200
		201	sub on_error {
		202	$_[0]{on_error} = $_[1];
		203	}
		204
		205	=item $handle->on_eof ($cb)
		206
		207	Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
		208
		209	=cut
		210
		211	sub on_eof {
		212	$_[0]{on_eof} = $_[1];
		213	}
		214
		215	#############################################################################
		216
		217	=back
		218
		219	=head2 WRITE QUEUE
		220
		221	AnyEvent::Handle manages two queues per handle, one for writing and one
		222	for reading.
		223
		224	The write queue is very simple: you can add data to its end, and
		225	AnyEvent::Handle will automatically try to get rid of it for you.
		226
		227	When data could be writtena nd the write buffer is shorter then the low
		228	water mark, the C<on_drain> callback will be invoked.
		229
		230	=over 4
		231
		232	=item $handle->on_drain ($cb)
		233
		234	Sets the C<on_drain> callback or clears it (see the description of
		235	C<on_drain> in the constructor).
		236
		237	=cut
		238
		239	sub on_drain {
		240	my ($self, $cb) = @_;
		241
		242	$self->{on_drain} = $cb;
		243
		244	$cb->($self)
		245	if $cb && $self->{low_water_mark} >= length $self->{wbuf};
		246	}
		247
		248	=item $handle->push_write ($data)
		249
		250	Queues the given scalar to be written. You can push as much data as you
		251	want (only limited by the available memory), as C<AnyEvent::Handle>
		252	buffers it independently of the kernel.
		253
		254	=cut
		255
		256	sub push_write {
		257	my ($self, $data) = @_;
		258
		259	$self->{wbuf} .= $data;
		260
		261	unless ($self->{ww}) {
		262	Scalar::Util::weaken $self;
		263	my $cb = sub {
		264	my $len = syswrite $self->{fh}, $self->{wbuf};
		265
		266	if ($len > 0) {
		267	substr $self->{wbuf}, 0, $len, "";
		268
		269
		270	$self->{on_drain}($self)
		271	if $self->{low_water_mark} >= length $self->{wbuf}
		272	&& $self->{on_drain};
		273
		274	delete $self->{ww} unless length $self->{wbuf};
		275	} elsif ($! != EAGAIN && $! != EINTR) {
		276	$self->error;
		277	}
		278	};
		279
		280	$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
		281
		282	$cb->($self);
		283	};
		284	}
		285
		286	#############################################################################
		287
		288	=back
		289
		290	=head2 READ QUEUE
		291
		292	AnyEvent::Handle manages two queues per handle, one for writing and one
		293	for reading.
		294
		295	The read queue is more complex than the write queue. It can be used in two
		296	ways, the "simple" way, using only C<on_read> and the "complex" way, using
		297	a queue.
		298
		299	In the simple case, you just install an C<on_read> callback and whenever
		300	new data arrives, it will be called. You can then remove some data (if
		301	enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
		302	or not.
		303
		304	In the more complex case, you want to queue multiple callbacks. In this
		305	case, AnyEvent::Handle will call the first queued callback each time new
		306	data arrives and removes it when it has done its job (see C<push_read>,
		307	below).
		308
		309	This way you can, for example, push three line-reads, followed by reading
		310	a chunk of data, and AnyEvent::Handle will execute them in order.
		311
		312	Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
		313	the specified number of bytes which give an XML datagram.
		314
		315	# in the default state, expect some header bytes
		316	$handle->on_read (sub {
		317	# some data is here, now queue the length-header-read (4 octets)
		318	shift->unshift_read_chunk (4, sub {
		319	# header arrived, decode
		320	my $len = unpack "N", $_[1];
		321
		322	# now read the payload
		323	shift->unshift_read_chunk ($len, sub {
		324	my $xml = $_[1];
		325	# handle xml
		326	});
		327	});
		328	});
		329
		330	Example 2: Implement a client for a protocol that replies either with
		331	"OK" and another line or "ERROR" for one request, and 64 bytes for the
		332	second request. Due tot he availability of a full queue, we can just
		333	pipeline sending both requests and manipulate the queue as necessary in
		334	the callbacks:
		335
		336	# request one
		337	$handle->push_write ("request 1\015\012");
		338
		339	# we expect "ERROR" or "OK" as response, so push a line read
		340	$handle->push_read_line (sub {
		341	# if we got an "OK", we have to _prepend_ another line,
		342	# so it will be read before the second request reads its 64 bytes
		343	# which are already in the queue when this callback is called
		344	# we don't do this in case we got an error
		345	if ($_[1] eq "OK") {
		346	$_[0]->unshift_read_line (sub {
		347	my $response = $_[1];
		348	...
		349	});
		350	}
		351	});
		352
		353	# request two
		354	$handle->push_write ("request 2\015\012");
		355
		356	# simply read 64 bytes, always
		357	$handle->push_read_chunk (64, sub {
		358	my $response = $_[1];
		359	...
		360	});
		361
		362	=over 4
		363
		364	=cut
		365
		366	sub _drain_rbuf {
		367	my ($self) = @_;
		368
		369	return if $self->{in_drain};
		370	local $self->{in_drain} = 1;
		371
		372	while (my $len = length $self->{rbuf}) {
		373	no strict 'refs';
		374	if (my $cb = shift @{ $self->{queue} }) {
		375	if (!$cb->($self)) {
		376	if ($self->{eof}) {
		377	# no progress can be made (not enough data and no data forthcoming)
		378	$! = &Errno::EPIPE; return $self->error;
54	}	379	}
		380
		381	unshift @{ $self->{queue} }, $cb;
		382	return;
55	}	383	}
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}) {	384	} elsif ($self->{on_read}) {
121	$self->readlines ($self->{on_readline});	385	$self->{on_read}($self);
122		386
123	} elsif ($self->{on_eof}) {	387	if (
124	$self->on_eof ($self->{on_eof});	388	$self->{eof} # if no further data will arrive
125		389	&& $len == length $self->{rbuf} # and no data has been consumed
126	} elsif ($self->{on_error}) {	390	&& !@{ $self->{queue} } # and the queue is still empty
127	$self->on_eof ($self->{on_error});	391	&& $self->{on_read} # and we still want to read data
		392	) {
		393	# then no progress can be made
		394	$! = &Errno::EPIPE; return $self->error;
		395	}
		396	} else {
		397	# read side becomes idle
		398	delete $self->{rw};
		399	return;
		400	}
128	}	401	}
129		402
130	return $self	403	if ($self->{eof}) {
		404	$self->_shutdown;
		405	$self->{on_eof}($self);
		406	}
131	}	407	}
132		408
133	=item B<fh>	409	=item $handle->on_read ($cb)
134		410
135	This method returns the filehandle of the L<AnyEvent::Handle> object.	411	This replaces the currently set C<on_read> callback, or clears it (when
136		412	the new callback is C<undef>). See the description of C<on_read> in the
137	=cut	413	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		414
149	=cut	415	=cut
150		416
151	sub on_read {	417	sub on_read {
152	my ($self, $cb) = @_;	418	my ($self, $cb) = @_;
		419
153	$self->{on_read} = $cb;	420	$self->{on_read} = $cb;
		421	}
154		422
155	unless (defined $self->{on_read}) {	423	=item $handle->rbuf
156	delete $self->{on_read_w};	424
157	return;	425	Returns the read buffer (as a modifiable lvalue).
		426
		427	You can access the read buffer directly as the C<< ->{rbuf} >> member, if
		428	you want.
		429
		430	NOTE: The read buffer should only be used or modified if the C<on_read>,
		431	C<push_read> or C<unshift_read> methods are used. The other read methods
		432	automatically manage the read buffer.
		433
		434	=cut
		435
		436	sub rbuf : lvalue {
		437	$_[0]{rbuf}
		438	}
		439
		440	=item $handle->push_read ($cb)
		441
		442	=item $handle->unshift_read ($cb)
		443
		444	Append the given callback to the end of the queue (C<push_read>) or
		445	prepend it (C<unshift_read>).
		446
		447	The callback is called each time some additional read data arrives.
		448
		449	It must check wether enough data is in the read buffer already.
		450
		451	If not enough data is available, it must return the empty list or a false
		452	value, in which case it will be called repeatedly until enough data is
		453	available (or an error condition is detected).
		454
		455	If enough data was available, then the callback must remove all data it is
		456	interested in (which can be none at all) and return a true value. After returning
		457	true, it will be removed from the queue.
		458
		459	=cut
		460
		461	sub push_read {
		462	my ($self, $cb) = @_;
		463
		464	push @{ $self->{queue} }, $cb;
		465	$self->_drain_rbuf;
		466	}
		467
		468	sub unshift_read {
		469	my ($self, $cb) = @_;
		470
		471	push @{ $self->{queue} }, $cb;
		472	$self->_drain_rbuf;
		473	}
		474
		475	=item $handle->push_read_chunk ($len, $cb->($self, $data))
		476
		477	=item $handle->unshift_read_chunk ($len, $cb->($self, $data))
		478
		479	Append the given callback to the end of the queue (C<push_read_chunk>) or
		480	prepend it (C<unshift_read_chunk>).
		481
		482	The callback will be called only once C<$len> bytes have been read, and
		483	these C<$len> bytes will be passed to the callback.
		484
		485	=cut
		486
		487	sub _read_chunk($$) {
		488	my ($self, $len, $cb) = @_;
		489
		490	sub {
		491	$len <= length $_[0]{rbuf} or return;
		492	$cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
		493	1
158	}	494	}
159		495	}
160	$self->{on_read_w} =	496
161	AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub {	497	sub push_read_chunk {
162	#d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n";	498	$_[0]->push_read (&_read_chunk);
163	my $rbuf_len = length $self->{rbuf};	499	}
164	my $l;	500
		501
		502	sub unshift_read_chunk {
		503	$_[0]->unshift_read (&_read_chunk);
		504	}
		505
		506	=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
		507
		508	=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
		509
		510	Append the given callback to the end of the queue (C<push_read_line>) or
		511	prepend it (C<unshift_read_line>).
		512
		513	The callback will be called only once a full line (including the end of
		514	line marker, C<$eol>) has been read. This line (excluding the end of line
		515	marker) will be passed to the callback as second argument (C<$line>), and
		516	the end of line marker as the third argument (C<$eol>).
		517
		518	The end of line marker, C<$eol>, can be either a string, in which case it
		519	will be interpreted as a fixed record end marker, or it can be a regex
		520	object (e.g. created by C<qr>), in which case it is interpreted as a
		521	regular expression.
		522
		523	The end of line marker argument C<$eol> is optional, if it is missing (NOT
		524	undef), then C<qr\|\015?\012\|> is used (which is good for most internet
		525	protocols).
		526
		527	Partial lines at the end of the stream will never be returned, as they are
		528	not marked by the end of line marker.
		529
		530	=cut
		531
		532	sub _read_line($$) {
		533	my $self = shift;
		534	my $cb = pop;
		535	my $eol = @_ ? shift : qr\|(\015?\012)\|;
		536	my $pos;
		537
		538	$eol = quotemeta $eol unless ref $eol;
		539	$eol = qr\|^(.*?)($eol)\|s;
		540
		541	sub {
		542	$_[0]{rbuf} =~ s/$eol// or return;
		543
		544	$cb->($_[0], $1, $2);
		545	1
		546	}
		547	}
		548
		549	sub push_read_line {
		550	$_[0]->push_read (&_read_line);
		551	}
		552
		553	sub unshift_read_line {
		554	$_[0]->unshift_read (&_read_line);
		555	}
		556
		557	=item $handle->stop_read
		558
		559	=item $handle->start_read
		560
		561	In rare cases you actually do not want to read anything form the
		562	socket. In this case you can call C<stop_read>. Neither C<on_read> no
		563	any queued callbacks will be executed then. To start readign again, call
		564	C<start_read>.
		565
		566	=cut
		567
		568	sub stop_read {
		569	my ($self) = @_;
		570
		571	delete $self->{rw};
		572	}
		573
		574	sub start_read {
		575	my ($self) = @_;
		576
		577	unless ($self->{rw} \|\| $self->{eof}) {
		578	Scalar::Util::weaken $self;
		579
		580	$self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
		581	my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} \|\| 8192, length $self->{rbuf};
		582
		583	if ($len > 0) {
165	if (defined $self->{read_size}) {	584	if (defined $self->{rbuf_max}) {
166	$l = sysread $self->{fh}, $self->{rbuf},	585	if ($self->{rbuf_max} < length $self->{rbuf}) {
167	($self->{read_size} - $rbuf_len), $rbuf_len;	586	$! = &Errno::ENOSPC; return $self->error;
168	} else {	587	}
169	$l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len;	588	}
		589
		590	} elsif (defined $len) {
		591	$self->{eof} = 1;
		592	delete $self->{rw};
		593
		594	} elsif ($! != EAGAIN && $! != EINTR) {
		595	return $self->error;
170	}	596	}
171	#d# warn "READL $l [$self->{rbuf}]\n";
172		597
173	if (not defined $l) {	598	$self->_drain_rbuf;
174	return if $! == EAGAIN \|\| $! == EINTR;
175	$self->{on_error}->($self) if $self->{on_error};
176	delete $self->{on_read_w};
177
178	} elsif ($l == 0) {
179	$self->{on_eof}->($self) if $self->{on_eof};
180	delete $self->{on_read_w};
181
182	} else {
183	$self->{on_read}->($self);
184	}
185	});	599	});
186	}
187
188	=item B<on_error ($callback)>
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) = @_;
240
241	$self->{read_cb} = $cb;
242	my $old_blk_size = $self->{read_block_size};
243	$self->{read_block_size} = $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	}
253	});
254	}
255
256	=item B<readlines ($callback)>
257
258	=item B<readlines ($sep, $callback)>
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	}	600	}
281
282	my $sep_len = length $sep;
283
284	$self->{on_readline} = $cb;
285
286	$self->on_read (sub {
287	my @lines;
288	my $rb = \$_[0]->{rbuf};
289	my $pos;
290	while (($pos = index ($$rb, $sep)) >= 0) {
291	push @lines, substr $$rb, 0, $pos + $sep_len, '';
292	}
293	$self->{on_readline}->($_[0], @lines);
294	});
295	}
296
297	=item B<write ($data)>
298
299	=item B<write ($callback)>
300
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
308
309	sub write {
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) = @_;
318
319	if ($self->{write_w}) {
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
329	my $wba = shift @{$self->{write_bufs}}
330	or return;
331
332	unless (defined $wba->[0]) {
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};
348	$self->{on_error}->($self) if $self->{on_error};
349
350	} else {
351	substr $self->{wbuf}, 0, $l, '';
352
353	if (length ($self->{wbuf}) == 0) {
354	$self->{write_cb}->($self) if $self->{write_cb};
355
356	delete $self->{write_w};
357	delete $self->{wbuf};
358	delete $self->{write_cb};
359
360	$self->_check_writer;
361	}
362	}
363	});
364	}	601	}
365		602
366	=back	603	=back
367		604
368	=head1 AUTHOR	605	=head1 AUTHOR
369		606
370	Robin Redeker, C<< <elmex at ta-sa.org> >>	607	Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
371		608
372	=cut	609	=cut
373		610
374	1; # End of AnyEvent::Handle	611	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.15 by root, Sat May 17 21:34:15 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.15 by root, Sat May 17 21:34:15 2008 UTC