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.55 by root, Tue Jun 3 16:15:30 2008 UTC vs.
Revision 1.84 by root, Thu Aug 21 19:13:05 2008 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict qw(subs vars);
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
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.12; 19our $VERSION = 4.232;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
49 49
50This module is a helper module to make it easier to do event-based I/O on 50This module is a helper module to make it easier to do event-based I/O on
51filehandles. For utility functions for doing non-blocking connects and accepts 51filehandles. For utility functions for doing non-blocking connects and accepts
52on sockets see L<AnyEvent::Util>. 52on sockets see L<AnyEvent::Util>.
53 53
54The L<AnyEvent::Intro> tutorial contains some well-documented
55AnyEvent::Handle examples.
56
54In the following, when the documentation refers to of "bytes" then this 57In the following, when the documentation refers to of "bytes" then this
55means characters. As sysread and syswrite are used for all I/O, their 58means characters. As sysread and syswrite are used for all I/O, their
56treatment of characters applies to this module as well. 59treatment of characters applies to this module as well.
57 60
58All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
70 73
71=item fh => $filehandle [MANDATORY] 74=item fh => $filehandle [MANDATORY]
72 75
73The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
74 77
75NOTE: The filehandle will be set to non-blocking (using 78NOTE: The filehandle will be set to non-blocking mode (using
76AnyEvent::Util::fh_nonblocking). 79C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in
80that mode.
77 81
78=item on_eof => $cb->($handle) 82=item on_eof => $cb->($handle)
79 83
80Set the callback to be called when an end-of-file condition is detcted, 84Set the callback to be called when an end-of-file condition is detected,
81i.e. in the case of a socket, when the other side has closed the 85i.e. in the case of a socket, when the other side has closed the
82connection cleanly. 86connection cleanly.
83 87
88For sockets, this just means that the other side has stopped sending data,
89you can still try to write data, and, in fact, one can return from the eof
90callback and continue writing data, as only the read part has been shut
91down.
92
84While not mandatory, it is highly recommended to set an eof callback, 93While not mandatory, it is I<highly> recommended to set an eof callback,
85otherwise you might end up with a closed socket while you are still 94otherwise you might end up with a closed socket while you are still
86waiting for data. 95waiting for data.
96
97If an EOF condition has been detected but no C<on_eof> callback has been
98set, then a fatal error will be raised with C<$!> set to <0>.
87 99
88=item on_error => $cb->($handle, $fatal) 100=item on_error => $cb->($handle, $fatal)
89 101
90This is the error callback, which is called when, well, some error 102This is the error callback, which is called when, well, some error
91occured, such as not being able to resolve the hostname, failure to 103occured, such as not being able to resolve the hostname, failure to
92connect or a read error. 104connect or a read error.
93 105
94Some errors are fatal (which is indicated by C<$fatal> being true). On 106Some errors are fatal (which is indicated by C<$fatal> being true). On
95fatal errors the handle object will be shut down and will not be 107fatal errors the handle object will be shut down and will not be usable
108(but you are free to look at the current C< ->rbuf >). Examples of fatal
109errors are an EOF condition with active (but unsatisifable) read watchers
110(C<EPIPE>) or I/O errors.
111
96usable. Non-fatal errors can be retried by simply returning, but it is 112Non-fatal errors can be retried by simply returning, but it is recommended
97recommended to simply ignore this parameter and instead abondon the handle 113to simply ignore this parameter and instead abondon the handle object
98object when this callback is invoked. 114when this callback is invoked. Examples of non-fatal errors are timeouts
115C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
99 116
100On callback entrance, the value of C<$!> contains the operating system 117On callback entrance, the value of C<$!> contains the operating system
101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 118error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
102 119
103While not mandatory, it is I<highly> recommended to set this callback, as 120While not mandatory, it is I<highly> recommended to set this callback, as
105C<croak>. 122C<croak>.
106 123
107=item on_read => $cb->($handle) 124=item on_read => $cb->($handle)
108 125
109This sets the default read callback, which is called when data arrives 126This sets the default read callback, which is called when data arrives
110and no read request is in the queue. 127and no read request is in the queue (unlike read queue callbacks, this
128callback will only be called when at least one octet of data is in the
129read buffer).
111 130
112To access (and remove data from) the read buffer, use the C<< ->rbuf >> 131To access (and remove data from) the read buffer, use the C<< ->rbuf >>
113method or access the C<$handle->{rbuf}> member directly. 132method or access the C<$handle->{rbuf}> member directly.
114 133
115When an EOF condition is detected then AnyEvent::Handle will first try to 134When an EOF condition is detected then AnyEvent::Handle will first try to
121 140
122This sets the callback that is called when the write buffer becomes empty 141This sets the callback that is called when the write buffer becomes empty
123(or when the callback is set and the buffer is empty already). 142(or when the callback is set and the buffer is empty already).
124 143
125To append to the write buffer, use the C<< ->push_write >> method. 144To append to the write buffer, use the C<< ->push_write >> method.
145
146This callback is useful when you don't want to put all of your write data
147into the queue at once, for example, when you want to write the contents
148of some file to the socket you might not want to read the whole file into
149memory and push it into the queue, but instead only read more data from
150the file when the write queue becomes empty.
126 151
127=item timeout => $fractional_seconds 152=item timeout => $fractional_seconds
128 153
129If non-zero, then this enables an "inactivity" timeout: whenever this many 154If non-zero, then this enables an "inactivity" timeout: whenever this many
130seconds pass without a successful read or write on the underlying file 155seconds pass without a successful read or write on the underlying file
154be configured to accept only so-and-so much data that it cannot act on 179be configured to accept only so-and-so much data that it cannot act on
155(for example, when expecting a line, an attacker could send an unlimited 180(for example, when expecting a line, an attacker could send an unlimited
156amount of data without a callback ever being called as long as the line 181amount of data without a callback ever being called as long as the line
157isn't finished). 182isn't finished).
158 183
184=item autocork => <boolean>
185
186When disabled (the default), then C<push_write> will try to immediately
187write the data to the handle if possible. This avoids having to register
188a write watcher and wait for the next event loop iteration, but can be
189inefficient if you write multiple small chunks (this disadvantage is
190usually avoided by your kernel's nagle algorithm, see C<low_delay>).
191
192When enabled, then writes will always be queued till the next event loop
193iteration. This is efficient when you do many small writes per iteration,
194but less efficient when you do a single write only.
195
196=item no_delay => <boolean>
197
198When doing small writes on sockets, your operating system kernel might
199wait a bit for more data before actually sending it out. This is called
200the Nagle algorithm, and usually it is beneficial.
201
202In some situations you want as low a delay as possible, which cna be
203accomplishd by setting this option to true.
204
205The default is your opertaing system's default behaviour, this option
206explicitly enables or disables it, if possible.
207
159=item read_size => <bytes> 208=item read_size => <bytes>
160 209
161The default read block size (the amount of bytes this module will try to read 210The default read block size (the amount of bytes this module will try to read
162during each (loop iteration). Default: C<8192>. 211during each (loop iteration). Default: C<8192>.
163 212
164=item low_water_mark => <bytes> 213=item low_water_mark => <bytes>
165 214
166Sets the amount of bytes (default: C<0>) that make up an "empty" write 215Sets the amount of bytes (default: C<0>) that make up an "empty" write
167buffer: If the write reaches this size or gets even samller it is 216buffer: If the write reaches this size or gets even samller it is
168considered empty. 217considered empty.
218
219=item linger => <seconds>
220
221If non-zero (default: C<3600>), then the destructor of the
222AnyEvent::Handle object will check wether there is still outstanding write
223data and will install a watcher that will write out this data. No errors
224will be reported (this mostly matches how the operating system treats
225outstanding data at socket close time).
226
227This will not work for partial TLS data that could not yet been
228encoded. This data will be lost.
169 229
170=item tls => "accept" | "connect" | Net::SSLeay::SSL object 230=item tls => "accept" | "connect" | Net::SSLeay::SSL object
171 231
172When this parameter is given, it enables TLS (SSL) mode, that means it 232When this parameter is given, it enables TLS (SSL) mode, that means it
173will start making tls handshake and will transparently encrypt/decrypt 233will start making tls handshake and will transparently encrypt/decrypt
182You can also provide your own TLS connection object, but you have 242You can also provide your own TLS connection object, but you have
183to make sure that you call either C<Net::SSLeay::set_connect_state> 243to make sure that you call either C<Net::SSLeay::set_connect_state>
184or C<Net::SSLeay::set_accept_state> on it before you pass it to 244or C<Net::SSLeay::set_accept_state> on it before you pass it to
185AnyEvent::Handle. 245AnyEvent::Handle.
186 246
187See the C<starttls> method if you need to start TLs negotiation later. 247See the C<starttls> method if you need to start TLS negotiation later.
188 248
189=item tls_ctx => $ssl_ctx 249=item tls_ctx => $ssl_ctx
190 250
191Use the given Net::SSLeay::CTX object to create the new TLS connection 251Use the given Net::SSLeay::CTX object to create the new TLS connection
192(unless a connection object was specified directly). If this parameter is 252(unless a connection object was specified directly). If this parameter is
224 if ($self->{tls}) { 284 if ($self->{tls}) {
225 require Net::SSLeay; 285 require Net::SSLeay;
226 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 286 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
227 } 287 }
228 288
229# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
230# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
231# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
232 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
233
234 $self->{_activity} = AnyEvent->now; 289 $self->{_activity} = AnyEvent->now;
235 $self->_timeout; 290 $self->_timeout;
236 291
292 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
293 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
294
237 $self->start_read; 295 $self->start_read
296 if $self->{on_read};
238 297
239 $self 298 $self
240} 299}
241 300
242sub _shutdown { 301sub _shutdown {
246 delete $self->{_rw}; 305 delete $self->{_rw};
247 delete $self->{_ww}; 306 delete $self->{_ww};
248 delete $self->{fh}; 307 delete $self->{fh};
249 308
250 $self->stoptls; 309 $self->stoptls;
310
311 delete $self->{on_read};
312 delete $self->{_queue};
251} 313}
252 314
253sub _error { 315sub _error {
254 my ($self, $errno, $fatal) = @_; 316 my ($self, $errno, $fatal) = @_;
255 317
301 363
302=cut 364=cut
303 365
304sub on_timeout { 366sub on_timeout {
305 $_[0]{on_timeout} = $_[1]; 367 $_[0]{on_timeout} = $_[1];
368}
369
370=item $handle->autocork ($boolean)
371
372Enables or disables the current autocork behaviour (see C<autocork>
373constructor argument).
374
375=cut
376
377=item $handle->no_delay ($boolean)
378
379Enables or disables the C<no_delay> setting (see constructor argument of
380the same name for details).
381
382=cut
383
384sub no_delay {
385 $_[0]{no_delay} = $_[1];
386
387 eval {
388 local $SIG{__DIE__};
389 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
390 };
306} 391}
307 392
308############################################################################# 393#############################################################################
309 394
310=item $handle->timeout ($seconds) 395=item $handle->timeout ($seconds)
339 $self->{on_timeout}($self); 424 $self->{on_timeout}($self);
340 } else { 425 } else {
341 $self->_error (&Errno::ETIMEDOUT); 426 $self->_error (&Errno::ETIMEDOUT);
342 } 427 }
343 428
344 # callbakx could have changed timeout value, optimise 429 # callback could have changed timeout value, optimise
345 return unless $self->{timeout}; 430 return unless $self->{timeout};
346 431
347 # calculate new after 432 # calculate new after
348 $after = $self->{timeout}; 433 $after = $self->{timeout};
349 } 434 }
350 435
351 Scalar::Util::weaken $self; 436 Scalar::Util::weaken $self;
437 return unless $self; # ->error could have destroyed $self
352 438
353 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 439 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
354 delete $self->{_tw}; 440 delete $self->{_tw};
355 $self->_timeout; 441 $self->_timeout;
356 }); 442 });
424 $self->_error ($!, 1); 510 $self->_error ($!, 1);
425 } 511 }
426 }; 512 };
427 513
428 # try to write data immediately 514 # try to write data immediately
429 $cb->(); 515 $cb->() unless $self->{autocork};
430 516
431 # if still data left in wbuf, we need to poll 517 # if still data left in wbuf, we need to poll
432 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 518 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
433 if length $self->{wbuf}; 519 if length $self->{wbuf};
434 }; 520 };
479 my ($self, $string) = @_; 565 my ($self, $string) = @_;
480 566
481 sprintf "%d:%s,", (length $string), $string 567 sprintf "%d:%s,", (length $string), $string
482}; 568};
483 569
570=item packstring => $format, $data
571
572An octet string prefixed with an encoded length. The encoding C<$format>
573uses the same format as a Perl C<pack> format, but must specify a single
574integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
575optional C<!>, C<< < >> or C<< > >> modifier).
576
577=cut
578
579register_write_type packstring => sub {
580 my ($self, $format, $string) = @_;
581
582 pack "$format/a*", $string
583};
584
484=item json => $array_or_hashref 585=item json => $array_or_hashref
485 586
486Encodes the given hash or array reference into a JSON object. Unless you 587Encodes the given hash or array reference into a JSON object. Unless you
487provide your own JSON object, this means it will be encoded to JSON text 588provide your own JSON object, this means it will be encoded to JSON text
488in UTF-8. 589in UTF-8.
520 621
521 $self->{json} ? $self->{json}->encode ($ref) 622 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref) 623 : JSON::encode_json ($ref)
523}; 624};
524 625
626=item storable => $reference
627
628Freezes the given reference using L<Storable> and writes it to the
629handle. Uses the C<nfreeze> format.
630
631=cut
632
633register_write_type storable => sub {
634 my ($self, $ref) = @_;
635
636 require Storable;
637
638 pack "w/a*", Storable::nfreeze ($ref)
639};
640
525=back 641=back
526 642
527=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 643=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
528 644
529This function (not method) lets you add your own types to C<push_write>. 645This function (not method) lets you add your own types to C<push_write>.
551ways, the "simple" way, using only C<on_read> and the "complex" way, using 667ways, the "simple" way, using only C<on_read> and the "complex" way, using
552a queue. 668a queue.
553 669
554In the simple case, you just install an C<on_read> callback and whenever 670In the simple case, you just install an C<on_read> callback and whenever
555new data arrives, it will be called. You can then remove some data (if 671new data arrives, it will be called. You can then remove some data (if
556enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 672enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
557or not. 673leave the data there if you want to accumulate more (e.g. when only a
674partial message has been received so far).
558 675
559In the more complex case, you want to queue multiple callbacks. In this 676In the more complex case, you want to queue multiple callbacks. In this
560case, AnyEvent::Handle will call the first queued callback each time new 677case, AnyEvent::Handle will call the first queued callback each time new
561data arrives and removes it when it has done its job (see C<push_read>, 678data arrives (also the first time it is queued) and removes it when it has
562below). 679done its job (see C<push_read>, below).
563 680
564This way you can, for example, push three line-reads, followed by reading 681This way you can, for example, push three line-reads, followed by reading
565a chunk of data, and AnyEvent::Handle will execute them in order. 682a chunk of data, and AnyEvent::Handle will execute them in order.
566 683
567Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 684Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
580 # handle xml 697 # handle xml
581 }); 698 });
582 }); 699 });
583 }); 700 });
584 701
585Example 2: Implement a client for a protocol that replies either with 702Example 2: Implement a client for a protocol that replies either with "OK"
586"OK" and another line or "ERROR" for one request, and 64 bytes for the 703and another line or "ERROR" for the first request that is sent, and 64
587second request. Due tot he availability of a full queue, we can just 704bytes for the second request. Due to the availability of a queue, we can
588pipeline sending both requests and manipulate the queue as necessary in 705just pipeline sending both requests and manipulate the queue as necessary
589the callbacks: 706in the callbacks.
590 707
591 # request one 708When the first callback is called and sees an "OK" response, it will
709C<unshift> another line-read. This line-read will be queued I<before> the
71064-byte chunk callback.
711
712 # request one, returns either "OK + extra line" or "ERROR"
592 $handle->push_write ("request 1\015\012"); 713 $handle->push_write ("request 1\015\012");
593 714
594 # we expect "ERROR" or "OK" as response, so push a line read 715 # we expect "ERROR" or "OK" as response, so push a line read
595 $handle->push_read (line => sub { 716 $handle->push_read (line => sub {
596 # if we got an "OK", we have to _prepend_ another line, 717 # if we got an "OK", we have to _prepend_ another line,
603 ... 724 ...
604 }); 725 });
605 } 726 }
606 }); 727 });
607 728
608 # request two 729 # request two, simply returns 64 octets
609 $handle->push_write ("request 2\015\012"); 730 $handle->push_write ("request 2\015\012");
610 731
611 # simply read 64 bytes, always 732 # simply read 64 bytes, always
612 $handle->push_read (chunk => 64, sub { 733 $handle->push_read (chunk => 64, sub {
613 my $response = $_[1]; 734 my $response = $_[1];
619=cut 740=cut
620 741
621sub _drain_rbuf { 742sub _drain_rbuf {
622 my ($self) = @_; 743 my ($self) = @_;
623 744
745 local $self->{_in_drain} = 1;
746
624 if ( 747 if (
625 defined $self->{rbuf_max} 748 defined $self->{rbuf_max}
626 && $self->{rbuf_max} < length $self->{rbuf} 749 && $self->{rbuf_max} < length $self->{rbuf}
627 ) { 750 ) {
628 return $self->_error (&Errno::ENOSPC, 1); 751 $self->_error (&Errno::ENOSPC, 1), return;
629 } 752 }
630 753
631 return if $self->{in_drain}; 754 while () {
632 local $self->{in_drain} = 1;
633
634 while (my $len = length $self->{rbuf}) { 755 my $len = length $self->{rbuf};
635 no strict 'refs'; 756
636 if (my $cb = shift @{ $self->{_queue} }) { 757 if (my $cb = shift @{ $self->{_queue} }) {
637 unless ($cb->($self)) { 758 unless ($cb->($self)) {
638 if ($self->{_eof}) { 759 if ($self->{_eof}) {
639 # no progress can be made (not enough data and no data forthcoming) 760 # no progress can be made (not enough data and no data forthcoming)
640 return $self->_error (&Errno::EPIPE, 1); 761 $self->_error (&Errno::EPIPE, 1), return;
641 } 762 }
642 763
643 unshift @{ $self->{_queue} }, $cb; 764 unshift @{ $self->{_queue} }, $cb;
644 last; 765 last;
645 } 766 }
646 } elsif ($self->{on_read}) { 767 } elsif ($self->{on_read}) {
768 last unless $len;
769
647 $self->{on_read}($self); 770 $self->{on_read}($self);
648 771
649 if ( 772 if (
650 $len == length $self->{rbuf} # if no data has been consumed 773 $len == length $self->{rbuf} # if no data has been consumed
651 && !@{ $self->{_queue} } # and the queue is still empty 774 && !@{ $self->{_queue} } # and the queue is still empty
652 && $self->{on_read} # but we still have on_read 775 && $self->{on_read} # but we still have on_read
653 ) { 776 ) {
654 # no further data will arrive 777 # no further data will arrive
655 # so no progress can be made 778 # so no progress can be made
656 return $self->_error (&Errno::EPIPE, 1) 779 $self->_error (&Errno::EPIPE, 1), return
657 if $self->{_eof}; 780 if $self->{_eof};
658 781
659 last; # more data might arrive 782 last; # more data might arrive
660 } 783 }
661 } else { 784 } else {
663 delete $self->{_rw}; 786 delete $self->{_rw};
664 last; 787 last;
665 } 788 }
666 } 789 }
667 790
791 if ($self->{_eof}) {
792 if ($self->{on_eof}) {
668 $self->{on_eof}($self) 793 $self->{on_eof}($self)
669 if $self->{_eof} && $self->{on_eof}; 794 } else {
795 $self->_error (0, 1);
796 }
797 }
670 798
671 # may need to restart read watcher 799 # may need to restart read watcher
672 unless ($self->{_rw}) { 800 unless ($self->{_rw}) {
673 $self->start_read 801 $self->start_read
674 if $self->{on_read} || @{ $self->{_queue} }; 802 if $self->{on_read} || @{ $self->{_queue} };
685 813
686sub on_read { 814sub on_read {
687 my ($self, $cb) = @_; 815 my ($self, $cb) = @_;
688 816
689 $self->{on_read} = $cb; 817 $self->{on_read} = $cb;
818 $self->_drain_rbuf if $cb && !$self->{_in_drain};
690} 819}
691 820
692=item $handle->rbuf 821=item $handle->rbuf
693 822
694Returns the read buffer (as a modifiable lvalue). 823Returns the read buffer (as a modifiable lvalue).
743 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 872 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
744 ->($self, $cb, @_); 873 ->($self, $cb, @_);
745 } 874 }
746 875
747 push @{ $self->{_queue} }, $cb; 876 push @{ $self->{_queue} }, $cb;
748 $self->_drain_rbuf; 877 $self->_drain_rbuf unless $self->{_in_drain};
749} 878}
750 879
751sub unshift_read { 880sub unshift_read {
752 my $self = shift; 881 my $self = shift;
753 my $cb = pop; 882 my $cb = pop;
759 ->($self, $cb, @_); 888 ->($self, $cb, @_);
760 } 889 }
761 890
762 891
763 unshift @{ $self->{_queue} }, $cb; 892 unshift @{ $self->{_queue} }, $cb;
764 $self->_drain_rbuf; 893 $self->_drain_rbuf unless $self->{_in_drain};
765} 894}
766 895
767=item $handle->push_read (type => @args, $cb) 896=item $handle->push_read (type => @args, $cb)
768 897
769=item $handle->unshift_read (type => @args, $cb) 898=item $handle->unshift_read (type => @args, $cb)
799 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 928 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
800 1 929 1
801 } 930 }
802}; 931};
803 932
804# compatibility with older API
805sub push_read_chunk {
806 $_[0]->push_read (chunk => $_[1], $_[2]);
807}
808
809sub unshift_read_chunk {
810 $_[0]->unshift_read (chunk => $_[1], $_[2]);
811}
812
813=item line => [$eol, ]$cb->($handle, $line, $eol) 933=item line => [$eol, ]$cb->($handle, $line, $eol)
814 934
815The callback will be called only once a full line (including the end of 935The callback will be called only once a full line (including the end of
816line marker, C<$eol>) has been read. This line (excluding the end of line 936line marker, C<$eol>) has been read. This line (excluding the end of line
817marker) will be passed to the callback as second argument (C<$line>), and 937marker) will be passed to the callback as second argument (C<$line>), and
832=cut 952=cut
833 953
834register_read_type line => sub { 954register_read_type line => sub {
835 my ($self, $cb, $eol) = @_; 955 my ($self, $cb, $eol) = @_;
836 956
837 $eol = qr|(\015?\012)| if @_ < 3; 957 if (@_ < 3) {
838 $eol = quotemeta $eol unless ref $eol; 958 # this is more than twice as fast as the generic code below
839 $eol = qr|^(.*?)($eol)|s;
840
841 sub { 959 sub {
842 $_[0]{rbuf} =~ s/$eol// or return; 960 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
843 961
844 $cb->($_[0], $1, $2); 962 $cb->($_[0], $1, $2);
845 1
846 }
847};
848
849# compatibility with older API
850sub push_read_line {
851 my $self = shift;
852 $self->push_read (line => @_);
853}
854
855sub unshift_read_line {
856 my $self = shift;
857 $self->unshift_read (line => @_);
858}
859
860=item netstring => $cb->($handle, $string)
861
862A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
863
864Throws an error with C<$!> set to EBADMSG on format violations.
865
866=cut
867
868register_read_type netstring => sub {
869 my ($self, $cb) = @_;
870
871 sub {
872 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
873 if ($_[0]{rbuf} =~ /[^0-9]/) {
874 $self->_error (&Errno::EBADMSG);
875 } 963 1
876 return;
877 } 964 }
965 } else {
966 $eol = quotemeta $eol unless ref $eol;
967 $eol = qr|^(.*?)($eol)|s;
878 968
879 my $len = $1; 969 sub {
970 $_[0]{rbuf} =~ s/$eol// or return;
880 971
881 $self->unshift_read (chunk => $len, sub { 972 $cb->($_[0], $1, $2);
882 my $string = $_[1];
883 $_[0]->unshift_read (chunk => 1, sub {
884 if ($_[1] eq ",") {
885 $cb->($_[0], $string);
886 } else {
887 $self->_error (&Errno::EBADMSG);
888 }
889 }); 973 1
890 }); 974 }
891
892 1
893 } 975 }
894}; 976};
895 977
896=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 978=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
897 979
961 1043
962 () 1044 ()
963 } 1045 }
964}; 1046};
965 1047
1048=item netstring => $cb->($handle, $string)
1049
1050A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1051
1052Throws an error with C<$!> set to EBADMSG on format violations.
1053
1054=cut
1055
1056register_read_type netstring => sub {
1057 my ($self, $cb) = @_;
1058
1059 sub {
1060 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1061 if ($_[0]{rbuf} =~ /[^0-9]/) {
1062 $self->_error (&Errno::EBADMSG);
1063 }
1064 return;
1065 }
1066
1067 my $len = $1;
1068
1069 $self->unshift_read (chunk => $len, sub {
1070 my $string = $_[1];
1071 $_[0]->unshift_read (chunk => 1, sub {
1072 if ($_[1] eq ",") {
1073 $cb->($_[0], $string);
1074 } else {
1075 $self->_error (&Errno::EBADMSG);
1076 }
1077 });
1078 });
1079
1080 1
1081 }
1082};
1083
1084=item packstring => $format, $cb->($handle, $string)
1085
1086An octet string prefixed with an encoded length. The encoding C<$format>
1087uses the same format as a Perl C<pack> format, but must specify a single
1088integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1089optional C<!>, C<< < >> or C<< > >> modifier).
1090
1091DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1092
1093Example: read a block of data prefixed by its length in BER-encoded
1094format (very efficient).
1095
1096 $handle->push_read (packstring => "w", sub {
1097 my ($handle, $data) = @_;
1098 });
1099
1100=cut
1101
1102register_read_type packstring => sub {
1103 my ($self, $cb, $format) = @_;
1104
1105 sub {
1106 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1107 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1108 or return;
1109
1110 $format = length pack $format, $len;
1111
1112 # bypass unshift if we already have the remaining chunk
1113 if ($format + $len <= length $_[0]{rbuf}) {
1114 my $data = substr $_[0]{rbuf}, $format, $len;
1115 substr $_[0]{rbuf}, 0, $format + $len, "";
1116 $cb->($_[0], $data);
1117 } else {
1118 # remove prefix
1119 substr $_[0]{rbuf}, 0, $format, "";
1120
1121 # read remaining chunk
1122 $_[0]->unshift_read (chunk => $len, $cb);
1123 }
1124
1125 1
1126 }
1127};
1128
966=item json => $cb->($handle, $hash_or_arrayref) 1129=item json => $cb->($handle, $hash_or_arrayref)
967 1130
968Reads a JSON object or array, decodes it and passes it to the callback. 1131Reads a JSON object or array, decodes it and passes it to the callback.
969 1132
970If a C<json> object was passed to the constructor, then that will be used 1133If a C<json> object was passed to the constructor, then that will be used
980the C<json> write type description, above, for an actual example. 1143the C<json> write type description, above, for an actual example.
981 1144
982=cut 1145=cut
983 1146
984register_read_type json => sub { 1147register_read_type json => sub {
985 my ($self, $cb, $accept, $reject, $skip) = @_; 1148 my ($self, $cb) = @_;
986 1149
987 require JSON; 1150 require JSON;
988 1151
989 my $data; 1152 my $data;
990 my $rbuf = \$self->{rbuf}; 1153 my $rbuf = \$self->{rbuf};
1005 () 1168 ()
1006 } 1169 }
1007 } 1170 }
1008}; 1171};
1009 1172
1173=item storable => $cb->($handle, $ref)
1174
1175Deserialises a L<Storable> frozen representation as written by the
1176C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1177data).
1178
1179Raises C<EBADMSG> error if the data could not be decoded.
1180
1181=cut
1182
1183register_read_type storable => sub {
1184 my ($self, $cb) = @_;
1185
1186 require Storable;
1187
1188 sub {
1189 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1190 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1191 or return;
1192
1193 my $format = length pack "w", $len;
1194
1195 # bypass unshift if we already have the remaining chunk
1196 if ($format + $len <= length $_[0]{rbuf}) {
1197 my $data = substr $_[0]{rbuf}, $format, $len;
1198 substr $_[0]{rbuf}, 0, $format + $len, "";
1199 $cb->($_[0], Storable::thaw ($data));
1200 } else {
1201 # remove prefix
1202 substr $_[0]{rbuf}, 0, $format, "";
1203
1204 # read remaining chunk
1205 $_[0]->unshift_read (chunk => $len, sub {
1206 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1207 $cb->($_[0], $ref);
1208 } else {
1209 $self->_error (&Errno::EBADMSG);
1210 }
1211 });
1212 }
1213
1214 1
1215 }
1216};
1217
1010=back 1218=back
1011 1219
1012=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1220=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1013 1221
1014This function (not method) lets you add your own types to C<push_read>. 1222This function (not method) lets you add your own types to C<push_read>.
1032=item $handle->stop_read 1240=item $handle->stop_read
1033 1241
1034=item $handle->start_read 1242=item $handle->start_read
1035 1243
1036In rare cases you actually do not want to read anything from the 1244In rare cases you actually do not want to read anything from the
1037socket. In this case you can call C<stop_read>. Neither C<on_read> no 1245socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1038any queued callbacks will be executed then. To start reading again, call 1246any queued callbacks will be executed then. To start reading again, call
1039C<start_read>. 1247C<start_read>.
1248
1249Note that AnyEvent::Handle will automatically C<start_read> for you when
1250you change the C<on_read> callback or push/unshift a read callback, and it
1251will automatically C<stop_read> for you when neither C<on_read> is set nor
1252there are any read requests in the queue.
1040 1253
1041=cut 1254=cut
1042 1255
1043sub stop_read { 1256sub stop_read {
1044 my ($self) = @_; 1257 my ($self) = @_;
1059 if ($len > 0) { 1272 if ($len > 0) {
1060 $self->{_activity} = AnyEvent->now; 1273 $self->{_activity} = AnyEvent->now;
1061 1274
1062 $self->{filter_r} 1275 $self->{filter_r}
1063 ? $self->{filter_r}($self, $rbuf) 1276 ? $self->{filter_r}($self, $rbuf)
1064 : $self->_drain_rbuf; 1277 : $self->{_in_drain} || $self->_drain_rbuf;
1065 1278
1066 } elsif (defined $len) { 1279 } elsif (defined $len) {
1067 delete $self->{_rw}; 1280 delete $self->{_rw};
1068 $self->{_eof} = 1; 1281 $self->{_eof} = 1;
1069 $self->_drain_rbuf; 1282 $self->_drain_rbuf unless $self->{_in_drain};
1070 1283
1071 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1284 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1072 return $self->_error ($!, 1); 1285 return $self->_error ($!, 1);
1073 } 1286 }
1074 }); 1287 });
1076} 1289}
1077 1290
1078sub _dotls { 1291sub _dotls {
1079 my ($self) = @_; 1292 my ($self) = @_;
1080 1293
1294 my $buf;
1295
1081 if (length $self->{_tls_wbuf}) { 1296 if (length $self->{_tls_wbuf}) {
1082 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1297 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1083 substr $self->{_tls_wbuf}, 0, $len, ""; 1298 substr $self->{_tls_wbuf}, 0, $len, "";
1084 } 1299 }
1085 } 1300 }
1086 1301
1087 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1302 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1088 $self->{wbuf} .= $buf; 1303 $self->{wbuf} .= $buf;
1089 $self->_drain_wbuf; 1304 $self->_drain_wbuf;
1090 } 1305 }
1091 1306
1092 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1307 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1308 if (length $buf) {
1093 $self->{rbuf} .= $buf; 1309 $self->{rbuf} .= $buf;
1094 $self->_drain_rbuf; 1310 $self->_drain_rbuf unless $self->{_in_drain};
1311 } else {
1312 # let's treat SSL-eof as we treat normal EOF
1313 $self->{_eof} = 1;
1314 $self->_shutdown;
1315 return;
1316 }
1095 } 1317 }
1096 1318
1097 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1319 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1098 1320
1099 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1321 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1185 1407
1186sub DESTROY { 1408sub DESTROY {
1187 my $self = shift; 1409 my $self = shift;
1188 1410
1189 $self->stoptls; 1411 $self->stoptls;
1412
1413 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1414
1415 if ($linger && length $self->{wbuf}) {
1416 my $fh = delete $self->{fh};
1417 my $wbuf = delete $self->{wbuf};
1418
1419 my @linger;
1420
1421 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1422 my $len = syswrite $fh, $wbuf, length $wbuf;
1423
1424 if ($len > 0) {
1425 substr $wbuf, 0, $len, "";
1426 } else {
1427 @linger = (); # end
1428 }
1429 });
1430 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1431 @linger = ();
1432 });
1433 }
1190} 1434}
1191 1435
1192=item AnyEvent::Handle::TLS_CTX 1436=item AnyEvent::Handle::TLS_CTX
1193 1437
1194This function creates and returns the Net::SSLeay::CTX object used by 1438This function creates and returns the Net::SSLeay::CTX object used by
1236=over 4 1480=over 4
1237 1481
1238=item * all constructor arguments become object members. 1482=item * all constructor arguments become object members.
1239 1483
1240At least initially, when you pass a C<tls>-argument to the constructor it 1484At least initially, when you pass a C<tls>-argument to the constructor it
1241will end up in C<< $handle->{tls} >>. Those members might be changes or 1485will end up in C<< $handle->{tls} >>. Those members might be changed or
1242mutated later on (for example C<tls> will hold the TLS connection object). 1486mutated later on (for example C<tls> will hold the TLS connection object).
1243 1487
1244=item * other object member names are prefixed with an C<_>. 1488=item * other object member names are prefixed with an C<_>.
1245 1489
1246All object members not explicitly documented (internal use) are prefixed 1490All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines