ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
Revision: 1.8
Committed: Fri May 2 15:36:10 2008 UTC (16 years, 1 month ago) by root
Branch: MAIN
Changes since 1.7: +332 -213 lines
Log Message:
*** empty log message ***

File Contents

# Content
1 package AnyEvent::Handle;
2
3 no warnings;
4 use strict;
5
6 use AnyEvent ();
7 use AnyEvent::Util ();
8 use Scalar::Util ();
9 use Carp ();
10 use Fcntl ();
11 use Errno qw/EAGAIN EINTR/;
12
13 =head1 NAME
14
15 AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent
16
17 =cut
18
19 our $VERSION = '0.02';
20
21 =head1 SYNOPSIS
22
23 use AnyEvent;
24 use AnyEvent::Handle;
25
26 my $cv = AnyEvent->condvar;
27
28 my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
29
30 #TODO
31
32 # or use the constructor to pass the callback:
33
34 my $ae_fh2 =
35 AnyEvent::Handle->new (
36 fh => \*STDIN,
37 on_eof => sub {
38 $cv->broadcast;
39 },
40 #TODO
41 );
42
43 $cv->wait;
44
45 =head1 DESCRIPTION
46
47 This module is a helper module to make it easier to do event-based I/O on
48 filehandles (and sockets, see L<AnyEvent::Socket> for an easy way to make
49 non-blocking resolves and connects).
50
51 In the following, when the documentation refers to of "bytes" then this
52 means characters. As sysread and syswrite are used for all I/O, their
53 treatment of characters applies to this module as well.
54
55 All callbacks will be invoked with the handle object as their first
56 argument.
57
58 =head1 METHODS
59
60 =over 4
61
62 =item B<new (%args)>
63
64 The constructor supports these arguments (all as key => value pairs).
65
66 =over 4
67
68 =item fh => $filehandle [MANDATORY]
69
70 The filehandle this L<AnyEvent::Handle> object will operate on.
71
72 NOTE: The filehandle will be set to non-blocking (using
73 AnyEvent::Util::fh_nonblocking).
74
75 =item on_error => $cb->($self) [MANDATORY]
76
77 This is the fatal error callback, that is called when a fatal error ocurs,
78 such as not being able to resolve the hostname, failure to connect or a
79 read error.
80
81 The object will not be in a usable state when this callback has been
82 called.
83
84 On callback entrance, the value of C<$!> contains the opertaing system
85 error (or C<ENOSPC> or C<EPIPE>).
86
87 =item on_eof => $cb->($self) [MANDATORY]
88
89 Set the callback to be called on EOF.
90
91 =item on_read => $cb->($self)
92
93 This sets the default read callback, which is called when data arrives
94 and no read request is in the queue. If the read callback is C<undef>
95 or has never been set, than AnyEvent::Handle will cease reading from the
96 filehandle.
97
98 To access (and remove data from) the read buffer, use the C<< ->rbuf >>
99 method or acces sthe C<$self->{rbuf}> member directly.
100
101 When an EOF condition is detected then AnyEvent::Handle will first try to
102 feed all the remaining data to the queued callbacks and C<on_read> before
103 calling the C<on_eof> callback. If no progress can be made, then a fatal
104 error will be raised (with C<$!> set to C<EPIPE>).
105
106 =item on_drain => $cb->()
107
108 This sets the callback that is called when the write buffer becomes empty
109 (or when the callback is set and the buffer is empty already).
110
111 To append to the write buffer, use the C<< ->push_write >> method.
112
113 =item rbuf_max => <bytes>
114
115 If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
116 when the read buffer ever (strictly) exceeds this size. This is useful to
117 avoid denial-of-service attacks.
118
119 For example, a server accepting connections from untrusted sources should
120 be configured to accept only so-and-so much data that it cannot act on
121 (for example, when expecting a line, an attacker could send an unlimited
122 amount of data without a callback ever being called as long as the line
123 isn't finished).
124
125 =item read_size => <bytes>
126
127 The default read block size (the amount of bytes this module will try to read
128 on each [loop iteration). Default: C<4096>.
129
130 =item low_water_mark => <bytes>
131
132 Sets the amount of bytes (default: C<0>) that make up an "empty" write
133 buffer: If the write reaches this size or gets even samller it is
134 considered empty.
135
136 =back
137
138 =cut
139
140 sub new {
141 my $class = shift;
142
143 my $self = bless { @_ }, $class;
144
145 $self->{fh} or Carp::croak "mandatory argument fh is missing";
146
147 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
148
149 $self->on_error ((delete $self->{on_error}) or Carp::croak "mandatory argument on_error is missing");
150 $self->on_eof ((delete $self->{on_eof} ) or Carp::croak "mandatory argument on_eof is missing");
151
152 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
153 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
154
155 $self
156 }
157
158 sub _shutdown {
159 my ($self) = @_;
160
161 delete $self->{rw};
162 delete $self->{ww};
163 delete $self->{fh};
164 }
165
166 sub error {
167 my ($self) = @_;
168
169 {
170 local $!;
171 $self->_shutdown;
172 }
173
174 $self->{on_error}($self);
175 }
176
177 =item $fh = $handle->fh
178
179 This method returns the filehandle of the L<AnyEvent::Handle> object.
180
181 =cut
182
183 sub fh { $_[0]->{fh} }
184
185 =item $handle->on_error ($cb)
186
187 Replace the current C<on_error> callback (see the C<on_error> constructor argument).
188
189 =cut
190
191 sub on_error {
192 $_[0]{on_error} = $_[1];
193 }
194
195 =item $handle->on_eof ($cb)
196
197 Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
198
199 =cut
200
201 #############################################################################
202
203 sub on_eof {
204 $_[0]{on_eof} = $_[1];
205 }
206
207 =item $handle->on_drain ($cb)
208
209 Sets the C<on_drain> callback or clears it (see the description of
210 C<on_drain> in the constructor).
211
212 =cut
213
214 sub on_drain {
215 my ($self, $cb) = @_;
216
217 $self->{on_drain} = $cb;
218
219 $cb->($self)
220 if $cb && $self->{low_water_mark} >= length $self->{wbuf};
221 }
222
223 =item $handle->push_write ($data)
224
225 Queues the given scalar to be written. You can push as much data as you
226 want (only limited by the available memory), as C<AnyEvent::Handle>
227 buffers it independently of the kernel.
228
229 =cut
230
231 sub push_write {
232 my ($self, $data) = @_;
233
234 $self->{wbuf} .= $data;
235
236 unless ($self->{ww}) {
237 Scalar::Util::weaken $self;
238 my $cb = sub {
239 my $len = syswrite $self->{fh}, $self->{wbuf};
240
241 if ($len > 0) {
242 substr $self->{wbuf}, 0, $len, "";
243
244
245 $self->{on_drain}($self)
246 if $self->{low_water_mark} >= length $self->{wbuf}
247 && $self->{on_drain};
248
249 delete $self->{ww} unless length $self->{wbuf};
250 } elsif ($! != EAGAIN && $! != EINTR) {
251 $self->error;
252 }
253 };
254
255 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb);
256
257 $cb->($self);
258 };
259 }
260
261 #############################################################################
262
263 sub _drain_rbuf {
264 my ($self) = @_;
265
266 return if exists $self->{in_drain};
267 local $self->{in_drain} = 1;
268
269 while (my $len = length $self->{rbuf}) {
270 no strict 'refs';
271 if (@{ $self->{queue} }) {
272 if ($self->{queue}[0]($self)) {
273 shift @{ $self->{queue} };
274 } elsif ($self->{eof}) {
275 # no progress can be made (not enough data and no data forthcoming)
276 $! = &Errno::EPIPE; return $self->error;
277 } else {
278 return;
279 }
280 } elsif ($self->{on_read}) {
281 $self->{on_read}($self);
282
283 if (
284 $self->{eof} # if no further data will arrive
285 && $len == length $self->{rbuf} # and no data has been consumed
286 && !@{ $self->{queue} } # and the queue is still empty
287 && $self->{on_read} # and we still want to read data
288 ) {
289 # then no progress can be made
290 $! = &Errno::EPIPE; return $self->error;
291 }
292 } else {
293 # read side becomes idle
294 delete $self->{rw};
295 return;
296 }
297 }
298
299 if ($self->{eof}) {
300 $self->_shutdown;
301 $self->{on_eof}($self);
302 }
303 }
304
305 =item $handle->on_read ($cb)
306
307 This replaces the currently set C<on_read> callback, or clears it (when
308 the new callback is C<undef>). See the description of C<on_read> in the
309 constructor.
310
311 =cut
312
313 sub on_read {
314 my ($self, $cb) = @_;
315
316 $self->{on_read} = $cb;
317
318 unless ($self->{rw} || $self->{eof}) {
319 Scalar::Util::weaken $self;
320
321 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
322 my $len = sysread $self->{fh}, $self->{rbuf}, $self->{read_size} || 8192, length $self->{rbuf};
323
324 if ($len > 0) {
325 if (exists $self->{rbuf_max}) {
326 if ($self->{rbuf_max} < length $self->{rbuf}) {
327 $! = &Errno::ENOSPC; return $self->error;
328 }
329 }
330
331 } elsif (defined $len) {
332 $self->{eof} = 1;
333 delete $self->{rw};
334
335 } elsif ($! != EAGAIN && $! != EINTR) {
336 return $self->error;
337 }
338
339 $self->_drain_rbuf;
340 });
341 }
342 }
343
344 =item $handle->rbuf
345
346 Returns the read buffer (as a modifiable lvalue).
347
348 You can access the read buffer directly as the C<< ->{rbuf} >> member, if
349 you want.
350
351 NOTE: The read buffer should only be used or modified if the C<on_read>,
352 C<push_read> or C<unshift_read> methods are used. The other read methods
353 automatically manage the read buffer.
354
355 =cut
356
357 sub rbuf : lvalue {
358 $_[0]{rbuf}
359 }
360
361 =item $handle->push_read ($cb)
362
363 =item $handle->unshift_read ($cb)
364
365 Append the given callback to the end of the queue (C<push_read>) or
366 prepend it (C<unshift_read>).
367
368 The callback is called each time some additional read data arrives.
369
370 It must check wether enough data is in the read buffer already.
371
372 If not enough data is available, it must return the empty list or a false
373 value, in which case it will be called repeatedly until enough data is
374 available (or an error condition is detected).
375
376 If enough data was available, then the callback must remove all data it is
377 interested in (which can be none at all) and return a true value. After returning
378 true, it will be removed from the queue.
379
380 =cut
381
382 sub push_read {
383 my ($self, $cb) = @_;
384
385 push @{ $self->{queue} }, $cb;
386 $self->_drain_rbuf;
387 }
388
389 sub unshift_read {
390 my ($self, $cb) = @_;
391
392 push @{ $self->{queue} }, $cb;
393 $self->_drain_rbuf;
394 }
395
396 =item $handle->push_read_chunk ($len, $cb->($self, $data))
397
398 =item $handle->unshift_read_chunk ($len, $cb->($self, $data))
399
400 Append the given callback to the end of the queue (C<push_read_chunk>) or
401 prepend it (C<unshift_read_chunk>).
402
403 The callback will be called only once C<$len> bytes have been read, and
404 these C<$len> bytes will be passed to the callback.
405
406 =cut
407
408 sub _read_chunk($$) {
409 my ($len, $cb) = @_;
410
411 sub {
412 $len <= length $_[0]{rbuf} or return;
413 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
414 1
415 }
416 }
417
418 sub push_read_chunk {
419 my ($self, $len, $cb) = @_;
420
421 $self->push_read (_read_chunk $len, $cb);
422 }
423
424
425 sub unshift_read_chunk {
426 my ($self, $len, $cb) = @_;
427
428 $self->unshift_read (_read_chunk $len, $cb);
429 }
430
431 =item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol))
432
433 =item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
434
435 Append the given callback to the end of the queue (C<push_read_line>) or
436 prepend it (C<unshift_read_line>).
437
438 The callback will be called only once a full line (including the end of
439 line marker, C<$eol>) has been read. This line (excluding the end of line
440 marker) will be passed to the callback as second argument (C<$line>), and
441 the end of line marker as the third argument (C<$eol>).
442
443 The end of line marker, C<$eol>, can be either a string, in which case it
444 will be interpreted as a fixed record end marker, or it can be a regex
445 object (e.g. created by C<qr>), in which case it is interpreted as a
446 regular expression.
447
448 The end of line marker argument C<$eol> is optional, if it is missing (NOT
449 undef), then C<qr|\015?\012|> is used (which is good for most internet
450 protocols).
451
452 Partial lines at the end of the stream will never be returned, as they are
453 not marked by the end of line marker.
454
455 =cut
456
457 sub _read_line($$) {
458 my $cb = pop;
459 my $eol = @_ ? shift : qr|(\015?\012)|;
460 my $pos;
461
462 $eol = qr|(\Q$eol\E)| unless ref $eol;
463 $eol = qr|^(.*?)($eol)|;
464
465 sub {
466 $_[0]{rbuf} =~ s/$eol// or return;
467
468 $cb->($1, $2);
469 1
470 }
471 }
472
473 sub push_read_line {
474 my $self = shift;
475
476 $self->push_read (&_read_line);
477 }
478
479 sub unshift_read_line {
480 my $self = shift;
481
482 $self->unshift_read (&_read_line);
483 }
484
485 =back
486
487 =head1 AUTHOR
488
489 Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
490
491 =cut
492
493 1; # End of AnyEvent::Handle