ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.34 by root, Mon May 26 03:27:52 2008 UTC vs.
Revision 1.61 by root, Fri Jun 6 10:23:50 2008 UTC

2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw(EAGAIN EINTR);
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = '0.04'; 19our $VERSION = 4.14;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
73The filehandle this L<AnyEvent::Handle> object will operate on. 73The filehandle this L<AnyEvent::Handle> object will operate on.
74 74
75NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
76AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
77 77
78=item on_eof => $cb->($self) 78=item on_eof => $cb->($handle)
79 79
80Set the callback to be called on EOF. 80Set the callback to be called when an end-of-file condition is detcted,
81i.e. in the case of a socket, when the other side has closed the
82connection cleanly.
81 83
82While not mandatory, it is highly recommended to set an eof callback, 84While not mandatory, it is highly recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still 85otherwise you might end up with a closed socket while you are still
84waiting for data. 86waiting for data.
85 87
86=item on_error => $cb->($self) 88=item on_error => $cb->($handle, $fatal)
87 89
88This is the fatal error callback, that is called when, well, a fatal error 90This is the error callback, which is called when, well, some error
89occurs, such as not being able to resolve the hostname, failure to connect 91occured, such as not being able to resolve the hostname, failure to
90or a read error. 92connect or a read error.
91 93
92The object will not be in a usable state when this callback has been 94Some errors are fatal (which is indicated by C<$fatal> being true). On
93called. 95fatal errors the handle object will be shut down and will not be
96usable. Non-fatal errors can be retried by simply returning, but it is
97recommended to simply ignore this parameter and instead abondon the handle
98object when this callback is invoked.
94 99
95On callback entrance, the value of C<$!> contains the operating system 100On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>). 101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 102
98While not mandatory, it is I<highly> recommended to set this callback, as 103While not mandatory, it is I<highly> recommended to set this callback, as
99you will not be notified of errors otherwise. The default simply calls 104you will not be notified of errors otherwise. The default simply calls
100die. 105C<croak>.
101 106
102=item on_read => $cb->($self) 107=item on_read => $cb->($handle)
103 108
104This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
105and no read request is in the queue. 110and no read request is in the queue (unlike read queue callbacks, this
111callback will only be called when at least one octet of data is in the
112read buffer).
106 113
107To access (and remove data from) the read buffer, use the C<< ->rbuf >> 114To access (and remove data from) the read buffer, use the C<< ->rbuf >>
108method or access the C<$self->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
109 116
110When an EOF condition is detected then AnyEvent::Handle will first try to 117When an EOF condition is detected then AnyEvent::Handle will first try to
111feed all the remaining data to the queued callbacks and C<on_read> before 118feed all the remaining data to the queued callbacks and C<on_read> before
112calling the C<on_eof> callback. If no progress can be made, then a fatal 119calling the C<on_eof> callback. If no progress can be made, then a fatal
113error will be raised (with C<$!> set to C<EPIPE>). 120error will be raised (with C<$!> set to C<EPIPE>).
114 121
115=item on_drain => $cb->() 122=item on_drain => $cb->($handle)
116 123
117This sets the callback that is called when the write buffer becomes empty 124This sets the callback that is called when the write buffer becomes empty
118(or when the callback is set and the buffer is empty already). 125(or when the callback is set and the buffer is empty already).
119 126
120To append to the write buffer, use the C<< ->push_write >> method. 127To append to the write buffer, use the C<< ->push_write >> method.
128
129=item timeout => $fractional_seconds
130
131If non-zero, then this enables an "inactivity" timeout: whenever this many
132seconds pass without a successful read or write on the underlying file
133handle, the C<on_timeout> callback will be invoked (and if that one is
134missing, an C<ETIMEDOUT> error will be raised).
135
136Note that timeout processing is also active when you currently do not have
137any outstanding read or write requests: If you plan to keep the connection
138idle then you should disable the timout temporarily or ignore the timeout
139in the C<on_timeout> callback.
140
141Zero (the default) disables this timeout.
142
143=item on_timeout => $cb->($handle)
144
145Called whenever the inactivity timeout passes. If you return from this
146callback, then the timeout will be reset as if some activity had happened,
147so this condition is not fatal in any way.
121 148
122=item rbuf_max => <bytes> 149=item rbuf_max => <bytes>
123 150
124If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 151If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
125when the read buffer ever (strictly) exceeds this size. This is useful to 152when the read buffer ever (strictly) exceeds this size. This is useful to
132isn't finished). 159isn't finished).
133 160
134=item read_size => <bytes> 161=item read_size => <bytes>
135 162
136The default read block size (the amount of bytes this module will try to read 163The default read block size (the amount of bytes this module will try to read
137on each [loop iteration). Default: C<4096>. 164during each (loop iteration). Default: C<8192>.
138 165
139=item low_water_mark => <bytes> 166=item low_water_mark => <bytes>
140 167
141Sets the amount of bytes (default: C<0>) that make up an "empty" write 168Sets the amount of bytes (default: C<0>) that make up an "empty" write
142buffer: If the write reaches this size or gets even samller it is 169buffer: If the write reaches this size or gets even samller it is
165 192
166Use the given Net::SSLeay::CTX object to create the new TLS connection 193Use the given Net::SSLeay::CTX object to create the new TLS connection
167(unless a connection object was specified directly). If this parameter is 194(unless a connection object was specified directly). If this parameter is
168missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 195missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
169 196
197=item json => JSON or JSON::XS object
198
199This is the json coder object used by the C<json> read and write types.
200
201If you don't supply it, then AnyEvent::Handle will create and use a
202suitable one, which will write and expect UTF-8 encoded JSON texts.
203
204Note that you are responsible to depend on the JSON module if you want to
205use this functionality, as AnyEvent does not have a dependency itself.
206
207=item filter_r => $cb
208
209=item filter_w => $cb
210
211These exist, but are undocumented at this time.
212
170=back 213=back
171 214
172=cut 215=cut
173 216
174sub new { 217sub new {
183 if ($self->{tls}) { 226 if ($self->{tls}) {
184 require Net::SSLeay; 227 require Net::SSLeay;
185 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 228 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
186 } 229 }
187 230
188 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 231 $self->{_activity} = AnyEvent->now;
189 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 232 $self->_timeout;
233
190 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 234 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
191 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
192
193 $self->start_read;
194 235
195 $self 236 $self
196} 237}
197 238
198sub _shutdown { 239sub _shutdown {
199 my ($self) = @_; 240 my ($self) = @_;
200 241
242 delete $self->{_tw};
201 delete $self->{rw}; 243 delete $self->{_rw};
202 delete $self->{ww}; 244 delete $self->{_ww};
203 delete $self->{fh}; 245 delete $self->{fh};
204}
205 246
247 $self->stoptls;
248}
249
206sub error { 250sub _error {
207 my ($self) = @_; 251 my ($self, $errno, $fatal) = @_;
208 252
209 {
210 local $!;
211 $self->_shutdown; 253 $self->_shutdown
212 } 254 if $fatal;
255
256 $! = $errno;
213 257
214 if ($self->{on_error}) { 258 if ($self->{on_error}) {
215 $self->{on_error}($self); 259 $self->{on_error}($self, $fatal);
216 } else { 260 } else {
217 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 261 Carp::croak "AnyEvent::Handle uncaught error: $!";
218 } 262 }
219} 263}
220 264
221=item $fh = $handle->fh 265=item $fh = $handle->fh
222 266
223This method returns the file handle of the L<AnyEvent::Handle> object. 267This method returns the file handle of the L<AnyEvent::Handle> object.
224 268
225=cut 269=cut
226 270
227sub fh { $_[0]->{fh} } 271sub fh { $_[0]{fh} }
228 272
229=item $handle->on_error ($cb) 273=item $handle->on_error ($cb)
230 274
231Replace the current C<on_error> callback (see the C<on_error> constructor argument). 275Replace the current C<on_error> callback (see the C<on_error> constructor argument).
232 276
242 286
243=cut 287=cut
244 288
245sub on_eof { 289sub on_eof {
246 $_[0]{on_eof} = $_[1]; 290 $_[0]{on_eof} = $_[1];
291}
292
293=item $handle->on_timeout ($cb)
294
295Replace the current C<on_timeout> callback, or disables the callback
296(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
297argument.
298
299=cut
300
301sub on_timeout {
302 $_[0]{on_timeout} = $_[1];
303}
304
305#############################################################################
306
307=item $handle->timeout ($seconds)
308
309Configures (or disables) the inactivity timeout.
310
311=cut
312
313sub timeout {
314 my ($self, $timeout) = @_;
315
316 $self->{timeout} = $timeout;
317 $self->_timeout;
318}
319
320# reset the timeout watcher, as neccessary
321# also check for time-outs
322sub _timeout {
323 my ($self) = @_;
324
325 if ($self->{timeout}) {
326 my $NOW = AnyEvent->now;
327
328 # when would the timeout trigger?
329 my $after = $self->{_activity} + $self->{timeout} - $NOW;
330
331 # now or in the past already?
332 if ($after <= 0) {
333 $self->{_activity} = $NOW;
334
335 if ($self->{on_timeout}) {
336 $self->{on_timeout}($self);
337 } else {
338 $self->_error (&Errno::ETIMEDOUT);
339 }
340
341 # callback could have changed timeout value, optimise
342 return unless $self->{timeout};
343
344 # calculate new after
345 $after = $self->{timeout};
346 }
347
348 Scalar::Util::weaken $self;
349 return unless $self; # ->error could have destroyed $self
350
351 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
352 delete $self->{_tw};
353 $self->_timeout;
354 });
355 } else {
356 delete $self->{_tw};
357 }
247} 358}
248 359
249############################################################################# 360#############################################################################
250 361
251=back 362=back
288=cut 399=cut
289 400
290sub _drain_wbuf { 401sub _drain_wbuf {
291 my ($self) = @_; 402 my ($self) = @_;
292 403
293 if (!$self->{ww} && length $self->{wbuf}) { 404 if (!$self->{_ww} && length $self->{wbuf}) {
405
294 Scalar::Util::weaken $self; 406 Scalar::Util::weaken $self;
407
295 my $cb = sub { 408 my $cb = sub {
296 my $len = syswrite $self->{fh}, $self->{wbuf}; 409 my $len = syswrite $self->{fh}, $self->{wbuf};
297 410
298 if ($len >= 0) { 411 if ($len >= 0) {
299 substr $self->{wbuf}, 0, $len, ""; 412 substr $self->{wbuf}, 0, $len, "";
413
414 $self->{_activity} = AnyEvent->now;
300 415
301 $self->{on_drain}($self) 416 $self->{on_drain}($self)
302 if $self->{low_water_mark} >= length $self->{wbuf} 417 if $self->{low_water_mark} >= length $self->{wbuf}
303 && $self->{on_drain}; 418 && $self->{on_drain};
304 419
305 delete $self->{ww} unless length $self->{wbuf}; 420 delete $self->{_ww} unless length $self->{wbuf};
306 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 421 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
307 $self->error; 422 $self->_error ($!, 1);
308 } 423 }
309 }; 424 };
310 425
426 # try to write data immediately
427 $cb->();
428
429 # if still data left in wbuf, we need to poll
311 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 430 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
312 431 if length $self->{wbuf};
313 $cb->($self);
314 }; 432 };
315} 433}
316 434
317our %WH; 435our %WH;
318 436
329 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 447 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
330 ->($self, @_); 448 ->($self, @_);
331 } 449 }
332 450
333 if ($self->{filter_w}) { 451 if ($self->{filter_w}) {
334 $self->{filter_w}->($self, \$_[0]); 452 $self->{filter_w}($self, \$_[0]);
335 } else { 453 } else {
336 $self->{wbuf} .= $_[0]; 454 $self->{wbuf} .= $_[0];
337 $self->_drain_wbuf; 455 $self->_drain_wbuf;
338 } 456 }
339} 457}
340 458
341=item $handle->push_write (type => @args) 459=item $handle->push_write (type => @args)
342 460
343=item $handle->unshift_write (type => @args)
344
345Instead of formatting your data yourself, you can also let this module do 461Instead of formatting your data yourself, you can also let this module do
346the job by specifying a type and type-specific arguments. 462the job by specifying a type and type-specific arguments.
347 463
348Predefined types are (if you have ideas for additional types, feel free to 464Predefined types are (if you have ideas for additional types, feel free to
349drop by and tell us): 465drop by and tell us):
353=item netstring => $string 469=item netstring => $string
354 470
355Formats the given value as netstring 471Formats the given value as netstring
356(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
357 473
358=back
359
360=cut 474=cut
361 475
362register_write_type netstring => sub { 476register_write_type netstring => sub {
363 my ($self, $string) = @_; 477 my ($self, $string) = @_;
364 478
365 sprintf "%d:%s,", (length $string), $string 479 sprintf "%d:%s,", (length $string), $string
366}; 480};
367 481
482=item packstring => $format, $data
483
484An octet string prefixed with an encoded length. The encoding C<$format>
485uses the same format as a Perl C<pack> format, but must specify a single
486integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
487optional C<!>, C<< < >> or C<< > >> modifier).
488
489=cut
490
491register_write_type packstring => sub {
492 my ($self, $format, $string) = @_;
493
494 pack "$format/a", $string
495};
496
497=item json => $array_or_hashref
498
499Encodes the given hash or array reference into a JSON object. Unless you
500provide your own JSON object, this means it will be encoded to JSON text
501in UTF-8.
502
503JSON objects (and arrays) are self-delimiting, so you can write JSON at
504one end of a handle and read them at the other end without using any
505additional framing.
506
507The generated JSON text is guaranteed not to contain any newlines: While
508this module doesn't need delimiters after or between JSON texts to be
509able to read them, many other languages depend on that.
510
511A simple RPC protocol that interoperates easily with others is to send
512JSON arrays (or objects, although arrays are usually the better choice as
513they mimic how function argument passing works) and a newline after each
514JSON text:
515
516 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
517 $handle->push_write ("\012");
518
519An AnyEvent::Handle receiver would simply use the C<json> read type and
520rely on the fact that the newline will be skipped as leading whitespace:
521
522 $handle->push_read (json => sub { my $array = $_[1]; ... });
523
524Other languages could read single lines terminated by a newline and pass
525this line into their JSON decoder of choice.
526
527=cut
528
529register_write_type json => sub {
530 my ($self, $ref) = @_;
531
532 require JSON;
533
534 $self->{json} ? $self->{json}->encode ($ref)
535 : JSON::encode_json ($ref)
536};
537
538=back
539
368=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args) 540=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
369 541
370This function (not method) lets you add your own types to C<push_write>. 542This function (not method) lets you add your own types to C<push_write>.
371Whenever the given C<type> is used, C<push_write> will invoke the code 543Whenever the given C<type> is used, C<push_write> will invoke the code
372reference with the handle object and the remaining arguments. 544reference with the handle object and the remaining arguments.
373 545
397enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 569enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
398or not. 570or not.
399 571
400In the more complex case, you want to queue multiple callbacks. In this 572In the more complex case, you want to queue multiple callbacks. In this
401case, AnyEvent::Handle will call the first queued callback each time new 573case, AnyEvent::Handle will call the first queued callback each time new
402data arrives and removes it when it has done its job (see C<push_read>, 574data arrives (also the first time it is queued) and removes it when it has
403below). 575done its job (see C<push_read>, below).
404 576
405This way you can, for example, push three line-reads, followed by reading 577This way you can, for example, push three line-reads, followed by reading
406a chunk of data, and AnyEvent::Handle will execute them in order. 578a chunk of data, and AnyEvent::Handle will execute them in order.
407 579
408Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 580Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
409the specified number of bytes which give an XML datagram. 581the specified number of bytes which give an XML datagram.
410 582
411 # in the default state, expect some header bytes 583 # in the default state, expect some header bytes
412 $handle->on_read (sub { 584 $handle->on_read (sub {
413 # some data is here, now queue the length-header-read (4 octets) 585 # some data is here, now queue the length-header-read (4 octets)
414 shift->unshift_read_chunk (4, sub { 586 shift->unshift_read (chunk => 4, sub {
415 # header arrived, decode 587 # header arrived, decode
416 my $len = unpack "N", $_[1]; 588 my $len = unpack "N", $_[1];
417 589
418 # now read the payload 590 # now read the payload
419 shift->unshift_read_chunk ($len, sub { 591 shift->unshift_read (chunk => $len, sub {
420 my $xml = $_[1]; 592 my $xml = $_[1];
421 # handle xml 593 # handle xml
422 }); 594 });
423 }); 595 });
424 }); 596 });
431 603
432 # request one 604 # request one
433 $handle->push_write ("request 1\015\012"); 605 $handle->push_write ("request 1\015\012");
434 606
435 # we expect "ERROR" or "OK" as response, so push a line read 607 # we expect "ERROR" or "OK" as response, so push a line read
436 $handle->push_read_line (sub { 608 $handle->push_read (line => sub {
437 # if we got an "OK", we have to _prepend_ another line, 609 # if we got an "OK", we have to _prepend_ another line,
438 # so it will be read before the second request reads its 64 bytes 610 # so it will be read before the second request reads its 64 bytes
439 # which are already in the queue when this callback is called 611 # which are already in the queue when this callback is called
440 # we don't do this in case we got an error 612 # we don't do this in case we got an error
441 if ($_[1] eq "OK") { 613 if ($_[1] eq "OK") {
442 $_[0]->unshift_read_line (sub { 614 $_[0]->unshift_read (line => sub {
443 my $response = $_[1]; 615 my $response = $_[1];
444 ... 616 ...
445 }); 617 });
446 } 618 }
447 }); 619 });
448 620
449 # request two 621 # request two
450 $handle->push_write ("request 2\015\012"); 622 $handle->push_write ("request 2\015\012");
451 623
452 # simply read 64 bytes, always 624 # simply read 64 bytes, always
453 $handle->push_read_chunk (64, sub { 625 $handle->push_read (chunk => 64, sub {
454 my $response = $_[1]; 626 my $response = $_[1];
455 ... 627 ...
456 }); 628 });
457 629
458=over 4 630=over 4
459 631
460=cut 632=cut
461 633
462sub _drain_rbuf { 634sub _drain_rbuf {
463 my ($self) = @_; 635 my ($self) = @_;
636
637 local $self->{_in_drain} = 1;
464 638
465 if ( 639 if (
466 defined $self->{rbuf_max} 640 defined $self->{rbuf_max}
467 && $self->{rbuf_max} < length $self->{rbuf} 641 && $self->{rbuf_max} < length $self->{rbuf}
468 ) { 642 ) {
469 $! = &Errno::ENOSPC; return $self->error; 643 return $self->_error (&Errno::ENOSPC, 1);
470 } 644 }
471 645
472 return if $self->{in_drain}; 646 while () {
473 local $self->{in_drain} = 1;
474
475 while (my $len = length $self->{rbuf}) {
476 no strict 'refs'; 647 no strict 'refs';
648
649 my $len = length $self->{rbuf};
650
477 if (my $cb = shift @{ $self->{queue} }) { 651 if (my $cb = shift @{ $self->{_queue} }) {
478 unless ($cb->($self)) { 652 unless ($cb->($self)) {
479 if ($self->{eof}) { 653 if ($self->{_eof}) {
480 # no progress can be made (not enough data and no data forthcoming) 654 # no progress can be made (not enough data and no data forthcoming)
481 $! = &Errno::EPIPE; return $self->error; 655 $self->_error (&Errno::EPIPE, 1), last;
482 } 656 }
483 657
484 unshift @{ $self->{queue} }, $cb; 658 unshift @{ $self->{_queue} }, $cb;
485 return; 659 last;
486 } 660 }
487 } elsif ($self->{on_read}) { 661 } elsif ($self->{on_read}) {
662 last unless $len;
663
488 $self->{on_read}($self); 664 $self->{on_read}($self);
489 665
490 if ( 666 if (
491 $self->{eof} # if no further data will arrive
492 && $len == length $self->{rbuf} # and no data has been consumed 667 $len == length $self->{rbuf} # if no data has been consumed
493 && !@{ $self->{queue} } # and the queue is still empty 668 && !@{ $self->{_queue} } # and the queue is still empty
494 && $self->{on_read} # and we still want to read data 669 && $self->{on_read} # but we still have on_read
495 ) { 670 ) {
671 # no further data will arrive
496 # then no progress can be made 672 # so no progress can be made
497 $! = &Errno::EPIPE; return $self->error; 673 $self->_error (&Errno::EPIPE, 1), last
674 if $self->{_eof};
675
676 last; # more data might arrive
498 } 677 }
499 } else { 678 } else {
500 # read side becomes idle 679 # read side becomes idle
501 delete $self->{rw}; 680 delete $self->{_rw};
502 return; 681 last;
503 } 682 }
504 } 683 }
505 684
506 if ($self->{eof}) {
507 $self->_shutdown;
508 $self->{on_eof}($self) 685 $self->{on_eof}($self)
509 if $self->{on_eof}; 686 if $self->{_eof} && $self->{on_eof};
687
688 # may need to restart read watcher
689 unless ($self->{_rw}) {
690 $self->start_read
691 if $self->{on_read} || @{ $self->{_queue} };
510 } 692 }
511} 693}
512 694
513=item $handle->on_read ($cb) 695=item $handle->on_read ($cb)
514 696
520 702
521sub on_read { 703sub on_read {
522 my ($self, $cb) = @_; 704 my ($self, $cb) = @_;
523 705
524 $self->{on_read} = $cb; 706 $self->{on_read} = $cb;
707 $self->_drain_rbuf if $cb && !$self->{_in_drain};
525} 708}
526 709
527=item $handle->rbuf 710=item $handle->rbuf
528 711
529Returns the read buffer (as a modifiable lvalue). 712Returns the read buffer (as a modifiable lvalue).
577 760
578 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 761 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
579 ->($self, $cb, @_); 762 ->($self, $cb, @_);
580 } 763 }
581 764
582 push @{ $self->{queue} }, $cb; 765 push @{ $self->{_queue} }, $cb;
583 $self->_drain_rbuf; 766 $self->_drain_rbuf unless $self->{_in_drain};
584} 767}
585 768
586sub unshift_read { 769sub unshift_read {
587 my $self = shift; 770 my $self = shift;
588 my $cb = pop; 771 my $cb = pop;
593 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 776 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
594 ->($self, $cb, @_); 777 ->($self, $cb, @_);
595 } 778 }
596 779
597 780
598 unshift @{ $self->{queue} }, $cb; 781 unshift @{ $self->{_queue} }, $cb;
599 $self->_drain_rbuf; 782 $self->_drain_rbuf unless $self->{_in_drain};
600} 783}
601 784
602=item $handle->push_read (type => @args, $cb) 785=item $handle->push_read (type => @args, $cb)
603 786
604=item $handle->unshift_read (type => @args, $cb) 787=item $handle->unshift_read (type => @args, $cb)
610Predefined types are (if you have ideas for additional types, feel free to 793Predefined types are (if you have ideas for additional types, feel free to
611drop by and tell us): 794drop by and tell us):
612 795
613=over 4 796=over 4
614 797
615=item chunk => $octets, $cb->($self, $data) 798=item chunk => $octets, $cb->($handle, $data)
616 799
617Invoke the callback only once C<$octets> bytes have been read. Pass the 800Invoke the callback only once C<$octets> bytes have been read. Pass the
618data read to the callback. The callback will never be called with less 801data read to the callback. The callback will never be called with less
619data. 802data.
620 803
643 826
644sub unshift_read_chunk { 827sub unshift_read_chunk {
645 $_[0]->unshift_read (chunk => $_[1], $_[2]); 828 $_[0]->unshift_read (chunk => $_[1], $_[2]);
646} 829}
647 830
648=item line => [$eol, ]$cb->($self, $line, $eol) 831=item line => [$eol, ]$cb->($handle, $line, $eol)
649 832
650The callback will be called only once a full line (including the end of 833The callback will be called only once a full line (including the end of
651line marker, C<$eol>) has been read. This line (excluding the end of line 834line marker, C<$eol>) has been read. This line (excluding the end of line
652marker) will be passed to the callback as second argument (C<$line>), and 835marker) will be passed to the callback as second argument (C<$line>), and
653the end of line marker as the third argument (C<$eol>). 836the end of line marker as the third argument (C<$eol>).
690sub unshift_read_line { 873sub unshift_read_line {
691 my $self = shift; 874 my $self = shift;
692 $self->unshift_read (line => @_); 875 $self->unshift_read (line => @_);
693} 876}
694 877
878=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
879
880Makes a regex match against the regex object C<$accept> and returns
881everything up to and including the match.
882
883Example: read a single line terminated by '\n'.
884
885 $handle->push_read (regex => qr<\n>, sub { ... });
886
887If C<$reject> is given and not undef, then it determines when the data is
888to be rejected: it is matched against the data when the C<$accept> regex
889does not match and generates an C<EBADMSG> error when it matches. This is
890useful to quickly reject wrong data (to avoid waiting for a timeout or a
891receive buffer overflow).
892
893Example: expect a single decimal number followed by whitespace, reject
894anything else (not the use of an anchor).
895
896 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
897
898If C<$skip> is given and not C<undef>, then it will be matched against
899the receive buffer when neither C<$accept> nor C<$reject> match,
900and everything preceding and including the match will be accepted
901unconditionally. This is useful to skip large amounts of data that you
902know cannot be matched, so that the C<$accept> or C<$reject> regex do not
903have to start matching from the beginning. This is purely an optimisation
904and is usually worth only when you expect more than a few kilobytes.
905
906Example: expect a http header, which ends at C<\015\012\015\012>. Since we
907expect the header to be very large (it isn't in practise, but...), we use
908a skip regex to skip initial portions. The skip regex is tricky in that
909it only accepts something not ending in either \015 or \012, as these are
910required for the accept regex.
911
912 $handle->push_read (regex =>
913 qr<\015\012\015\012>,
914 undef, # no reject
915 qr<^.*[^\015\012]>,
916 sub { ... });
917
918=cut
919
920register_read_type regex => sub {
921 my ($self, $cb, $accept, $reject, $skip) = @_;
922
923 my $data;
924 my $rbuf = \$self->{rbuf};
925
926 sub {
927 # accept
928 if ($$rbuf =~ $accept) {
929 $data .= substr $$rbuf, 0, $+[0], "";
930 $cb->($self, $data);
931 return 1;
932 }
933
934 # reject
935 if ($reject && $$rbuf =~ $reject) {
936 $self->_error (&Errno::EBADMSG);
937 }
938
939 # skip
940 if ($skip && $$rbuf =~ $skip) {
941 $data .= substr $$rbuf, 0, $+[0], "";
942 }
943
944 ()
945 }
946};
947
695=item netstring => $cb->($string) 948=item netstring => $cb->($handle, $string)
696 949
697A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement). 950A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
698 951
699Throws an error with C<$!> set to EBADMSG on format violations. 952Throws an error with C<$!> set to EBADMSG on format violations.
700 953
704 my ($self, $cb) = @_; 957 my ($self, $cb) = @_;
705 958
706 sub { 959 sub {
707 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 960 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
708 if ($_[0]{rbuf} =~ /[^0-9]/) { 961 if ($_[0]{rbuf} =~ /[^0-9]/) {
709 $! = &Errno::EBADMSG; 962 $self->_error (&Errno::EBADMSG);
710 $self->error;
711 } 963 }
712 return; 964 return;
713 } 965 }
714 966
715 my $len = $1; 967 my $len = $1;
718 my $string = $_[1]; 970 my $string = $_[1];
719 $_[0]->unshift_read (chunk => 1, sub { 971 $_[0]->unshift_read (chunk => 1, sub {
720 if ($_[1] eq ",") { 972 if ($_[1] eq ",") {
721 $cb->($_[0], $string); 973 $cb->($_[0], $string);
722 } else { 974 } else {
723 $! = &Errno::EBADMSG; 975 $self->_error (&Errno::EBADMSG);
724 $self->error;
725 } 976 }
726 }); 977 });
727 }); 978 });
728 979
729 1 980 1
730 } 981 }
731}; 982};
732 983
984=item packstring => $format, $cb->($handle, $string)
985
986An octet string prefixed with an encoded length. The encoding C<$format>
987uses the same format as a Perl C<pack> format, but must specify a single
988integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
989optional C<!>, C<< < >> or C<< > >> modifier).
990
991DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
992
993Example: read a block of data prefixed by its length in BER-encoded
994format (very efficient).
995
996 $handle->push_read (packstring => "w", sub {
997 my ($handle, $data) = @_;
998 });
999
1000=cut
1001
1002register_read_type packstring => sub {
1003 my ($self, $cb, $format) = @_;
1004
1005 sub {
1006 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1007 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1008 or return;
1009
1010 # remove prefix
1011 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1012
1013 # read rest
1014 $_[0]->unshift_read (chunk => $len, $cb);
1015
1016 1
1017 }
1018};
1019
1020=item json => $cb->($handle, $hash_or_arrayref)
1021
1022Reads a JSON object or array, decodes it and passes it to the callback.
1023
1024If a C<json> object was passed to the constructor, then that will be used
1025for the final decode, otherwise it will create a JSON coder expecting UTF-8.
1026
1027This read type uses the incremental parser available with JSON version
10282.09 (and JSON::XS version 2.2) and above. You have to provide a
1029dependency on your own: this module will load the JSON module, but
1030AnyEvent does not depend on it itself.
1031
1032Since JSON texts are fully self-delimiting, the C<json> read and write
1033types are an ideal simple RPC protocol: just exchange JSON datagrams. See
1034the C<json> write type description, above, for an actual example.
1035
1036=cut
1037
1038register_read_type json => sub {
1039 my ($self, $cb, $accept, $reject, $skip) = @_;
1040
1041 require JSON;
1042
1043 my $data;
1044 my $rbuf = \$self->{rbuf};
1045
1046 my $json = $self->{json} ||= JSON->new->utf8;
1047
1048 sub {
1049 my $ref = $json->incr_parse ($self->{rbuf});
1050
1051 if ($ref) {
1052 $self->{rbuf} = $json->incr_text;
1053 $json->incr_text = "";
1054 $cb->($self, $ref);
1055
1056 1
1057 } else {
1058 $self->{rbuf} = "";
1059 ()
1060 }
1061 }
1062};
1063
733=back 1064=back
734 1065
735=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args) 1066=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
736 1067
737This function (not method) lets you add your own types to C<push_read>. 1068This function (not method) lets you add your own types to C<push_read>.
738 1069
739Whenever the given C<type> is used, C<push_read> will invoke the code 1070Whenever the given C<type> is used, C<push_read> will invoke the code
740reference with the handle object, the callback and the remaining 1071reference with the handle object, the callback and the remaining
742 1073
743The code reference is supposed to return a callback (usually a closure) 1074The code reference is supposed to return a callback (usually a closure)
744that works as a plain read callback (see C<< ->push_read ($cb) >>). 1075that works as a plain read callback (see C<< ->push_read ($cb) >>).
745 1076
746It should invoke the passed callback when it is done reading (remember to 1077It should invoke the passed callback when it is done reading (remember to
747pass C<$self> as first argument as all other callbacks do that). 1078pass C<$handle> as first argument as all other callbacks do that).
748 1079
749Note that this is a function, and all types registered this way will be 1080Note that this is a function, and all types registered this way will be
750global, so try to use unique names. 1081global, so try to use unique names.
751 1082
752For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1083For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
755=item $handle->stop_read 1086=item $handle->stop_read
756 1087
757=item $handle->start_read 1088=item $handle->start_read
758 1089
759In rare cases you actually do not want to read anything from the 1090In rare cases you actually do not want to read anything from the
760socket. In this case you can call C<stop_read>. Neither C<on_read> no 1091socket. In this case you can call C<stop_read>. Neither C<on_read> nor
761any queued callbacks will be executed then. To start reading again, call 1092any queued callbacks will be executed then. To start reading again, call
762C<start_read>. 1093C<start_read>.
763 1094
1095Note that AnyEvent::Handle will automatically C<start_read> for you when
1096you change the C<on_read> callback or push/unshift a read callback, and it
1097will automatically C<stop_read> for you when neither C<on_read> is set nor
1098there are any read requests in the queue.
1099
764=cut 1100=cut
765 1101
766sub stop_read { 1102sub stop_read {
767 my ($self) = @_; 1103 my ($self) = @_;
768 1104
769 delete $self->{rw}; 1105 delete $self->{_rw};
770} 1106}
771 1107
772sub start_read { 1108sub start_read {
773 my ($self) = @_; 1109 my ($self) = @_;
774 1110
775 unless ($self->{rw} || $self->{eof}) { 1111 unless ($self->{_rw} || $self->{_eof}) {
776 Scalar::Util::weaken $self; 1112 Scalar::Util::weaken $self;
777 1113
778 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1114 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
779 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1115 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
780 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1116 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
781 1117
782 if ($len > 0) { 1118 if ($len > 0) {
1119 $self->{_activity} = AnyEvent->now;
1120
783 $self->{filter_r} 1121 $self->{filter_r}
784 ? $self->{filter_r}->($self, $rbuf) 1122 ? $self->{filter_r}($self, $rbuf)
785 : $self->_drain_rbuf; 1123 : $self->{_in_drain} || $self->_drain_rbuf;
786 1124
787 } elsif (defined $len) { 1125 } elsif (defined $len) {
788 delete $self->{rw}; 1126 delete $self->{_rw};
789 $self->{eof} = 1; 1127 $self->{_eof} = 1;
790 $self->_drain_rbuf; 1128 $self->_drain_rbuf unless $self->{_in_drain};
791 1129
792 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1130 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
793 return $self->error; 1131 return $self->_error ($!, 1);
794 } 1132 }
795 }); 1133 });
796 } 1134 }
797} 1135}
798 1136
799sub _dotls { 1137sub _dotls {
800 my ($self) = @_; 1138 my ($self) = @_;
801 1139
1140 my $buf;
1141
802 if (length $self->{tls_wbuf}) { 1142 if (length $self->{_tls_wbuf}) {
803 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) { 1143 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
804 substr $self->{tls_wbuf}, 0, $len, ""; 1144 substr $self->{_tls_wbuf}, 0, $len, "";
805 } 1145 }
806 } 1146 }
807 1147
808 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) { 1148 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
809 $self->{wbuf} .= $buf; 1149 $self->{wbuf} .= $buf;
810 $self->_drain_wbuf; 1150 $self->_drain_wbuf;
811 } 1151 }
812 1152
813 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1153 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1154 if (length $buf) {
814 $self->{rbuf} .= $buf; 1155 $self->{rbuf} .= $buf;
815 $self->_drain_rbuf; 1156 $self->_drain_rbuf unless $self->{_in_drain};
1157 } else {
1158 # let's treat SSL-eof as we treat normal EOF
1159 $self->{_eof} = 1;
1160 $self->_shutdown;
1161 return;
1162 }
816 } 1163 }
817 1164
818 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1165 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
819 1166
820 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1167 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
821 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1168 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
822 $self->error; 1169 return $self->_error ($!, 1);
823 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1170 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
824 $! = &Errno::EIO; 1171 return $self->_error (&Errno::EIO, 1);
825 $self->error;
826 } 1172 }
827 1173
828 # all others are fine for our purposes 1174 # all others are fine for our purposes
829 } 1175 }
830} 1176}
839C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1185C<"connect">, C<"accept"> or an existing Net::SSLeay object).
840 1186
841The second argument is the optional C<Net::SSLeay::CTX> object that is 1187The second argument is the optional C<Net::SSLeay::CTX> object that is
842used when AnyEvent::Handle has to create its own TLS connection object. 1188used when AnyEvent::Handle has to create its own TLS connection object.
843 1189
844=cut 1190The TLS connection object will end up in C<< $handle->{tls} >> after this
1191call and can be used or changed to your liking. Note that the handshake
1192might have already started when this function returns.
845 1193
846# TODO: maybe document... 1194=cut
1195
847sub starttls { 1196sub starttls {
848 my ($self, $ssl, $ctx) = @_; 1197 my ($self, $ssl, $ctx) = @_;
849 1198
850 $self->stoptls; 1199 $self->stoptls;
851 1200
866 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1215 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
867 Net::SSLeay::CTX_set_mode ($self->{tls}, 1216 Net::SSLeay::CTX_set_mode ($self->{tls},
868 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1217 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
869 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1218 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
870 1219
871 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1220 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
872 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1221 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
873 1222
874 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 1223 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
875 1224
876 $self->{filter_w} = sub { 1225 $self->{filter_w} = sub {
877 $_[0]{tls_wbuf} .= ${$_[1]}; 1226 $_[0]{_tls_wbuf} .= ${$_[1]};
878 &_dotls; 1227 &_dotls;
879 }; 1228 };
880 $self->{filter_r} = sub { 1229 $self->{filter_r} = sub {
881 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]}); 1230 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
882 &_dotls; 1231 &_dotls;
883 }; 1232 };
884} 1233}
885 1234
886=item $handle->stoptls 1235=item $handle->stoptls
892 1241
893sub stoptls { 1242sub stoptls {
894 my ($self) = @_; 1243 my ($self) = @_;
895 1244
896 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1245 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
1246
897 delete $self->{tls_rbio}; 1247 delete $self->{_rbio};
898 delete $self->{tls_wbio}; 1248 delete $self->{_wbio};
899 delete $self->{tls_wbuf}; 1249 delete $self->{_tls_wbuf};
900 delete $self->{filter_r}; 1250 delete $self->{filter_r};
901 delete $self->{filter_w}; 1251 delete $self->{filter_w};
902} 1252}
903 1253
904sub DESTROY { 1254sub DESTROY {
942 } 1292 }
943} 1293}
944 1294
945=back 1295=back
946 1296
1297=head1 SUBCLASSING AnyEvent::Handle
1298
1299In many cases, you might want to subclass AnyEvent::Handle.
1300
1301To make this easier, a given version of AnyEvent::Handle uses these
1302conventions:
1303
1304=over 4
1305
1306=item * all constructor arguments become object members.
1307
1308At least initially, when you pass a C<tls>-argument to the constructor it
1309will end up in C<< $handle->{tls} >>. Those members might be changes or
1310mutated later on (for example C<tls> will hold the TLS connection object).
1311
1312=item * other object member names are prefixed with an C<_>.
1313
1314All object members not explicitly documented (internal use) are prefixed
1315with an underscore character, so the remaining non-C<_>-namespace is free
1316for use for subclasses.
1317
1318=item * all members not documented here and not prefixed with an underscore
1319are free to use in subclasses.
1320
1321Of course, new versions of AnyEvent::Handle may introduce more "public"
1322member variables, but thats just life, at least it is documented.
1323
1324=back
1325
947=head1 AUTHOR 1326=head1 AUTHOR
948 1327
949Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 1328Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
950 1329
951=cut 1330=cut

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines