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.42 by root, Tue May 27 06:23:15 2008 UTC vs.
Revision 1.62 by root, Fri Jun 6 10:49:20 2008 UTC

… …
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 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;
… …
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->($handle) 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->($handle) 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
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100 102
101While 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
102you will not be notified of errors otherwise. The default simply calls 104you will not be notified of errors otherwise. The default simply calls
103die. 105C<croak>.
104 106
105=item on_read => $cb->($handle) 107=item on_read => $cb->($handle)
106 108
107This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
108and 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).
109 113
110To 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 >>
111method or access the C<$handle->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
112 116
113When 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
… …
119 123
120This 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
121(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).
122 126
123To 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.
124 148
125=item rbuf_max => <bytes> 149=item rbuf_max => <bytes>
126 150
127If 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>)
128when 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
… …
135isn't finished). 159isn't finished).
136 160
137=item read_size => <bytes> 161=item read_size => <bytes>
138 162
139The 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
140on each [loop iteration). Default: C<4096>. 164during each (loop iteration). Default: C<8192>.
141 165
142=item low_water_mark => <bytes> 166=item low_water_mark => <bytes>
143 167
144Sets 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
145buffer: 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
146considered empty. 170considered empty.
171
172=item linger => <seconds>
173
174If non-zero (default: C<3600>), then the destructor of the
175AnyEvent::Handle object will check wether there is still outstanding write
176data and will install a watcher that will write out this data. No errors
177will be reported (this mostly matches how the operating system treats
178outstanding data at socket close time).
179
180This will not work for partial TLS data that could not yet been
181encoded. This data will be lost.
147 182
148=item tls => "accept" | "connect" | Net::SSLeay::SSL object 183=item tls => "accept" | "connect" | Net::SSLeay::SSL object
149 184
150When this parameter is given, it enables TLS (SSL) mode, that means it 185When this parameter is given, it enables TLS (SSL) mode, that means it
151will start making tls handshake and will transparently encrypt/decrypt 186will start making tls handshake and will transparently encrypt/decrypt
… …
202 if ($self->{tls}) { 237 if ($self->{tls}) {
203 require Net::SSLeay; 238 require Net::SSLeay;
204 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 239 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
205 } 240 }
206 241
207 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 242 $self->{_activity} = AnyEvent->now;
208 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 243 $self->_timeout;
244
209 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
210 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
211
212 $self->start_read;
213 246
214 $self 247 $self
215} 248}
216 249
217sub _shutdown { 250sub _shutdown {
218 my ($self) = @_; 251 my ($self) = @_;
219 252
253 delete $self->{_tw};
220 delete $self->{_rw}; 254 delete $self->{_rw};
221 delete $self->{_ww}; 255 delete $self->{_ww};
222 delete $self->{fh}; 256 delete $self->{fh};
223}
224 257
258 $self->stoptls;
259}
260
225sub error { 261sub _error {
226 my ($self) = @_; 262 my ($self, $errno, $fatal) = @_;
227 263
228 {
229 local $!;
230 $self->_shutdown; 264 $self->_shutdown
231 } 265 if $fatal;
232 266
233 $self->{on_error}($self) 267 $! = $errno;
268
234 if $self->{on_error}; 269 if ($self->{on_error}) {
235 270 $self->{on_error}($self, $fatal);
271 } else {
236 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 272 Carp::croak "AnyEvent::Handle uncaught error: $!";
273 }
237} 274}
238 275
239=item $fh = $handle->fh 276=item $fh = $handle->fh
240 277
241This method returns the file handle of the L<AnyEvent::Handle> object. 278This method returns the file handle of the L<AnyEvent::Handle> object.
… …
260 297
261=cut 298=cut
262 299
263sub on_eof { 300sub on_eof {
264 $_[0]{on_eof} = $_[1]; 301 $_[0]{on_eof} = $_[1];
302}
303
304=item $handle->on_timeout ($cb)
305
306Replace the current C<on_timeout> callback, or disables the callback
307(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
308argument.
309
310=cut
311
312sub on_timeout {
313 $_[0]{on_timeout} = $_[1];
314}
315
316#############################################################################
317
318=item $handle->timeout ($seconds)
319
320Configures (or disables) the inactivity timeout.
321
322=cut
323
324sub timeout {
325 my ($self, $timeout) = @_;
326
327 $self->{timeout} = $timeout;
328 $self->_timeout;
329}
330
331# reset the timeout watcher, as neccessary
332# also check for time-outs
333sub _timeout {
334 my ($self) = @_;
335
336 if ($self->{timeout}) {
337 my $NOW = AnyEvent->now;
338
339 # when would the timeout trigger?
340 my $after = $self->{_activity} + $self->{timeout} - $NOW;
341
342 # now or in the past already?
343 if ($after <= 0) {
344 $self->{_activity} = $NOW;
345
346 if ($self->{on_timeout}) {
347 $self->{on_timeout}($self);
348 } else {
349 $self->_error (&Errno::ETIMEDOUT);
350 }
351
352 # callback could have changed timeout value, optimise
353 return unless $self->{timeout};
354
355 # calculate new after
356 $after = $self->{timeout};
357 }
358
359 Scalar::Util::weaken $self;
360 return unless $self; # ->error could have destroyed $self
361
362 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
363 delete $self->{_tw};
364 $self->_timeout;
365 });
366 } else {
367 delete $self->{_tw};
368 }
265} 369}
266 370
267############################################################################# 371#############################################################################
268 372
269=back 373=back
… …
316 my $len = syswrite $self->{fh}, $self->{wbuf}; 420 my $len = syswrite $self->{fh}, $self->{wbuf};
317 421
318 if ($len >= 0) { 422 if ($len >= 0) {
319 substr $self->{wbuf}, 0, $len, ""; 423 substr $self->{wbuf}, 0, $len, "";
320 424
425 $self->{_activity} = AnyEvent->now;
426
321 $self->{on_drain}($self) 427 $self->{on_drain}($self)
322 if $self->{low_water_mark} >= length $self->{wbuf} 428 if $self->{low_water_mark} >= length $self->{wbuf}
323 && $self->{on_drain}; 429 && $self->{on_drain};
324 430
325 delete $self->{_ww} unless length $self->{wbuf}; 431 delete $self->{_ww} unless length $self->{wbuf};
326 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 432 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
327 $self->error; 433 $self->_error ($!, 1);
328 } 434 }
329 }; 435 };
330 436
331 # try to write data immediately 437 # try to write data immediately
332 $cb->(); 438 $cb->();
… …
352 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 458 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
353 ->($self, @_); 459 ->($self, @_);
354 } 460 }
355 461
356 if ($self->{filter_w}) { 462 if ($self->{filter_w}) {
357 $self->{filter_w}->($self, \$_[0]); 463 $self->{filter_w}($self, \$_[0]);
358 } else { 464 } else {
359 $self->{wbuf} .= $_[0]; 465 $self->{wbuf} .= $_[0];
360 $self->_drain_wbuf; 466 $self->_drain_wbuf;
361 } 467 }
362} 468}
363 469
364=item $handle->push_write (type => @args) 470=item $handle->push_write (type => @args)
365 471
366=item $handle->unshift_write (type => @args)
367
368Instead of formatting your data yourself, you can also let this module do 472Instead of formatting your data yourself, you can also let this module do
369the job by specifying a type and type-specific arguments. 473the job by specifying a type and type-specific arguments.
370 474
371Predefined types are (if you have ideas for additional types, feel free to 475Predefined types are (if you have ideas for additional types, feel free to
372drop by and tell us): 476drop by and tell us):
… …
376=item netstring => $string 480=item netstring => $string
377 481
378Formats the given value as netstring 482Formats the given value as netstring
379(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 483(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
380 484
381=back
382
383=cut 485=cut
384 486
385register_write_type netstring => sub { 487register_write_type netstring => sub {
386 my ($self, $string) = @_; 488 my ($self, $string) = @_;
387 489
388 sprintf "%d:%s,", (length $string), $string 490 sprintf "%d:%s,", (length $string), $string
491};
492
493=item packstring => $format, $data
494
495An octet string prefixed with an encoded length. The encoding C<$format>
496uses the same format as a Perl C<pack> format, but must specify a single
497integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
498optional C<!>, C<< < >> or C<< > >> modifier).
499
500=cut
501
502register_write_type packstring => sub {
503 my ($self, $format, $string) = @_;
504
505 pack "$format/a", $string
389}; 506};
390 507
391=item json => $array_or_hashref 508=item json => $array_or_hashref
392 509
393Encodes the given hash or array reference into a JSON object. Unless you 510Encodes the given hash or array reference into a JSON object. Unless you
… …
427 544
428 $self->{json} ? $self->{json}->encode ($ref) 545 $self->{json} ? $self->{json}->encode ($ref)
429 : JSON::encode_json ($ref) 546 : JSON::encode_json ($ref)
430}; 547};
431 548
549=back
550
432=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 551=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
433 552
434This function (not method) lets you add your own types to C<push_write>. 553This function (not method) lets you add your own types to C<push_write>.
435Whenever the given C<type> is used, C<push_write> will invoke the code 554Whenever the given C<type> is used, C<push_write> will invoke the code
436reference with the handle object and the remaining arguments. 555reference with the handle object and the remaining arguments.
… …
461enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 580enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
462or not. 581or not.
463 582
464In the more complex case, you want to queue multiple callbacks. In this 583In the more complex case, you want to queue multiple callbacks. In this
465case, AnyEvent::Handle will call the first queued callback each time new 584case, AnyEvent::Handle will call the first queued callback each time new
466data arrives and removes it when it has done its job (see C<push_read>, 585data arrives (also the first time it is queued) and removes it when it has
467below). 586done its job (see C<push_read>, below).
468 587
469This way you can, for example, push three line-reads, followed by reading 588This way you can, for example, push three line-reads, followed by reading
470a chunk of data, and AnyEvent::Handle will execute them in order. 589a chunk of data, and AnyEvent::Handle will execute them in order.
471 590
472Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 591Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
473the specified number of bytes which give an XML datagram. 592the specified number of bytes which give an XML datagram.
474 593
475 # in the default state, expect some header bytes 594 # in the default state, expect some header bytes
476 $handle->on_read (sub { 595 $handle->on_read (sub {
477 # some data is here, now queue the length-header-read (4 octets) 596 # some data is here, now queue the length-header-read (4 octets)
478 shift->unshift_read_chunk (4, sub { 597 shift->unshift_read (chunk => 4, sub {
479 # header arrived, decode 598 # header arrived, decode
480 my $len = unpack "N", $_[1]; 599 my $len = unpack "N", $_[1];
481 600
482 # now read the payload 601 # now read the payload
483 shift->unshift_read_chunk ($len, sub { 602 shift->unshift_read (chunk => $len, sub {
484 my $xml = $_[1]; 603 my $xml = $_[1];
485 # handle xml 604 # handle xml
486 }); 605 });
487 }); 606 });
488 }); 607 });
… …
495 614
496 # request one 615 # request one
497 $handle->push_write ("request 1\015\012"); 616 $handle->push_write ("request 1\015\012");
498 617
499 # we expect "ERROR" or "OK" as response, so push a line read 618 # we expect "ERROR" or "OK" as response, so push a line read
500 $handle->push_read_line (sub { 619 $handle->push_read (line => sub {
501 # if we got an "OK", we have to _prepend_ another line, 620 # if we got an "OK", we have to _prepend_ another line,
502 # so it will be read before the second request reads its 64 bytes 621 # so it will be read before the second request reads its 64 bytes
503 # which are already in the queue when this callback is called 622 # which are already in the queue when this callback is called
504 # we don't do this in case we got an error 623 # we don't do this in case we got an error
505 if ($_[1] eq "OK") { 624 if ($_[1] eq "OK") {
506 $_[0]->unshift_read_line (sub { 625 $_[0]->unshift_read (line => sub {
507 my $response = $_[1]; 626 my $response = $_[1];
508 ... 627 ...
509 }); 628 });
510 } 629 }
511 }); 630 });
512 631
513 # request two 632 # request two
514 $handle->push_write ("request 2\015\012"); 633 $handle->push_write ("request 2\015\012");
515 634
516 # simply read 64 bytes, always 635 # simply read 64 bytes, always
517 $handle->push_read_chunk (64, sub { 636 $handle->push_read (chunk => 64, sub {
518 my $response = $_[1]; 637 my $response = $_[1];
519 ... 638 ...
520 }); 639 });
521 640
522=over 4 641=over 4
523 642
524=cut 643=cut
525 644
526sub _drain_rbuf { 645sub _drain_rbuf {
527 my ($self) = @_; 646 my ($self) = @_;
647
648 local $self->{_in_drain} = 1;
528 649
529 if ( 650 if (
530 defined $self->{rbuf_max} 651 defined $self->{rbuf_max}
531 && $self->{rbuf_max} < length $self->{rbuf} 652 && $self->{rbuf_max} < length $self->{rbuf}
532 ) { 653 ) {
533 $! = &Errno::ENOSPC; 654 return $self->_error (&Errno::ENOSPC, 1);
534 $self->error;
535 } 655 }
536 656
537 return if $self->{in_drain}; 657 while () {
538 local $self->{in_drain} = 1;
539
540 while (my $len = length $self->{rbuf}) {
541 no strict 'refs'; 658 no strict 'refs';
659
660 my $len = length $self->{rbuf};
661
542 if (my $cb = shift @{ $self->{_queue} }) { 662 if (my $cb = shift @{ $self->{_queue} }) {
543 unless ($cb->($self)) { 663 unless ($cb->($self)) {
544 if ($self->{_eof}) { 664 if ($self->{_eof}) {
545 # no progress can be made (not enough data and no data forthcoming) 665 # no progress can be made (not enough data and no data forthcoming)
546 $! = &Errno::EPIPE; 666 $self->_error (&Errno::EPIPE, 1), last;
547 $self->error;
548 } 667 }
549 668
550 unshift @{ $self->{_queue} }, $cb; 669 unshift @{ $self->{_queue} }, $cb;
551 return; 670 last;
552 } 671 }
553 } elsif ($self->{on_read}) { 672 } elsif ($self->{on_read}) {
673 last unless $len;
674
554 $self->{on_read}($self); 675 $self->{on_read}($self);
555 676
556 if ( 677 if (
557 $self->{_eof} # if no further data will arrive
558 && $len == length $self->{rbuf} # and no data has been consumed 678 $len == length $self->{rbuf} # if no data has been consumed
559 && !@{ $self->{_queue} } # and the queue is still empty 679 && !@{ $self->{_queue} } # and the queue is still empty
560 && $self->{on_read} # and we still want to read data 680 && $self->{on_read} # but we still have on_read
561 ) { 681 ) {
682 # no further data will arrive
562 # then no progress can be made 683 # so no progress can be made
563 $! = &Errno::EPIPE; 684 $self->_error (&Errno::EPIPE, 1), last
564 $self->error; 685 if $self->{_eof};
686
687 last; # more data might arrive
565 } 688 }
566 } else { 689 } else {
567 # read side becomes idle 690 # read side becomes idle
568 delete $self->{_rw}; 691 delete $self->{_rw};
569 return; 692 last;
570 } 693 }
571 } 694 }
572 695
573 if ($self->{_eof}) {
574 $self->_shutdown;
575 $self->{on_eof}($self) 696 $self->{on_eof}($self)
576 if $self->{on_eof}; 697 if $self->{_eof} && $self->{on_eof};
698
699 # may need to restart read watcher
700 unless ($self->{_rw}) {
701 $self->start_read
702 if $self->{on_read} || @{ $self->{_queue} };
577 } 703 }
578} 704}
579 705
580=item $handle->on_read ($cb) 706=item $handle->on_read ($cb)
581 707
… …
587 713
588sub on_read { 714sub on_read {
589 my ($self, $cb) = @_; 715 my ($self, $cb) = @_;
590 716
591 $self->{on_read} = $cb; 717 $self->{on_read} = $cb;
718 $self->_drain_rbuf if $cb && !$self->{_in_drain};
592} 719}
593 720
594=item $handle->rbuf 721=item $handle->rbuf
595 722
596Returns the read buffer (as a modifiable lvalue). 723Returns the read buffer (as a modifiable lvalue).
… …
645 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 772 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
646 ->($self, $cb, @_); 773 ->($self, $cb, @_);
647 } 774 }
648 775
649 push @{ $self->{_queue} }, $cb; 776 push @{ $self->{_queue} }, $cb;
650 $self->_drain_rbuf; 777 $self->_drain_rbuf unless $self->{_in_drain};
651} 778}
652 779
653sub unshift_read { 780sub unshift_read {
654 my $self = shift; 781 my $self = shift;
655 my $cb = pop; 782 my $cb = pop;
… …
661 ->($self, $cb, @_); 788 ->($self, $cb, @_);
662 } 789 }
663 790
664 791
665 unshift @{ $self->{_queue} }, $cb; 792 unshift @{ $self->{_queue} }, $cb;
666 $self->_drain_rbuf; 793 $self->_drain_rbuf unless $self->{_in_drain};
667} 794}
668 795
669=item $handle->push_read (type => @args, $cb) 796=item $handle->push_read (type => @args, $cb)
670 797
671=item $handle->unshift_read (type => @args, $cb) 798=item $handle->unshift_read (type => @args, $cb)
… …
756 883
757sub unshift_read_line { 884sub unshift_read_line {
758 my $self = shift; 885 my $self = shift;
759 $self->unshift_read (line => @_); 886 $self->unshift_read (line => @_);
760} 887}
761
762=item netstring => $cb->($handle, $string)
763
764A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
765
766Throws an error with C<$!> set to EBADMSG on format violations.
767
768=cut
769
770register_read_type netstring => sub {
771 my ($self, $cb) = @_;
772
773 sub {
774 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
775 if ($_[0]{rbuf} =~ /[^0-9]/) {
776 $! = &Errno::EBADMSG;
777 $self->error;
778 }
779 return;
780 }
781
782 my $len = $1;
783
784 $self->unshift_read (chunk => $len, sub {
785 my $string = $_[1];
786 $_[0]->unshift_read (chunk => 1, sub {
787 if ($_[1] eq ",") {
788 $cb->($_[0], $string);
789 } else {
790 $! = &Errno::EBADMSG;
791 $self->error;
792 }
793 });
794 });
795
796 1
797 }
798};
799 888
800=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 889=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
801 890
802Makes a regex match against the regex object C<$accept> and returns 891Makes a regex match against the regex object C<$accept> and returns
803everything up to and including the match. 892everything up to and including the match.
… …
853 return 1; 942 return 1;
854 } 943 }
855 944
856 # reject 945 # reject
857 if ($reject && $$rbuf =~ $reject) { 946 if ($reject && $$rbuf =~ $reject) {
858 $! = &Errno::EBADMSG; 947 $self->_error (&Errno::EBADMSG);
859 $self->error;
860 } 948 }
861 949
862 # skip 950 # skip
863 if ($skip && $$rbuf =~ $skip) { 951 if ($skip && $$rbuf =~ $skip) {
864 $data .= substr $$rbuf, 0, $+[0], ""; 952 $data .= substr $$rbuf, 0, $+[0], "";
865 } 953 }
866 954
867 () 955 ()
956 }
957};
958
959=item netstring => $cb->($handle, $string)
960
961A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
962
963Throws an error with C<$!> set to EBADMSG on format violations.
964
965=cut
966
967register_read_type netstring => sub {
968 my ($self, $cb) = @_;
969
970 sub {
971 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
972 if ($_[0]{rbuf} =~ /[^0-9]/) {
973 $self->_error (&Errno::EBADMSG);
974 }
975 return;
976 }
977
978 my $len = $1;
979
980 $self->unshift_read (chunk => $len, sub {
981 my $string = $_[1];
982 $_[0]->unshift_read (chunk => 1, sub {
983 if ($_[1] eq ",") {
984 $cb->($_[0], $string);
985 } else {
986 $self->_error (&Errno::EBADMSG);
987 }
988 });
989 });
990
991 1
992 }
993};
994
995=item packstring => $format, $cb->($handle, $string)
996
997An octet string prefixed with an encoded length. The encoding C<$format>
998uses the same format as a Perl C<pack> format, but must specify a single
999integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1000optional C<!>, C<< < >> or C<< > >> modifier).
1001
1002DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1003
1004Example: read a block of data prefixed by its length in BER-encoded
1005format (very efficient).
1006
1007 $handle->push_read (packstring => "w", sub {
1008 my ($handle, $data) = @_;
1009 });
1010
1011=cut
1012
1013register_read_type packstring => sub {
1014 my ($self, $cb, $format) = @_;
1015
1016 sub {
1017 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1018 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1019 or return;
1020
1021 # remove prefix
1022 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1023
1024 # read rest
1025 $_[0]->unshift_read (chunk => $len, $cb);
1026
1027 1
868 } 1028 }
869}; 1029};
870 1030
871=item json => $cb->($handle, $hash_or_arrayref) 1031=item json => $cb->($handle, $hash_or_arrayref)
872 1032
… …
937=item $handle->stop_read 1097=item $handle->stop_read
938 1098
939=item $handle->start_read 1099=item $handle->start_read
940 1100
941In rare cases you actually do not want to read anything from the 1101In rare cases you actually do not want to read anything from the
942socket. In this case you can call C<stop_read>. Neither C<on_read> no 1102socket. In this case you can call C<stop_read>. Neither C<on_read> nor
943any queued callbacks will be executed then. To start reading again, call 1103any queued callbacks will be executed then. To start reading again, call
944C<start_read>. 1104C<start_read>.
1105
1106Note that AnyEvent::Handle will automatically C<start_read> for you when
1107you change the C<on_read> callback or push/unshift a read callback, and it
1108will automatically C<stop_read> for you when neither C<on_read> is set nor
1109there are any read requests in the queue.
945 1110
946=cut 1111=cut
947 1112
948sub stop_read { 1113sub stop_read {
949 my ($self) = @_; 1114 my ($self) = @_;
… …
960 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1125 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
961 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1126 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
962 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1127 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
963 1128
964 if ($len > 0) { 1129 if ($len > 0) {
1130 $self->{_activity} = AnyEvent->now;
1131
965 $self->{filter_r} 1132 $self->{filter_r}
966 ? $self->{filter_r}->($self, $rbuf) 1133 ? $self->{filter_r}($self, $rbuf)
967 : $self->_drain_rbuf; 1134 : $self->{_in_drain} || $self->_drain_rbuf;
968 1135
969 } elsif (defined $len) { 1136 } elsif (defined $len) {
970 delete $self->{_rw}; 1137 delete $self->{_rw};
971 $self->{_eof} = 1; 1138 $self->{_eof} = 1;
972 $self->_drain_rbuf; 1139 $self->_drain_rbuf unless $self->{_in_drain};
973 1140
974 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1141 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
975 return $self->error; 1142 return $self->_error ($!, 1);
976 } 1143 }
977 }); 1144 });
978 } 1145 }
979} 1146}
980 1147
981sub _dotls { 1148sub _dotls {
982 my ($self) = @_; 1149 my ($self) = @_;
1150
1151 my $buf;
983 1152
984 if (length $self->{_tls_wbuf}) { 1153 if (length $self->{_tls_wbuf}) {
985 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1154 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
986 substr $self->{_tls_wbuf}, 0, $len, ""; 1155 substr $self->{_tls_wbuf}, 0, $len, "";
987 } 1156 }
988 } 1157 }
989 1158
990 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1159 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
991 $self->{wbuf} .= $buf; 1160 $self->{wbuf} .= $buf;
992 $self->_drain_wbuf; 1161 $self->_drain_wbuf;
993 } 1162 }
994 1163
995 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1164 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1165 if (length $buf) {
996 $self->{rbuf} .= $buf; 1166 $self->{rbuf} .= $buf;
997 $self->_drain_rbuf; 1167 $self->_drain_rbuf unless $self->{_in_drain};
1168 } else {
1169 # let's treat SSL-eof as we treat normal EOF
1170 $self->{_eof} = 1;
1171 $self->_shutdown;
1172 return;
1173 }
998 } 1174 }
999 1175
1000 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1176 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1001 1177
1002 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1178 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1003 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1179 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1004 $self->error; 1180 return $self->_error ($!, 1);
1005 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1181 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1006 $! = &Errno::EIO; 1182 return $self->_error (&Errno::EIO, 1);
1007 $self->error;
1008 } 1183 }
1009 1184
1010 # all others are fine for our purposes 1185 # all others are fine for our purposes
1011 } 1186 }
1012} 1187}
… …
1027call and can be used or changed to your liking. Note that the handshake 1202call and can be used or changed to your liking. Note that the handshake
1028might have already started when this function returns. 1203might have already started when this function returns.
1029 1204
1030=cut 1205=cut
1031 1206
1032# TODO: maybe document...
1033sub starttls { 1207sub starttls {
1034 my ($self, $ssl, $ctx) = @_; 1208 my ($self, $ssl, $ctx) = @_;
1035 1209
1036 $self->stoptls; 1210 $self->stoptls;
1037 1211
… …
1090 1264
1091sub DESTROY { 1265sub DESTROY {
1092 my $self = shift; 1266 my $self = shift;
1093 1267
1094 $self->stoptls; 1268 $self->stoptls;
1269
1270 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1271
1272 if ($linger && length $self->{wbuf}) {
1273 my $fh = delete $self->{fh};
1274 my $wbuf = delete $self->{wbuf};
1275
1276 my @linger;
1277
1278 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1279 my $len = syswrite $fh, $wbuf, length $wbuf;
1280
1281 if ($len > 0) {
1282 substr $wbuf, 0, $len, "";
1283 } else {
1284 @linger = (); # end
1285 }
1286 });
1287 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1288 @linger = ();
1289 });
1290 }
1095} 1291}
1096 1292
1097=item AnyEvent::Handle::TLS_CTX 1293=item AnyEvent::Handle::TLS_CTX
1098 1294
1099This function creates and returns the Net::SSLeay::CTX object used by 1295This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

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