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.50 by root, Fri May 30 21:38:46 2008 UTC vs.
Revision 1.69 by root, Sun Jun 15 21:44:56 2008 UTC

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 = 4.1; 19our $VERSION = 4.151;
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>, C<ETIMEDOUT> or C<EBADMSG>). 101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 102
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100
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
129This callback is useful when you don't want to put all of your write data
130into the queue at once, for example, when you want to write the contents
131of some file to the socket you might not want to read the whole file into
132memory and push it into the queue, but instead only read more data from
133the file when the write queue becomes empty.
124 134
125=item timeout => $fractional_seconds 135=item timeout => $fractional_seconds
126 136
127If non-zero, then this enables an "inactivity" timeout: whenever this many 137If non-zero, then this enables an "inactivity" timeout: whenever this many
128seconds pass without a successful read or write on the underlying file 138seconds pass without a successful read or write on the underlying file
163 173
164Sets the amount of bytes (default: C<0>) that make up an "empty" write 174Sets the amount of bytes (default: C<0>) that make up an "empty" write
165buffer: If the write reaches this size or gets even samller it is 175buffer: If the write reaches this size or gets even samller it is
166considered empty. 176considered empty.
167 177
178=item linger => <seconds>
179
180If non-zero (default: C<3600>), then the destructor of the
181AnyEvent::Handle object will check wether there is still outstanding write
182data and will install a watcher that will write out this data. No errors
183will be reported (this mostly matches how the operating system treats
184outstanding data at socket close time).
185
186This will not work for partial TLS data that could not yet been
187encoded. This data will be lost.
188
168=item tls => "accept" | "connect" | Net::SSLeay::SSL object 189=item tls => "accept" | "connect" | Net::SSLeay::SSL object
169 190
170When this parameter is given, it enables TLS (SSL) mode, that means it 191When this parameter is given, it enables TLS (SSL) mode, that means it
171will start making tls handshake and will transparently encrypt/decrypt 192will start making tls handshake and will transparently encrypt/decrypt
172data. 193data.
222 if ($self->{tls}) { 243 if ($self->{tls}) {
223 require Net::SSLeay; 244 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 245 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 } 246 }
226 247
227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
228# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
229# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
231
232 $self->{_activity} = AnyEvent->now; 248 $self->{_activity} = AnyEvent->now;
233 $self->_timeout; 249 $self->_timeout;
234 250
251 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
252
235 $self->start_read; 253 $self->start_read
254 if $self->{on_read};
236 255
237 $self 256 $self
238} 257}
239 258
240sub _shutdown { 259sub _shutdown {
242 261
243 delete $self->{_tw}; 262 delete $self->{_tw};
244 delete $self->{_rw}; 263 delete $self->{_rw};
245 delete $self->{_ww}; 264 delete $self->{_ww};
246 delete $self->{fh}; 265 delete $self->{fh};
247}
248 266
267 $self->stoptls;
268}
269
249sub error { 270sub _error {
250 my ($self) = @_; 271 my ($self, $errno, $fatal) = @_;
251 272
252 {
253 local $!;
254 $self->_shutdown; 273 $self->_shutdown
255 } 274 if $fatal;
256 275
257 $self->{on_error}($self) 276 $! = $errno;
277
258 if $self->{on_error}; 278 if ($self->{on_error}) {
259 279 $self->{on_error}($self, $fatal);
280 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 281 Carp::croak "AnyEvent::Handle uncaught error: $!";
282 }
261} 283}
262 284
263=item $fh = $handle->fh 285=item $fh = $handle->fh
264 286
265This method returns the file handle of the L<AnyEvent::Handle> object. 287This method returns the file handle of the L<AnyEvent::Handle> object.
331 $self->{_activity} = $NOW; 353 $self->{_activity} = $NOW;
332 354
333 if ($self->{on_timeout}) { 355 if ($self->{on_timeout}) {
334 $self->{on_timeout}($self); 356 $self->{on_timeout}($self);
335 } else { 357 } else {
336 $! = Errno::ETIMEDOUT; 358 $self->_error (&Errno::ETIMEDOUT);
337 $self->error;
338 } 359 }
339 360
340 # callbakx could have changed timeout value, optimise 361 # callback could have changed timeout value, optimise
341 return unless $self->{timeout}; 362 return unless $self->{timeout};
342 363
343 # calculate new after 364 # calculate new after
344 $after = $self->{timeout}; 365 $after = $self->{timeout};
345 } 366 }
346 367
347 Scalar::Util::weaken $self; 368 Scalar::Util::weaken $self;
369 return unless $self; # ->error could have destroyed $self
348 370
349 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 371 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
350 delete $self->{_tw}; 372 delete $self->{_tw};
351 $self->_timeout; 373 $self->_timeout;
352 }); 374 });
415 if $self->{low_water_mark} >= length $self->{wbuf} 437 if $self->{low_water_mark} >= length $self->{wbuf}
416 && $self->{on_drain}; 438 && $self->{on_drain};
417 439
418 delete $self->{_ww} unless length $self->{wbuf}; 440 delete $self->{_ww} unless length $self->{wbuf};
419 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 441 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
420 $self->error; 442 $self->_error ($!, 1);
421 } 443 }
422 }; 444 };
423 445
424 # try to write data immediately 446 # try to write data immediately
425 $cb->(); 447 $cb->();
454 } 476 }
455} 477}
456 478
457=item $handle->push_write (type => @args) 479=item $handle->push_write (type => @args)
458 480
459=item $handle->unshift_write (type => @args)
460
461Instead of formatting your data yourself, you can also let this module do 481Instead of formatting your data yourself, you can also let this module do
462the job by specifying a type and type-specific arguments. 482the job by specifying a type and type-specific arguments.
463 483
464Predefined types are (if you have ideas for additional types, feel free to 484Predefined types are (if you have ideas for additional types, feel free to
465drop by and tell us): 485drop by and tell us):
469=item netstring => $string 489=item netstring => $string
470 490
471Formats the given value as netstring 491Formats the given value as netstring
472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 492(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
473 493
474=back
475
476=cut 494=cut
477 495
478register_write_type netstring => sub { 496register_write_type netstring => sub {
479 my ($self, $string) = @_; 497 my ($self, $string) = @_;
480 498
481 sprintf "%d:%s,", (length $string), $string 499 sprintf "%d:%s,", (length $string), $string
500};
501
502=item packstring => $format, $data
503
504An octet string prefixed with an encoded length. The encoding C<$format>
505uses the same format as a Perl C<pack> format, but must specify a single
506integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
507optional C<!>, C<< < >> or C<< > >> modifier).
508
509=cut
510
511register_write_type packstring => sub {
512 my ($self, $format, $string) = @_;
513
514 pack "$format/a*", $string
482}; 515};
483 516
484=item json => $array_or_hashref 517=item json => $array_or_hashref
485 518
486Encodes the given hash or array reference into a JSON object. Unless you 519Encodes the given hash or array reference into a JSON object. Unless you
520 553
521 $self->{json} ? $self->{json}->encode ($ref) 554 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref) 555 : JSON::encode_json ($ref)
523}; 556};
524 557
558=item storable => $reference
559
560Freezes the given reference using L<Storable> and writes it to the
561handle. Uses the C<nfreeze> format.
562
563=cut
564
565register_write_type storable => sub {
566 my ($self, $ref) = @_;
567
568 require Storable;
569
570 pack "w/a*", Storable::nfreeze ($ref)
571};
572
573=back
574
525=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 575=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
526 576
527This function (not method) lets you add your own types to C<push_write>. 577This function (not method) lets you add your own types to C<push_write>.
528Whenever the given C<type> is used, C<push_write> will invoke the code 578Whenever the given C<type> is used, C<push_write> will invoke the code
529reference with the handle object and the remaining arguments. 579reference with the handle object and the remaining arguments.
549ways, the "simple" way, using only C<on_read> and the "complex" way, using 599ways, the "simple" way, using only C<on_read> and the "complex" way, using
550a queue. 600a queue.
551 601
552In the simple case, you just install an C<on_read> callback and whenever 602In the simple case, you just install an C<on_read> callback and whenever
553new data arrives, it will be called. You can then remove some data (if 603new data arrives, it will be called. You can then remove some data (if
554enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 604enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
555or not. 605leave the data there if you want to accumulate more (e.g. when only a
606partial message has been received so far).
556 607
557In the more complex case, you want to queue multiple callbacks. In this 608In the more complex case, you want to queue multiple callbacks. In this
558case, AnyEvent::Handle will call the first queued callback each time new 609case, AnyEvent::Handle will call the first queued callback each time new
559data arrives and removes it when it has done its job (see C<push_read>, 610data arrives (also the first time it is queued) and removes it when it has
560below). 611done its job (see C<push_read>, below).
561 612
562This way you can, for example, push three line-reads, followed by reading 613This way you can, for example, push three line-reads, followed by reading
563a chunk of data, and AnyEvent::Handle will execute them in order. 614a chunk of data, and AnyEvent::Handle will execute them in order.
564 615
565Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 616Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
566the specified number of bytes which give an XML datagram. 617the specified number of bytes which give an XML datagram.
567 618
568 # in the default state, expect some header bytes 619 # in the default state, expect some header bytes
569 $handle->on_read (sub { 620 $handle->on_read (sub {
570 # some data is here, now queue the length-header-read (4 octets) 621 # some data is here, now queue the length-header-read (4 octets)
571 shift->unshift_read_chunk (4, sub { 622 shift->unshift_read (chunk => 4, sub {
572 # header arrived, decode 623 # header arrived, decode
573 my $len = unpack "N", $_[1]; 624 my $len = unpack "N", $_[1];
574 625
575 # now read the payload 626 # now read the payload
576 shift->unshift_read_chunk ($len, sub { 627 shift->unshift_read (chunk => $len, sub {
577 my $xml = $_[1]; 628 my $xml = $_[1];
578 # handle xml 629 # handle xml
579 }); 630 });
580 }); 631 });
581 }); 632 });
582 633
583Example 2: Implement a client for a protocol that replies either with 634Example 2: Implement a client for a protocol that replies either with "OK"
584"OK" and another line or "ERROR" for one request, and 64 bytes for the 635and another line or "ERROR" for the first request that is sent, and 64
585second request. Due tot he availability of a full queue, we can just 636bytes for the second request. Due to the availability of a queue, we can
586pipeline sending both requests and manipulate the queue as necessary in 637just pipeline sending both requests and manipulate the queue as necessary
587the callbacks: 638in the callbacks.
588 639
589 # request one 640When the first callback is called and sees an "OK" response, it will
641C<unshift> another line-read. This line-read will be queued I<before> the
64264-byte chunk callback.
643
644 # request one, returns either "OK + extra line" or "ERROR"
590 $handle->push_write ("request 1\015\012"); 645 $handle->push_write ("request 1\015\012");
591 646
592 # we expect "ERROR" or "OK" as response, so push a line read 647 # we expect "ERROR" or "OK" as response, so push a line read
593 $handle->push_read_line (sub { 648 $handle->push_read (line => sub {
594 # if we got an "OK", we have to _prepend_ another line, 649 # if we got an "OK", we have to _prepend_ another line,
595 # so it will be read before the second request reads its 64 bytes 650 # so it will be read before the second request reads its 64 bytes
596 # which are already in the queue when this callback is called 651 # which are already in the queue when this callback is called
597 # we don't do this in case we got an error 652 # we don't do this in case we got an error
598 if ($_[1] eq "OK") { 653 if ($_[1] eq "OK") {
599 $_[0]->unshift_read_line (sub { 654 $_[0]->unshift_read (line => sub {
600 my $response = $_[1]; 655 my $response = $_[1];
601 ... 656 ...
602 }); 657 });
603 } 658 }
604 }); 659 });
605 660
606 # request two 661 # request two, simply returns 64 octets
607 $handle->push_write ("request 2\015\012"); 662 $handle->push_write ("request 2\015\012");
608 663
609 # simply read 64 bytes, always 664 # simply read 64 bytes, always
610 $handle->push_read_chunk (64, sub { 665 $handle->push_read (chunk => 64, sub {
611 my $response = $_[1]; 666 my $response = $_[1];
612 ... 667 ...
613 }); 668 });
614 669
615=over 4 670=over 4
616 671
617=cut 672=cut
618 673
619sub _drain_rbuf { 674sub _drain_rbuf {
620 my ($self) = @_; 675 my ($self) = @_;
676
677 local $self->{_in_drain} = 1;
621 678
622 if ( 679 if (
623 defined $self->{rbuf_max} 680 defined $self->{rbuf_max}
624 && $self->{rbuf_max} < length $self->{rbuf} 681 && $self->{rbuf_max} < length $self->{rbuf}
625 ) { 682 ) {
626 $! = &Errno::ENOSPC; 683 return $self->_error (&Errno::ENOSPC, 1);
627 $self->error;
628 } 684 }
629 685
630 return if $self->{in_drain}; 686 while () {
631 local $self->{in_drain} = 1;
632
633 while (my $len = length $self->{rbuf}) {
634 no strict 'refs'; 687 no strict 'refs';
688
689 my $len = length $self->{rbuf};
690
635 if (my $cb = shift @{ $self->{_queue} }) { 691 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) { 692 unless ($cb->($self)) {
637 if ($self->{_eof}) { 693 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming) 694 # no progress can be made (not enough data and no data forthcoming)
639 $! = &Errno::EPIPE; 695 $self->_error (&Errno::EPIPE, 1), last;
640 $self->error;
641 } 696 }
642 697
643 unshift @{ $self->{_queue} }, $cb; 698 unshift @{ $self->{_queue} }, $cb;
644 return; 699 last;
645 } 700 }
646 } elsif ($self->{on_read}) { 701 } elsif ($self->{on_read}) {
702 last unless $len;
703
647 $self->{on_read}($self); 704 $self->{on_read}($self);
648 705
649 if ( 706 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed 707 $len == length $self->{rbuf} # if no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty 708 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data 709 && $self->{on_read} # but we still have on_read
654 ) { 710 ) {
711 # no further data will arrive
655 # then no progress can be made 712 # so no progress can be made
656 $! = &Errno::EPIPE; 713 $self->_error (&Errno::EPIPE, 1), last
657 $self->error; 714 if $self->{_eof};
715
716 last; # more data might arrive
658 } 717 }
659 } else { 718 } else {
660 # read side becomes idle 719 # read side becomes idle
661 delete $self->{_rw}; 720 delete $self->{_rw};
662 return; 721 last;
663 } 722 }
664 } 723 }
665 724
666 $self->{on_eof}($self) 725 $self->{on_eof}($self)
667 if $self->{_eof} && $self->{on_eof}; 726 if $self->{_eof} && $self->{on_eof};
727
728 # may need to restart read watcher
729 unless ($self->{_rw}) {
730 $self->start_read
731 if $self->{on_read} || @{ $self->{_queue} };
732 }
668} 733}
669 734
670=item $handle->on_read ($cb) 735=item $handle->on_read ($cb)
671 736
672This replaces the currently set C<on_read> callback, or clears it (when 737This replaces the currently set C<on_read> callback, or clears it (when
677 742
678sub on_read { 743sub on_read {
679 my ($self, $cb) = @_; 744 my ($self, $cb) = @_;
680 745
681 $self->{on_read} = $cb; 746 $self->{on_read} = $cb;
747 $self->_drain_rbuf if $cb && !$self->{_in_drain};
682} 748}
683 749
684=item $handle->rbuf 750=item $handle->rbuf
685 751
686Returns the read buffer (as a modifiable lvalue). 752Returns the read buffer (as a modifiable lvalue).
735 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 801 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
736 ->($self, $cb, @_); 802 ->($self, $cb, @_);
737 } 803 }
738 804
739 push @{ $self->{_queue} }, $cb; 805 push @{ $self->{_queue} }, $cb;
740 $self->_drain_rbuf; 806 $self->_drain_rbuf unless $self->{_in_drain};
741} 807}
742 808
743sub unshift_read { 809sub unshift_read {
744 my $self = shift; 810 my $self = shift;
745 my $cb = pop; 811 my $cb = pop;
751 ->($self, $cb, @_); 817 ->($self, $cb, @_);
752 } 818 }
753 819
754 820
755 unshift @{ $self->{_queue} }, $cb; 821 unshift @{ $self->{_queue} }, $cb;
756 $self->_drain_rbuf; 822 $self->_drain_rbuf unless $self->{_in_drain};
757} 823}
758 824
759=item $handle->push_read (type => @args, $cb) 825=item $handle->push_read (type => @args, $cb)
760 826
761=item $handle->unshift_read (type => @args, $cb) 827=item $handle->unshift_read (type => @args, $cb)
846 912
847sub unshift_read_line { 913sub unshift_read_line {
848 my $self = shift; 914 my $self = shift;
849 $self->unshift_read (line => @_); 915 $self->unshift_read (line => @_);
850} 916}
851
852=item netstring => $cb->($handle, $string)
853
854A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
855
856Throws an error with C<$!> set to EBADMSG on format violations.
857
858=cut
859
860register_read_type netstring => sub {
861 my ($self, $cb) = @_;
862
863 sub {
864 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
865 if ($_[0]{rbuf} =~ /[^0-9]/) {
866 $! = &Errno::EBADMSG;
867 $self->error;
868 }
869 return;
870 }
871
872 my $len = $1;
873
874 $self->unshift_read (chunk => $len, sub {
875 my $string = $_[1];
876 $_[0]->unshift_read (chunk => 1, sub {
877 if ($_[1] eq ",") {
878 $cb->($_[0], $string);
879 } else {
880 $! = &Errno::EBADMSG;
881 $self->error;
882 }
883 });
884 });
885
886 1
887 }
888};
889 917
890=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 918=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
891 919
892Makes a regex match against the regex object C<$accept> and returns 920Makes a regex match against the regex object C<$accept> and returns
893everything up to and including the match. 921everything up to and including the match.
943 return 1; 971 return 1;
944 } 972 }
945 973
946 # reject 974 # reject
947 if ($reject && $$rbuf =~ $reject) { 975 if ($reject && $$rbuf =~ $reject) {
948 $! = &Errno::EBADMSG; 976 $self->_error (&Errno::EBADMSG);
949 $self->error;
950 } 977 }
951 978
952 # skip 979 # skip
953 if ($skip && $$rbuf =~ $skip) { 980 if ($skip && $$rbuf =~ $skip) {
954 $data .= substr $$rbuf, 0, $+[0], ""; 981 $data .= substr $$rbuf, 0, $+[0], "";
956 983
957 () 984 ()
958 } 985 }
959}; 986};
960 987
988=item netstring => $cb->($handle, $string)
989
990A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
991
992Throws an error with C<$!> set to EBADMSG on format violations.
993
994=cut
995
996register_read_type netstring => sub {
997 my ($self, $cb) = @_;
998
999 sub {
1000 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1001 if ($_[0]{rbuf} =~ /[^0-9]/) {
1002 $self->_error (&Errno::EBADMSG);
1003 }
1004 return;
1005 }
1006
1007 my $len = $1;
1008
1009 $self->unshift_read (chunk => $len, sub {
1010 my $string = $_[1];
1011 $_[0]->unshift_read (chunk => 1, sub {
1012 if ($_[1] eq ",") {
1013 $cb->($_[0], $string);
1014 } else {
1015 $self->_error (&Errno::EBADMSG);
1016 }
1017 });
1018 });
1019
1020 1
1021 }
1022};
1023
1024=item packstring => $format, $cb->($handle, $string)
1025
1026An octet string prefixed with an encoded length. The encoding C<$format>
1027uses the same format as a Perl C<pack> format, but must specify a single
1028integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1029optional C<!>, C<< < >> or C<< > >> modifier).
1030
1031DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1032
1033Example: read a block of data prefixed by its length in BER-encoded
1034format (very efficient).
1035
1036 $handle->push_read (packstring => "w", sub {
1037 my ($handle, $data) = @_;
1038 });
1039
1040=cut
1041
1042register_read_type packstring => sub {
1043 my ($self, $cb, $format) = @_;
1044
1045 sub {
1046 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1047 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1048 or return;
1049
1050 # remove prefix
1051 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1052
1053 # read rest
1054 $_[0]->unshift_read (chunk => $len, $cb);
1055
1056 1
1057 }
1058};
1059
961=item json => $cb->($handle, $hash_or_arrayref) 1060=item json => $cb->($handle, $hash_or_arrayref)
962 1061
963Reads a JSON object or array, decodes it and passes it to the callback. 1062Reads a JSON object or array, decodes it and passes it to the callback.
964 1063
965If a C<json> object was passed to the constructor, then that will be used 1064If a C<json> object was passed to the constructor, then that will be used
975the C<json> write type description, above, for an actual example. 1074the C<json> write type description, above, for an actual example.
976 1075
977=cut 1076=cut
978 1077
979register_read_type json => sub { 1078register_read_type json => sub {
980 my ($self, $cb, $accept, $reject, $skip) = @_; 1079 my ($self, $cb) = @_;
981 1080
982 require JSON; 1081 require JSON;
983 1082
984 my $data; 1083 my $data;
985 my $rbuf = \$self->{rbuf}; 1084 my $rbuf = \$self->{rbuf};
1000 () 1099 ()
1001 } 1100 }
1002 } 1101 }
1003}; 1102};
1004 1103
1104=item storable => $cb->($handle, $ref)
1105
1106Deserialises a L<Storable> frozen representation as written by the
1107C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1108data).
1109
1110Raises C<EBADMSG> error if the data could not be decoded.
1111
1112=cut
1113
1114register_read_type storable => sub {
1115 my ($self, $cb) = @_;
1116
1117 require Storable;
1118
1119 sub {
1120 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1121 defined (my $len = eval { unpack "w", $_[0]->{rbuf} })
1122 or return;
1123
1124 # remove prefix
1125 substr $_[0]->{rbuf}, 0, (length pack "w", $len), "";
1126
1127 # read rest
1128 $_[0]->unshift_read (chunk => $len, sub {
1129 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1130 $cb->($_[0], $ref);
1131 } else {
1132 $self->_error (&Errno::EBADMSG);
1133 }
1134 });
1135 }
1136};
1137
1005=back 1138=back
1006 1139
1007=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1140=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1008 1141
1009This function (not method) lets you add your own types to C<push_read>. 1142This function (not method) lets you add your own types to C<push_read>.
1027=item $handle->stop_read 1160=item $handle->stop_read
1028 1161
1029=item $handle->start_read 1162=item $handle->start_read
1030 1163
1031In rare cases you actually do not want to read anything from the 1164In rare cases you actually do not want to read anything from the
1032socket. In this case you can call C<stop_read>. Neither C<on_read> no 1165socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1033any queued callbacks will be executed then. To start reading again, call 1166any queued callbacks will be executed then. To start reading again, call
1034C<start_read>. 1167C<start_read>.
1168
1169Note that AnyEvent::Handle will automatically C<start_read> for you when
1170you change the C<on_read> callback or push/unshift a read callback, and it
1171will automatically C<stop_read> for you when neither C<on_read> is set nor
1172there are any read requests in the queue.
1035 1173
1036=cut 1174=cut
1037 1175
1038sub stop_read { 1176sub stop_read {
1039 my ($self) = @_; 1177 my ($self) = @_;
1054 if ($len > 0) { 1192 if ($len > 0) {
1055 $self->{_activity} = AnyEvent->now; 1193 $self->{_activity} = AnyEvent->now;
1056 1194
1057 $self->{filter_r} 1195 $self->{filter_r}
1058 ? $self->{filter_r}($self, $rbuf) 1196 ? $self->{filter_r}($self, $rbuf)
1059 : $self->_drain_rbuf; 1197 : $self->{_in_drain} || $self->_drain_rbuf;
1060 1198
1061 } elsif (defined $len) { 1199 } elsif (defined $len) {
1062 delete $self->{_rw}; 1200 delete $self->{_rw};
1063 $self->{_eof} = 1; 1201 $self->{_eof} = 1;
1064 $self->_drain_rbuf; 1202 $self->_drain_rbuf unless $self->{_in_drain};
1065 1203
1066 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1204 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1067 return $self->error; 1205 return $self->_error ($!, 1);
1068 } 1206 }
1069 }); 1207 });
1070 } 1208 }
1071} 1209}
1072 1210
1073sub _dotls { 1211sub _dotls {
1074 my ($self) = @_; 1212 my ($self) = @_;
1213
1214 my $buf;
1075 1215
1076 if (length $self->{_tls_wbuf}) { 1216 if (length $self->{_tls_wbuf}) {
1077 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1217 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1078 substr $self->{_tls_wbuf}, 0, $len, ""; 1218 substr $self->{_tls_wbuf}, 0, $len, "";
1079 } 1219 }
1080 } 1220 }
1081 1221
1082 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1222 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1083 $self->{wbuf} .= $buf; 1223 $self->{wbuf} .= $buf;
1084 $self->_drain_wbuf; 1224 $self->_drain_wbuf;
1085 } 1225 }
1086 1226
1087 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1227 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1228 if (length $buf) {
1088 $self->{rbuf} .= $buf; 1229 $self->{rbuf} .= $buf;
1089 $self->_drain_rbuf; 1230 $self->_drain_rbuf unless $self->{_in_drain};
1231 } else {
1232 # let's treat SSL-eof as we treat normal EOF
1233 $self->{_eof} = 1;
1234 $self->_shutdown;
1235 return;
1236 }
1090 } 1237 }
1091 1238
1092 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1239 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1093 1240
1094 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1241 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1095 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1242 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1096 $self->error; 1243 return $self->_error ($!, 1);
1097 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1244 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1098 $! = &Errno::EIO; 1245 return $self->_error (&Errno::EIO, 1);
1099 $self->error;
1100 } 1246 }
1101 1247
1102 # all others are fine for our purposes 1248 # all others are fine for our purposes
1103 } 1249 }
1104} 1250}
1119call and can be used or changed to your liking. Note that the handshake 1265call and can be used or changed to your liking. Note that the handshake
1120might have already started when this function returns. 1266might have already started when this function returns.
1121 1267
1122=cut 1268=cut
1123 1269
1124# TODO: maybe document...
1125sub starttls { 1270sub starttls {
1126 my ($self, $ssl, $ctx) = @_; 1271 my ($self, $ssl, $ctx) = @_;
1127 1272
1128 $self->stoptls; 1273 $self->stoptls;
1129 1274
1182 1327
1183sub DESTROY { 1328sub DESTROY {
1184 my $self = shift; 1329 my $self = shift;
1185 1330
1186 $self->stoptls; 1331 $self->stoptls;
1332
1333 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1334
1335 if ($linger && length $self->{wbuf}) {
1336 my $fh = delete $self->{fh};
1337 my $wbuf = delete $self->{wbuf};
1338
1339 my @linger;
1340
1341 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1342 my $len = syswrite $fh, $wbuf, length $wbuf;
1343
1344 if ($len > 0) {
1345 substr $wbuf, 0, $len, "";
1346 } else {
1347 @linger = (); # end
1348 }
1349 });
1350 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1351 @linger = ();
1352 });
1353 }
1187} 1354}
1188 1355
1189=item AnyEvent::Handle::TLS_CTX 1356=item AnyEvent::Handle::TLS_CTX
1190 1357
1191This function creates and returns the Net::SSLeay::CTX object used by 1358This function creates and returns the Net::SSLeay::CTX object used by

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines