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.43 by root, Wed May 28 23:57:38 2008 UTC vs.
Revision 1.62 by root, Fri Jun 6 10:49:20 2008 UTC

… …
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);
12use Time::HiRes qw(time);
13 12
14=head1 NAME 13=head1 NAME
15 14
16AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
17 16
18=cut 17=cut
19 18
20our $VERSION = '0.04'; 19our $VERSION = 4.14;
21 20
22=head1 SYNOPSIS 21=head1 SYNOPSIS
23 22
24 use AnyEvent; 23 use AnyEvent;
25 use AnyEvent::Handle; 24 use AnyEvent::Handle;
… …
76NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
77AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
78 77
79=item on_eof => $cb->($handle) 78=item on_eof => $cb->($handle)
80 79
81Set 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.
82 83
83While not mandatory, it is highly recommended to set an eof callback, 84While not mandatory, it is highly recommended to set an eof callback,
84otherwise 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
85waiting for data. 86waiting for data.
86 87
87=item on_error => $cb->($handle) 88=item on_error => $cb->($handle, $fatal)
88 89
89This 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
90occurs, such as not being able to resolve the hostname, failure to connect 91occured, such as not being able to resolve the hostname, failure to
91or a read error. 92connect or a read error.
92 93
93The 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
94called. 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.
95 99
96On callback entrance, the value of C<$!> contains the operating system 100On callback entrance, the value of C<$!> contains the operating system
97error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
98 102
99The callback should throw an exception. If it returns, then
100AnyEvent::Handle will C<croak> for you.
101
102While 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
103you will not be notified of errors otherwise. The default simply calls 104you will not be notified of errors otherwise. The default simply calls
104die. 105C<croak>.
105 106
106=item on_read => $cb->($handle) 107=item on_read => $cb->($handle)
107 108
108This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
109and 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).
110 113
111To 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 >>
112method or access the C<$handle->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
113 116
114When 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
… …
126=item timeout => $fractional_seconds 129=item timeout => $fractional_seconds
127 130
128If non-zero, then this enables an "inactivity" timeout: whenever this many 131If non-zero, then this enables an "inactivity" timeout: whenever this many
129seconds pass without a successful read or write on the underlying file 132seconds pass without a successful read or write on the underlying file
130handle, the C<on_timeout> callback will be invoked (and if that one is 133handle, the C<on_timeout> callback will be invoked (and if that one is
131missing, an C<ETIMEDOUT> errror will be raised). 134missing, an C<ETIMEDOUT> error will be raised).
132 135
133Note that timeout processing is also active when you currently do not have 136Note that timeout processing is also active when you currently do not have
134any outstanding read or write requests: If you plan to keep the connection 137any outstanding read or write requests: If you plan to keep the connection
135idle then you should disable the timout temporarily or ignore the timeout 138idle then you should disable the timout temporarily or ignore the timeout
136in the C<on_timeout> callback. 139in the C<on_timeout> callback.
… …
156isn't finished). 159isn't finished).
157 160
158=item read_size => <bytes> 161=item read_size => <bytes>
159 162
160The 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
161on each [loop iteration). Default: C<4096>. 164during each (loop iteration). Default: C<8192>.
162 165
163=item low_water_mark => <bytes> 166=item low_water_mark => <bytes>
164 167
165Sets 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
166buffer: 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
167considered 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.
168 182
169=item tls => "accept" | "connect" | Net::SSLeay::SSL object 183=item tls => "accept" | "connect" | Net::SSLeay::SSL object
170 184
171When 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
172will start making tls handshake and will transparently encrypt/decrypt 186will start making tls handshake and will transparently encrypt/decrypt
… …
223 if ($self->{tls}) { 237 if ($self->{tls}) {
224 require Net::SSLeay; 238 require Net::SSLeay;
225 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 239 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
226 } 240 }
227 241
228# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop 242 $self->{_activity} = AnyEvent->now;
229# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop 243 $self->_timeout;
230# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop 244
231 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
232
233 $self->{_activity} = time;
234 $self->_timeout;
235
236 $self->start_read;
237 246
238 $self 247 $self
239} 248}
240 249
241sub _shutdown { 250sub _shutdown {
242 my ($self) = @_; 251 my ($self) = @_;
243 252
253 delete $self->{_tw};
244 delete $self->{_rw}; 254 delete $self->{_rw};
245 delete $self->{_ww}; 255 delete $self->{_ww};
246 delete $self->{fh}; 256 delete $self->{fh};
247}
248 257
258 $self->stoptls;
259}
260
249sub error { 261sub _error {
250 my ($self) = @_; 262 my ($self, $errno, $fatal) = @_;
251 263
252 {
253 local $!;
254 $self->_shutdown; 264 $self->_shutdown
255 } 265 if $fatal;
256 266
257 $self->{on_error}($self) 267 $! = $errno;
268
258 if $self->{on_error}; 269 if ($self->{on_error}) {
259 270 $self->{on_error}($self, $fatal);
271 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 272 Carp::croak "AnyEvent::Handle uncaught error: $!";
273 }
261} 274}
262 275
263=item $fh = $handle->fh 276=item $fh = $handle->fh
264 277
265This method returns the file handle of the L<AnyEvent::Handle> object. 278This method returns the file handle of the L<AnyEvent::Handle> object.
… …
319# also check for time-outs 332# also check for time-outs
320sub _timeout { 333sub _timeout {
321 my ($self) = @_; 334 my ($self) = @_;
322 335
323 if ($self->{timeout}) { 336 if ($self->{timeout}) {
324 my $NOW = time; 337 my $NOW = AnyEvent->now;
325 338
326 # when would the timeout trigger? 339 # when would the timeout trigger?
327 my $after = $self->{_activity} + $self->{timeout} - $NOW; 340 my $after = $self->{_activity} + $self->{timeout} - $NOW;
328
329 warn "next to in $after\n";#d#
330 341
331 # now or in the past already? 342 # now or in the past already?
332 if ($after <= 0) { 343 if ($after <= 0) {
333 $self->{_activity} = $NOW; 344 $self->{_activity} = $NOW;
334 345
335 if ($self->{on_timeout}) { 346 if ($self->{on_timeout}) {
336 $self->{on_timeout}->($self); 347 $self->{on_timeout}($self);
337 } else { 348 } else {
338 $! = Errno::ETIMEDOUT; 349 $self->_error (&Errno::ETIMEDOUT);
339 $self->error;
340 } 350 }
341 351
342 # callbakx could have changed timeout value, optimise 352 # callback could have changed timeout value, optimise
343 return unless $self->{timeout}; 353 return unless $self->{timeout};
344 354
345 # calculate new after 355 # calculate new after
346 $after = $self->{timeout}; 356 $after = $self->{timeout};
347 } 357 }
348 358
349 Scalar::Util::weaken $self; 359 Scalar::Util::weaken $self;
360 return unless $self; # ->error could have destroyed $self
350 361
351 warn "after $after\n";#d#
352 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 362 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
353 delete $self->{_tw}; 363 delete $self->{_tw};
354 $self->_timeout; 364 $self->_timeout;
355 }); 365 });
356 } else { 366 } else {
… …
410 my $len = syswrite $self->{fh}, $self->{wbuf}; 420 my $len = syswrite $self->{fh}, $self->{wbuf};
411 421
412 if ($len >= 0) { 422 if ($len >= 0) {
413 substr $self->{wbuf}, 0, $len, ""; 423 substr $self->{wbuf}, 0, $len, "";
414 424
415 $self->{_activity} = time; 425 $self->{_activity} = AnyEvent->now;
416 426
417 $self->{on_drain}($self) 427 $self->{on_drain}($self)
418 if $self->{low_water_mark} >= length $self->{wbuf} 428 if $self->{low_water_mark} >= length $self->{wbuf}
419 && $self->{on_drain}; 429 && $self->{on_drain};
420 430
421 delete $self->{_ww} unless length $self->{wbuf}; 431 delete $self->{_ww} unless length $self->{wbuf};
422 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 432 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
423 $self->error; 433 $self->_error ($!, 1);
424 } 434 }
425 }; 435 };
426 436
427 # try to write data immediately 437 # try to write data immediately
428 $cb->(); 438 $cb->();
… …
448 @_ = ($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")
449 ->($self, @_); 459 ->($self, @_);
450 } 460 }
451 461
452 if ($self->{filter_w}) { 462 if ($self->{filter_w}) {
453 $self->{filter_w}->($self, \$_[0]); 463 $self->{filter_w}($self, \$_[0]);
454 } else { 464 } else {
455 $self->{wbuf} .= $_[0]; 465 $self->{wbuf} .= $_[0];
456 $self->_drain_wbuf; 466 $self->_drain_wbuf;
457 } 467 }
458} 468}
459 469
460=item $handle->push_write (type => @args) 470=item $handle->push_write (type => @args)
461 471
462=item $handle->unshift_write (type => @args)
463
464Instead 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
465the job by specifying a type and type-specific arguments. 473the job by specifying a type and type-specific arguments.
466 474
467Predefined 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
468drop by and tell us): 476drop by and tell us):
… …
472=item netstring => $string 480=item netstring => $string
473 481
474Formats the given value as netstring 482Formats the given value as netstring
475(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).
476 484
477=back
478
479=cut 485=cut
480 486
481register_write_type netstring => sub { 487register_write_type netstring => sub {
482 my ($self, $string) = @_; 488 my ($self, $string) = @_;
483 489
484 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
485}; 506};
486 507
487=item json => $array_or_hashref 508=item json => $array_or_hashref
488 509
489Encodes 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
… …
523 544
524 $self->{json} ? $self->{json}->encode ($ref) 545 $self->{json} ? $self->{json}->encode ($ref)
525 : JSON::encode_json ($ref) 546 : JSON::encode_json ($ref)
526}; 547};
527 548
549=back
550
528=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 551=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
529 552
530This 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>.
531Whenever 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
532reference with the handle object and the remaining arguments. 555reference with the handle object and the remaining arguments.
… …
557enough 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
558or not. 581or not.
559 582
560In 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
561case, AnyEvent::Handle will call the first queued callback each time new 584case, AnyEvent::Handle will call the first queued callback each time new
562data 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
563below). 586done its job (see C<push_read>, below).
564 587
565This 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
566a chunk of data, and AnyEvent::Handle will execute them in order. 589a chunk of data, and AnyEvent::Handle will execute them in order.
567 590
568Example 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
569the specified number of bytes which give an XML datagram. 592the specified number of bytes which give an XML datagram.
570 593
571 # in the default state, expect some header bytes 594 # in the default state, expect some header bytes
572 $handle->on_read (sub { 595 $handle->on_read (sub {
573 # 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)
574 shift->unshift_read_chunk (4, sub { 597 shift->unshift_read (chunk => 4, sub {
575 # header arrived, decode 598 # header arrived, decode
576 my $len = unpack "N", $_[1]; 599 my $len = unpack "N", $_[1];
577 600
578 # now read the payload 601 # now read the payload
579 shift->unshift_read_chunk ($len, sub { 602 shift->unshift_read (chunk => $len, sub {
580 my $xml = $_[1]; 603 my $xml = $_[1];
581 # handle xml 604 # handle xml
582 }); 605 });
583 }); 606 });
584 }); 607 });
… …
591 614
592 # request one 615 # request one
593 $handle->push_write ("request 1\015\012"); 616 $handle->push_write ("request 1\015\012");
594 617
595 # 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
596 $handle->push_read_line (sub { 619 $handle->push_read (line => sub {
597 # if we got an "OK", we have to _prepend_ another line, 620 # if we got an "OK", we have to _prepend_ another line,
598 # 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
599 # which are already in the queue when this callback is called 622 # which are already in the queue when this callback is called
600 # we don't do this in case we got an error 623 # we don't do this in case we got an error
601 if ($_[1] eq "OK") { 624 if ($_[1] eq "OK") {
602 $_[0]->unshift_read_line (sub { 625 $_[0]->unshift_read (line => sub {
603 my $response = $_[1]; 626 my $response = $_[1];
604 ... 627 ...
605 }); 628 });
606 } 629 }
607 }); 630 });
608 631
609 # request two 632 # request two
610 $handle->push_write ("request 2\015\012"); 633 $handle->push_write ("request 2\015\012");
611 634
612 # simply read 64 bytes, always 635 # simply read 64 bytes, always
613 $handle->push_read_chunk (64, sub { 636 $handle->push_read (chunk => 64, sub {
614 my $response = $_[1]; 637 my $response = $_[1];
615 ... 638 ...
616 }); 639 });
617 640
618=over 4 641=over 4
619 642
620=cut 643=cut
621 644
622sub _drain_rbuf { 645sub _drain_rbuf {
623 my ($self) = @_; 646 my ($self) = @_;
647
648 local $self->{_in_drain} = 1;
624 649
625 if ( 650 if (
626 defined $self->{rbuf_max} 651 defined $self->{rbuf_max}
627 && $self->{rbuf_max} < length $self->{rbuf} 652 && $self->{rbuf_max} < length $self->{rbuf}
628 ) { 653 ) {
629 $! = &Errno::ENOSPC; 654 return $self->_error (&Errno::ENOSPC, 1);
630 $self->error;
631 } 655 }
632 656
633 return if $self->{in_drain}; 657 while () {
634 local $self->{in_drain} = 1;
635
636 while (my $len = length $self->{rbuf}) {
637 no strict 'refs'; 658 no strict 'refs';
659
660 my $len = length $self->{rbuf};
661
638 if (my $cb = shift @{ $self->{_queue} }) { 662 if (my $cb = shift @{ $self->{_queue} }) {
639 unless ($cb->($self)) { 663 unless ($cb->($self)) {
640 if ($self->{_eof}) { 664 if ($self->{_eof}) {
641 # 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)
642 $! = &Errno::EPIPE; 666 $self->_error (&Errno::EPIPE, 1), last;
643 $self->error;
644 } 667 }
645 668
646 unshift @{ $self->{_queue} }, $cb; 669 unshift @{ $self->{_queue} }, $cb;
647 return; 670 last;
648 } 671 }
649 } elsif ($self->{on_read}) { 672 } elsif ($self->{on_read}) {
673 last unless $len;
674
650 $self->{on_read}($self); 675 $self->{on_read}($self);
651 676
652 if ( 677 if (
653 $self->{_eof} # if no further data will arrive
654 && $len == length $self->{rbuf} # and no data has been consumed 678 $len == length $self->{rbuf} # if no data has been consumed
655 && !@{ $self->{_queue} } # and the queue is still empty 679 && !@{ $self->{_queue} } # and the queue is still empty
656 && $self->{on_read} # and we still want to read data 680 && $self->{on_read} # but we still have on_read
657 ) { 681 ) {
682 # no further data will arrive
658 # then no progress can be made 683 # so no progress can be made
659 $! = &Errno::EPIPE; 684 $self->_error (&Errno::EPIPE, 1), last
660 $self->error; 685 if $self->{_eof};
686
687 last; # more data might arrive
661 } 688 }
662 } else { 689 } else {
663 # read side becomes idle 690 # read side becomes idle
664 delete $self->{_rw}; 691 delete $self->{_rw};
665 return; 692 last;
666 } 693 }
667 } 694 }
668 695
669 if ($self->{_eof}) {
670 $self->_shutdown;
671 $self->{on_eof}($self) 696 $self->{on_eof}($self)
672 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} };
673 } 703 }
674} 704}
675 705
676=item $handle->on_read ($cb) 706=item $handle->on_read ($cb)
677 707
… …
683 713
684sub on_read { 714sub on_read {
685 my ($self, $cb) = @_; 715 my ($self, $cb) = @_;
686 716
687 $self->{on_read} = $cb; 717 $self->{on_read} = $cb;
718 $self->_drain_rbuf if $cb && !$self->{_in_drain};
688} 719}
689 720
690=item $handle->rbuf 721=item $handle->rbuf
691 722
692Returns the read buffer (as a modifiable lvalue). 723Returns the read buffer (as a modifiable lvalue).
… …
741 $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")
742 ->($self, $cb, @_); 773 ->($self, $cb, @_);
743 } 774 }
744 775
745 push @{ $self->{_queue} }, $cb; 776 push @{ $self->{_queue} }, $cb;
746 $self->_drain_rbuf; 777 $self->_drain_rbuf unless $self->{_in_drain};
747} 778}
748 779
749sub unshift_read { 780sub unshift_read {
750 my $self = shift; 781 my $self = shift;
751 my $cb = pop; 782 my $cb = pop;
… …
757 ->($self, $cb, @_); 788 ->($self, $cb, @_);
758 } 789 }
759 790
760 791
761 unshift @{ $self->{_queue} }, $cb; 792 unshift @{ $self->{_queue} }, $cb;
762 $self->_drain_rbuf; 793 $self->_drain_rbuf unless $self->{_in_drain};
763} 794}
764 795
765=item $handle->push_read (type => @args, $cb) 796=item $handle->push_read (type => @args, $cb)
766 797
767=item $handle->unshift_read (type => @args, $cb) 798=item $handle->unshift_read (type => @args, $cb)
… …
852 883
853sub unshift_read_line { 884sub unshift_read_line {
854 my $self = shift; 885 my $self = shift;
855 $self->unshift_read (line => @_); 886 $self->unshift_read (line => @_);
856} 887}
857
858=item netstring => $cb->($handle, $string)
859
860A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
861
862Throws an error with C<$!> set to EBADMSG on format violations.
863
864=cut
865
866register_read_type netstring => sub {
867 my ($self, $cb) = @_;
868
869 sub {
870 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
871 if ($_[0]{rbuf} =~ /[^0-9]/) {
872 $! = &Errno::EBADMSG;
873 $self->error;
874 }
875 return;
876 }
877
878 my $len = $1;
879
880 $self->unshift_read (chunk => $len, sub {
881 my $string = $_[1];
882 $_[0]->unshift_read (chunk => 1, sub {
883 if ($_[1] eq ",") {
884 $cb->($_[0], $string);
885 } else {
886 $! = &Errno::EBADMSG;
887 $self->error;
888 }
889 });
890 });
891
892 1
893 }
894};
895 888
896=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 889=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
897 890
898Makes a regex match against the regex object C<$accept> and returns 891Makes a regex match against the regex object C<$accept> and returns
899everything up to and including the match. 892everything up to and including the match.
… …
949 return 1; 942 return 1;
950 } 943 }
951 944
952 # reject 945 # reject
953 if ($reject && $$rbuf =~ $reject) { 946 if ($reject && $$rbuf =~ $reject) {
954 $! = &Errno::EBADMSG; 947 $self->_error (&Errno::EBADMSG);
955 $self->error;
956 } 948 }
957 949
958 # skip 950 # skip
959 if ($skip && $$rbuf =~ $skip) { 951 if ($skip && $$rbuf =~ $skip) {
960 $data .= substr $$rbuf, 0, $+[0], ""; 952 $data .= substr $$rbuf, 0, $+[0], "";
961 } 953 }
962 954
963 () 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
964 } 1028 }
965}; 1029};
966 1030
967=item json => $cb->($handle, $hash_or_arrayref) 1031=item json => $cb->($handle, $hash_or_arrayref)
968 1032
… …
1033=item $handle->stop_read 1097=item $handle->stop_read
1034 1098
1035=item $handle->start_read 1099=item $handle->start_read
1036 1100
1037In 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
1038socket. 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
1039any queued callbacks will be executed then. To start reading again, call 1103any queued callbacks will be executed then. To start reading again, call
1040C<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.
1041 1110
1042=cut 1111=cut
1043 1112
1044sub stop_read { 1113sub stop_read {
1045 my ($self) = @_; 1114 my ($self) = @_;
… …
1056 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1125 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1057 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1126 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
1058 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;
1059 1128
1060 if ($len > 0) { 1129 if ($len > 0) {
1061 $self->{_activity} = time; 1130 $self->{_activity} = AnyEvent->now;
1062 1131
1063 $self->{filter_r} 1132 $self->{filter_r}
1064 ? $self->{filter_r}->($self, $rbuf) 1133 ? $self->{filter_r}($self, $rbuf)
1065 : $self->_drain_rbuf; 1134 : $self->{_in_drain} || $self->_drain_rbuf;
1066 1135
1067 } elsif (defined $len) { 1136 } elsif (defined $len) {
1068 delete $self->{_rw}; 1137 delete $self->{_rw};
1069 delete $self->{_ww};
1070 delete $self->{_tw};
1071 $self->{_eof} = 1; 1138 $self->{_eof} = 1;
1072 $self->_drain_rbuf; 1139 $self->_drain_rbuf unless $self->{_in_drain};
1073 1140
1074 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1141 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1075 return $self->error; 1142 return $self->_error ($!, 1);
1076 } 1143 }
1077 }); 1144 });
1078 } 1145 }
1079} 1146}
1080 1147
1081sub _dotls { 1148sub _dotls {
1082 my ($self) = @_; 1149 my ($self) = @_;
1150
1151 my $buf;
1083 1152
1084 if (length $self->{_tls_wbuf}) { 1153 if (length $self->{_tls_wbuf}) {
1085 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) {
1086 substr $self->{_tls_wbuf}, 0, $len, ""; 1155 substr $self->{_tls_wbuf}, 0, $len, "";
1087 } 1156 }
1088 } 1157 }
1089 1158
1090 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1159 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1091 $self->{wbuf} .= $buf; 1160 $self->{wbuf} .= $buf;
1092 $self->_drain_wbuf; 1161 $self->_drain_wbuf;
1093 } 1162 }
1094 1163
1095 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1164 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1165 if (length $buf) {
1096 $self->{rbuf} .= $buf; 1166 $self->{rbuf} .= $buf;
1097 $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 }
1098 } 1174 }
1099 1175
1100 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1176 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1101 1177
1102 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1178 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1103 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1179 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1104 $self->error; 1180 return $self->_error ($!, 1);
1105 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1181 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1106 $! = &Errno::EIO; 1182 return $self->_error (&Errno::EIO, 1);
1107 $self->error;
1108 } 1183 }
1109 1184
1110 # all others are fine for our purposes 1185 # all others are fine for our purposes
1111 } 1186 }
1112} 1187}
… …
1127call 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
1128might have already started when this function returns. 1203might have already started when this function returns.
1129 1204
1130=cut 1205=cut
1131 1206
1132# TODO: maybe document...
1133sub starttls { 1207sub starttls {
1134 my ($self, $ssl, $ctx) = @_; 1208 my ($self, $ssl, $ctx) = @_;
1135 1209
1136 $self->stoptls; 1210 $self->stoptls;
1137 1211
… …
1190 1264
1191sub DESTROY { 1265sub DESTROY {
1192 my $self = shift; 1266 my $self = shift;
1193 1267
1194 $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 }
1195} 1291}
1196 1292
1197=item AnyEvent::Handle::TLS_CTX 1293=item AnyEvent::Handle::TLS_CTX
1198 1294
1199This 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