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.41 by root, Tue May 27 05:47:36 2008 UTC vs.
Revision 1.82 by root, Thu Aug 21 18:45:16 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(WSAWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw(EAGAIN EINTR);
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = '0.04'; 19our $VERSION = 4.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;
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 detected,
81i.e. in the case of a socket, when the other side has closed the
82connection cleanly.
81 83
84For sockets, this just means that the other side has stopped sending data,
85you can still try to write data, and, in fact, one can return from the eof
86callback and continue writing data, as only the read part has been shut
87down.
88
82While not mandatory, it is highly recommended to set an eof callback, 89While not mandatory, it is I<highly> recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still 90otherwise you might end up with a closed socket while you are still
84waiting for data. 91waiting for data.
85 92
93If an EOF condition has been detected but no C<on_eof> callback has been
94set, then a fatal error will be raised with C<$!> set to <0>.
95
86=item on_error => $cb->($handle) 96=item on_error => $cb->($handle, $fatal)
87 97
88This is the fatal error callback, that is called when, well, a fatal error 98This is the error callback, which is called when, well, some error
89occurs, such as not being able to resolve the hostname, failure to connect 99occured, such as not being able to resolve the hostname, failure to
90or a read error. 100connect or a read error.
91 101
92The object will not be in a usable state when this callback has been 102Some errors are fatal (which is indicated by C<$fatal> being true). On
93called. 103fatal errors the handle object will be shut down and will not be usable
104(but you are free to look at the current C< ->rbuf >). Examples of fatal
105errors are an EOF condition with active (but unsatisifable) read watchers
106(C<EPIPE>) or I/O errors.
107
108Non-fatal errors can be retried by simply returning, but it is recommended
109to simply ignore this parameter and instead abondon the handle object
110when this callback is invoked. Examples of non-fatal errors are timeouts
111C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
94 112
95On callback entrance, the value of C<$!> contains the operating system 113On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>). 114error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100 115
101While not mandatory, it is I<highly> recommended to set this callback, as 116While not mandatory, it is I<highly> recommended to set this callback, as
102you will not be notified of errors otherwise. The default simply calls 117you will not be notified of errors otherwise. The default simply calls
103die. 118C<croak>.
104 119
105=item on_read => $cb->($handle) 120=item on_read => $cb->($handle)
106 121
107This sets the default read callback, which is called when data arrives 122This sets the default read callback, which is called when data arrives
108and no read request is in the queue. 123and no read request is in the queue (unlike read queue callbacks, this
124callback will only be called when at least one octet of data is in the
125read buffer).
109 126
110To access (and remove data from) the read buffer, use the C<< ->rbuf >> 127To access (and remove data from) the read buffer, use the C<< ->rbuf >>
111method or access the C<$handle->{rbuf}> member directly. 128method or access the C<$handle->{rbuf}> member directly.
112 129
113When an EOF condition is detected then AnyEvent::Handle will first try to 130When an EOF condition is detected then AnyEvent::Handle will first try to
119 136
120This sets the callback that is called when the write buffer becomes empty 137This 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). 138(or when the callback is set and the buffer is empty already).
122 139
123To append to the write buffer, use the C<< ->push_write >> method. 140To append to the write buffer, use the C<< ->push_write >> method.
141
142This callback is useful when you don't want to put all of your write data
143into the queue at once, for example, when you want to write the contents
144of some file to the socket you might not want to read the whole file into
145memory and push it into the queue, but instead only read more data from
146the file when the write queue becomes empty.
147
148=item timeout => $fractional_seconds
149
150If non-zero, then this enables an "inactivity" timeout: whenever this many
151seconds pass without a successful read or write on the underlying file
152handle, the C<on_timeout> callback will be invoked (and if that one is
153missing, an C<ETIMEDOUT> error will be raised).
154
155Note that timeout processing is also active when you currently do not have
156any outstanding read or write requests: If you plan to keep the connection
157idle then you should disable the timout temporarily or ignore the timeout
158in the C<on_timeout> callback.
159
160Zero (the default) disables this timeout.
161
162=item on_timeout => $cb->($handle)
163
164Called whenever the inactivity timeout passes. If you return from this
165callback, then the timeout will be reset as if some activity had happened,
166so this condition is not fatal in any way.
124 167
125=item rbuf_max => <bytes> 168=item rbuf_max => <bytes>
126 169
127If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 170If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
128when the read buffer ever (strictly) exceeds this size. This is useful to 171when the read buffer ever (strictly) exceeds this size. This is useful to
132be configured to accept only so-and-so much data that it cannot act on 175be configured to accept only so-and-so much data that it cannot act on
133(for example, when expecting a line, an attacker could send an unlimited 176(for example, when expecting a line, an attacker could send an unlimited
134amount of data without a callback ever being called as long as the line 177amount of data without a callback ever being called as long as the line
135isn't finished). 178isn't finished).
136 179
180=item autocork => <boolean>
181
182When disabled (the default), then C<push_write> will try to immediately
183write the data to the handle if possible. This avoids having to register
184a write watcher and wait for the next event loop iteration, but can be
185inefficient if you write multiple small chunks (this disadvantage is
186usually avoided by your kernel's nagle algorithm, see C<low_delay>).
187
188When enabled, then writes will always be queued till the next event loop
189iteration. This is efficient when you do many small writes per iteration,
190but less efficient when you do a single write only.
191
192=item no_delay => <boolean>
193
194When doing small writes on sockets, your operating system kernel might
195wait a bit for more data before actually sending it out. This is called
196the Nagle algorithm, and usually it is beneficial.
197
198In some situations you want as low a delay as possible, which cna be
199accomplishd by setting this option to true.
200
201The default is your opertaing system's default behaviour, this option
202explicitly enables or disables it, if possible.
203
137=item read_size => <bytes> 204=item read_size => <bytes>
138 205
139The default read block size (the amount of bytes this module will try to read 206The default read block size (the amount of bytes this module will try to read
140on each [loop iteration). Default: C<4096>. 207during each (loop iteration). Default: C<8192>.
141 208
142=item low_water_mark => <bytes> 209=item low_water_mark => <bytes>
143 210
144Sets the amount of bytes (default: C<0>) that make up an "empty" write 211Sets the amount of bytes (default: C<0>) that make up an "empty" write
145buffer: If the write reaches this size or gets even samller it is 212buffer: If the write reaches this size or gets even samller it is
146considered empty. 213considered empty.
214
215=item linger => <seconds>
216
217If non-zero (default: C<3600>), then the destructor of the
218AnyEvent::Handle object will check wether there is still outstanding write
219data and will install a watcher that will write out this data. No errors
220will be reported (this mostly matches how the operating system treats
221outstanding data at socket close time).
222
223This will not work for partial TLS data that could not yet been
224encoded. This data will be lost.
147 225
148=item tls => "accept" | "connect" | Net::SSLeay::SSL object 226=item tls => "accept" | "connect" | Net::SSLeay::SSL object
149 227
150When this parameter is given, it enables TLS (SSL) mode, that means it 228When this parameter is given, it enables TLS (SSL) mode, that means it
151will start making tls handshake and will transparently encrypt/decrypt 229will start making tls handshake and will transparently encrypt/decrypt
160You can also provide your own TLS connection object, but you have 238You can also provide your own TLS connection object, but you have
161to make sure that you call either C<Net::SSLeay::set_connect_state> 239to make sure that you call either C<Net::SSLeay::set_connect_state>
162or C<Net::SSLeay::set_accept_state> on it before you pass it to 240or C<Net::SSLeay::set_accept_state> on it before you pass it to
163AnyEvent::Handle. 241AnyEvent::Handle.
164 242
165See the C<starttls> method if you need to start TLs negotiation later. 243See the C<starttls> method if you need to start TLS negotiation later.
166 244
167=item tls_ctx => $ssl_ctx 245=item tls_ctx => $ssl_ctx
168 246
169Use the given Net::SSLeay::CTX object to create the new TLS connection 247Use the given Net::SSLeay::CTX object to create the new TLS connection
170(unless a connection object was specified directly). If this parameter is 248(unless a connection object was specified directly). If this parameter is
202 if ($self->{tls}) { 280 if ($self->{tls}) {
203 require Net::SSLeay; 281 require Net::SSLeay;
204 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 282 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
205 } 283 }
206 284
207 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 285 $self->{_activity} = AnyEvent->now;
208 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 286 $self->_timeout;
287
209 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 288 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
210 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 289 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
211 290
212 $self->start_read; 291 $self->start_read
292 if $self->{on_read};
213 293
214 $self 294 $self
215} 295}
216 296
217sub _shutdown { 297sub _shutdown {
218 my ($self) = @_; 298 my ($self) = @_;
219 299
300 delete $self->{_tw};
220 delete $self->{_rw}; 301 delete $self->{_rw};
221 delete $self->{_ww}; 302 delete $self->{_ww};
222 delete $self->{fh}; 303 delete $self->{fh};
223}
224 304
305 $self->stoptls;
306
307 delete $self->{on_read};
308 delete $self->{_queue};
309}
310
225sub error { 311sub _error {
226 my ($self) = @_; 312 my ($self, $errno, $fatal) = @_;
227 313
228 {
229 local $!;
230 $self->_shutdown; 314 $self->_shutdown
231 } 315 if $fatal;
232 316
233 $self->{on_error}($self) 317 $! = $errno;
318
234 if $self->{on_error}; 319 if ($self->{on_error}) {
235 320 $self->{on_error}($self, $fatal);
321 } else {
236 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 322 Carp::croak "AnyEvent::Handle uncaught error: $!";
323 }
237} 324}
238 325
239=item $fh = $handle->fh 326=item $fh = $handle->fh
240 327
241This method returns the file handle of the L<AnyEvent::Handle> object. 328This method returns the file handle of the L<AnyEvent::Handle> object.
260 347
261=cut 348=cut
262 349
263sub on_eof { 350sub on_eof {
264 $_[0]{on_eof} = $_[1]; 351 $_[0]{on_eof} = $_[1];
352}
353
354=item $handle->on_timeout ($cb)
355
356Replace the current C<on_timeout> callback, or disables the callback
357(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
358argument.
359
360=cut
361
362sub on_timeout {
363 $_[0]{on_timeout} = $_[1];
364}
365
366=item $handle->autocork ($boolean)
367
368Enables or disables the current autocork behaviour (see C<autocork>
369constructor argument).
370
371=cut
372
373=item $handle->no_delay ($boolean)
374
375Enables or disables the C<no_delay> setting (see constructor argument of
376the same name for details).
377
378=cut
379
380sub no_delay {
381 $_[0]{no_delay} = $_[1];
382
383 eval {
384 local $SIG{__DIE__};
385 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
386 };
387}
388
389#############################################################################
390
391=item $handle->timeout ($seconds)
392
393Configures (or disables) the inactivity timeout.
394
395=cut
396
397sub timeout {
398 my ($self, $timeout) = @_;
399
400 $self->{timeout} = $timeout;
401 $self->_timeout;
402}
403
404# reset the timeout watcher, as neccessary
405# also check for time-outs
406sub _timeout {
407 my ($self) = @_;
408
409 if ($self->{timeout}) {
410 my $NOW = AnyEvent->now;
411
412 # when would the timeout trigger?
413 my $after = $self->{_activity} + $self->{timeout} - $NOW;
414
415 # now or in the past already?
416 if ($after <= 0) {
417 $self->{_activity} = $NOW;
418
419 if ($self->{on_timeout}) {
420 $self->{on_timeout}($self);
421 } else {
422 $self->_error (&Errno::ETIMEDOUT);
423 }
424
425 # callback could have changed timeout value, optimise
426 return unless $self->{timeout};
427
428 # calculate new after
429 $after = $self->{timeout};
430 }
431
432 Scalar::Util::weaken $self;
433 return unless $self; # ->error could have destroyed $self
434
435 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
436 delete $self->{_tw};
437 $self->_timeout;
438 });
439 } else {
440 delete $self->{_tw};
441 }
265} 442}
266 443
267############################################################################# 444#############################################################################
268 445
269=back 446=back
316 my $len = syswrite $self->{fh}, $self->{wbuf}; 493 my $len = syswrite $self->{fh}, $self->{wbuf};
317 494
318 if ($len >= 0) { 495 if ($len >= 0) {
319 substr $self->{wbuf}, 0, $len, ""; 496 substr $self->{wbuf}, 0, $len, "";
320 497
498 $self->{_activity} = AnyEvent->now;
499
321 $self->{on_drain}($self) 500 $self->{on_drain}($self)
322 if $self->{low_water_mark} >= length $self->{wbuf} 501 if $self->{low_water_mark} >= length $self->{wbuf}
323 && $self->{on_drain}; 502 && $self->{on_drain};
324 503
325 delete $self->{_ww} unless length $self->{wbuf}; 504 delete $self->{_ww} unless length $self->{wbuf};
326 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 505 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
327 $self->error; 506 $self->_error ($!, 1);
328 } 507 }
329 }; 508 };
330 509
331 # try to write data immediately 510 # try to write data immediately
332 $cb->(); 511 $cb->() unless $self->{autocork};
333 512
334 # if still data left in wbuf, we need to poll 513 # if still data left in wbuf, we need to poll
335 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 514 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
336 if length $self->{wbuf}; 515 if length $self->{wbuf};
337 }; 516 };
352 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 531 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
353 ->($self, @_); 532 ->($self, @_);
354 } 533 }
355 534
356 if ($self->{filter_w}) { 535 if ($self->{filter_w}) {
357 $self->{filter_w}->($self, \$_[0]); 536 $self->{filter_w}($self, \$_[0]);
358 } else { 537 } else {
359 $self->{wbuf} .= $_[0]; 538 $self->{wbuf} .= $_[0];
360 $self->_drain_wbuf; 539 $self->_drain_wbuf;
361 } 540 }
362} 541}
363 542
364=item $handle->push_write (type => @args) 543=item $handle->push_write (type => @args)
365 544
366=item $handle->unshift_write (type => @args)
367
368Instead of formatting your data yourself, you can also let this module do 545Instead of formatting your data yourself, you can also let this module do
369the job by specifying a type and type-specific arguments. 546the job by specifying a type and type-specific arguments.
370 547
371Predefined types are (if you have ideas for additional types, feel free to 548Predefined types are (if you have ideas for additional types, feel free to
372drop by and tell us): 549drop by and tell us):
376=item netstring => $string 553=item netstring => $string
377 554
378Formats the given value as netstring 555Formats the given value as netstring
379(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 556(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
380 557
381=back
382
383=cut 558=cut
384 559
385register_write_type netstring => sub { 560register_write_type netstring => sub {
386 my ($self, $string) = @_; 561 my ($self, $string) = @_;
387 562
388 sprintf "%d:%s,", (length $string), $string 563 sprintf "%d:%s,", (length $string), $string
564};
565
566=item packstring => $format, $data
567
568An octet string prefixed with an encoded length. The encoding C<$format>
569uses the same format as a Perl C<pack> format, but must specify a single
570integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
571optional C<!>, C<< < >> or C<< > >> modifier).
572
573=cut
574
575register_write_type packstring => sub {
576 my ($self, $format, $string) = @_;
577
578 pack "$format/a*", $string
389}; 579};
390 580
391=item json => $array_or_hashref 581=item json => $array_or_hashref
392 582
393Encodes the given hash or array reference into a JSON object. Unless you 583Encodes the given hash or array reference into a JSON object. Unless you
427 617
428 $self->{json} ? $self->{json}->encode ($ref) 618 $self->{json} ? $self->{json}->encode ($ref)
429 : JSON::encode_json ($ref) 619 : JSON::encode_json ($ref)
430}; 620};
431 621
622=item storable => $reference
623
624Freezes the given reference using L<Storable> and writes it to the
625handle. Uses the C<nfreeze> format.
626
627=cut
628
629register_write_type storable => sub {
630 my ($self, $ref) = @_;
631
632 require Storable;
633
634 pack "w/a*", Storable::nfreeze ($ref)
635};
636
637=back
638
432=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 639=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
433 640
434This function (not method) lets you add your own types to C<push_write>. 641This function (not method) lets you add your own types to C<push_write>.
435Whenever the given C<type> is used, C<push_write> will invoke the code 642Whenever the given C<type> is used, C<push_write> will invoke the code
436reference with the handle object and the remaining arguments. 643reference with the handle object and the remaining arguments.
456ways, the "simple" way, using only C<on_read> and the "complex" way, using 663ways, the "simple" way, using only C<on_read> and the "complex" way, using
457a queue. 664a queue.
458 665
459In the simple case, you just install an C<on_read> callback and whenever 666In the simple case, you just install an C<on_read> callback and whenever
460new data arrives, it will be called. You can then remove some data (if 667new data arrives, it will be called. You can then remove some data (if
461enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 668enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
462or not. 669leave the data there if you want to accumulate more (e.g. when only a
670partial message has been received so far).
463 671
464In the more complex case, you want to queue multiple callbacks. In this 672In the more complex case, you want to queue multiple callbacks. In this
465case, AnyEvent::Handle will call the first queued callback each time new 673case, AnyEvent::Handle will call the first queued callback each time new
466data arrives and removes it when it has done its job (see C<push_read>, 674data arrives (also the first time it is queued) and removes it when it has
467below). 675done its job (see C<push_read>, below).
468 676
469This way you can, for example, push three line-reads, followed by reading 677This way you can, for example, push three line-reads, followed by reading
470a chunk of data, and AnyEvent::Handle will execute them in order. 678a chunk of data, and AnyEvent::Handle will execute them in order.
471 679
472Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 680Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
473the specified number of bytes which give an XML datagram. 681the specified number of bytes which give an XML datagram.
474 682
475 # in the default state, expect some header bytes 683 # in the default state, expect some header bytes
476 $handle->on_read (sub { 684 $handle->on_read (sub {
477 # some data is here, now queue the length-header-read (4 octets) 685 # some data is here, now queue the length-header-read (4 octets)
478 shift->unshift_read_chunk (4, sub { 686 shift->unshift_read (chunk => 4, sub {
479 # header arrived, decode 687 # header arrived, decode
480 my $len = unpack "N", $_[1]; 688 my $len = unpack "N", $_[1];
481 689
482 # now read the payload 690 # now read the payload
483 shift->unshift_read_chunk ($len, sub { 691 shift->unshift_read (chunk => $len, sub {
484 my $xml = $_[1]; 692 my $xml = $_[1];
485 # handle xml 693 # handle xml
486 }); 694 });
487 }); 695 });
488 }); 696 });
489 697
490Example 2: Implement a client for a protocol that replies either with 698Example 2: Implement a client for a protocol that replies either with "OK"
491"OK" and another line or "ERROR" for one request, and 64 bytes for the 699and another line or "ERROR" for the first request that is sent, and 64
492second request. Due tot he availability of a full queue, we can just 700bytes for the second request. Due to the availability of a queue, we can
493pipeline sending both requests and manipulate the queue as necessary in 701just pipeline sending both requests and manipulate the queue as necessary
494the callbacks: 702in the callbacks.
495 703
496 # request one 704When the first callback is called and sees an "OK" response, it will
705C<unshift> another line-read. This line-read will be queued I<before> the
70664-byte chunk callback.
707
708 # request one, returns either "OK + extra line" or "ERROR"
497 $handle->push_write ("request 1\015\012"); 709 $handle->push_write ("request 1\015\012");
498 710
499 # we expect "ERROR" or "OK" as response, so push a line read 711 # we expect "ERROR" or "OK" as response, so push a line read
500 $handle->push_read_line (sub { 712 $handle->push_read (line => sub {
501 # if we got an "OK", we have to _prepend_ another line, 713 # if we got an "OK", we have to _prepend_ another line,
502 # so it will be read before the second request reads its 64 bytes 714 # so it will be read before the second request reads its 64 bytes
503 # which are already in the queue when this callback is called 715 # which are already in the queue when this callback is called
504 # we don't do this in case we got an error 716 # we don't do this in case we got an error
505 if ($_[1] eq "OK") { 717 if ($_[1] eq "OK") {
506 $_[0]->unshift_read_line (sub { 718 $_[0]->unshift_read (line => sub {
507 my $response = $_[1]; 719 my $response = $_[1];
508 ... 720 ...
509 }); 721 });
510 } 722 }
511 }); 723 });
512 724
513 # request two 725 # request two, simply returns 64 octets
514 $handle->push_write ("request 2\015\012"); 726 $handle->push_write ("request 2\015\012");
515 727
516 # simply read 64 bytes, always 728 # simply read 64 bytes, always
517 $handle->push_read_chunk (64, sub { 729 $handle->push_read (chunk => 64, sub {
518 my $response = $_[1]; 730 my $response = $_[1];
519 ... 731 ...
520 }); 732 });
521 733
522=over 4 734=over 4
523 735
524=cut 736=cut
525 737
526sub _drain_rbuf { 738sub _drain_rbuf {
527 my ($self) = @_; 739 my ($self) = @_;
740
741 local $self->{_in_drain} = 1;
528 742
529 if ( 743 if (
530 defined $self->{rbuf_max} 744 defined $self->{rbuf_max}
531 && $self->{rbuf_max} < length $self->{rbuf} 745 && $self->{rbuf_max} < length $self->{rbuf}
532 ) { 746 ) {
533 $! = &Errno::ENOSPC; 747 $self->_error (&Errno::ENOSPC, 1), return;
534 $self->error;
535 } 748 }
536 749
537 return if $self->{in_drain}; 750 while () {
538 local $self->{in_drain} = 1;
539
540 while (my $len = length $self->{rbuf}) { 751 my $len = length $self->{rbuf};
541 no strict 'refs'; 752
542 if (my $cb = shift @{ $self->{_queue} }) { 753 if (my $cb = shift @{ $self->{_queue} }) {
543 unless ($cb->($self)) { 754 unless ($cb->($self)) {
544 if ($self->{_eof}) { 755 if ($self->{_eof}) {
545 # no progress can be made (not enough data and no data forthcoming) 756 # no progress can be made (not enough data and no data forthcoming)
546 $! = &Errno::EPIPE; 757 $self->_error (&Errno::EPIPE, 1), return;
547 $self->error;
548 } 758 }
549 759
550 unshift @{ $self->{_queue} }, $cb; 760 unshift @{ $self->{_queue} }, $cb;
551 return; 761 last;
552 } 762 }
553 } elsif ($self->{on_read}) { 763 } elsif ($self->{on_read}) {
764 last unless $len;
765
554 $self->{on_read}($self); 766 $self->{on_read}($self);
555 767
556 if ( 768 if (
557 $self->{_eof} # if no further data will arrive
558 && $len == length $self->{rbuf} # and no data has been consumed 769 $len == length $self->{rbuf} # if no data has been consumed
559 && !@{ $self->{_queue} } # and the queue is still empty 770 && !@{ $self->{_queue} } # and the queue is still empty
560 && $self->{on_read} # and we still want to read data 771 && $self->{on_read} # but we still have on_read
561 ) { 772 ) {
773 # no further data will arrive
562 # then no progress can be made 774 # so no progress can be made
563 $! = &Errno::EPIPE; 775 $self->_error (&Errno::EPIPE, 1), return
564 $self->error; 776 if $self->{_eof};
777
778 last; # more data might arrive
565 } 779 }
566 } else { 780 } else {
567 # read side becomes idle 781 # read side becomes idle
568 delete $self->{_rw}; 782 delete $self->{_rw};
569 return; 783 last;
570 } 784 }
571 } 785 }
572 786
573 if ($self->{_eof}) { 787 if ($self->{_eof}) {
574 $self->_shutdown; 788 if ($self->{on_eof}) {
575 $self->{on_eof}($self) 789 $self->{on_eof}($self)
576 if $self->{on_eof}; 790 } else {
791 $self->_error (0, 1);
792 }
793 }
794
795 # may need to restart read watcher
796 unless ($self->{_rw}) {
797 $self->start_read
798 if $self->{on_read} || @{ $self->{_queue} };
577 } 799 }
578} 800}
579 801
580=item $handle->on_read ($cb) 802=item $handle->on_read ($cb)
581 803
587 809
588sub on_read { 810sub on_read {
589 my ($self, $cb) = @_; 811 my ($self, $cb) = @_;
590 812
591 $self->{on_read} = $cb; 813 $self->{on_read} = $cb;
814 $self->_drain_rbuf if $cb && !$self->{_in_drain};
592} 815}
593 816
594=item $handle->rbuf 817=item $handle->rbuf
595 818
596Returns the read buffer (as a modifiable lvalue). 819Returns the read buffer (as a modifiable lvalue).
645 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 868 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
646 ->($self, $cb, @_); 869 ->($self, $cb, @_);
647 } 870 }
648 871
649 push @{ $self->{_queue} }, $cb; 872 push @{ $self->{_queue} }, $cb;
650 $self->_drain_rbuf; 873 $self->_drain_rbuf unless $self->{_in_drain};
651} 874}
652 875
653sub unshift_read { 876sub unshift_read {
654 my $self = shift; 877 my $self = shift;
655 my $cb = pop; 878 my $cb = pop;
661 ->($self, $cb, @_); 884 ->($self, $cb, @_);
662 } 885 }
663 886
664 887
665 unshift @{ $self->{_queue} }, $cb; 888 unshift @{ $self->{_queue} }, $cb;
666 $self->_drain_rbuf; 889 $self->_drain_rbuf unless $self->{_in_drain};
667} 890}
668 891
669=item $handle->push_read (type => @args, $cb) 892=item $handle->push_read (type => @args, $cb)
670 893
671=item $handle->unshift_read (type => @args, $cb) 894=item $handle->unshift_read (type => @args, $cb)
701 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 924 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
702 1 925 1
703 } 926 }
704}; 927};
705 928
706# compatibility with older API
707sub push_read_chunk {
708 $_[0]->push_read (chunk => $_[1], $_[2]);
709}
710
711sub unshift_read_chunk {
712 $_[0]->unshift_read (chunk => $_[1], $_[2]);
713}
714
715=item line => [$eol, ]$cb->($handle, $line, $eol) 929=item line => [$eol, ]$cb->($handle, $line, $eol)
716 930
717The callback will be called only once a full line (including the end of 931The callback will be called only once a full line (including the end of
718line marker, C<$eol>) has been read. This line (excluding the end of line 932line marker, C<$eol>) has been read. This line (excluding the end of line
719marker) will be passed to the callback as second argument (C<$line>), and 933marker) will be passed to the callback as second argument (C<$line>), and
734=cut 948=cut
735 949
736register_read_type line => sub { 950register_read_type line => sub {
737 my ($self, $cb, $eol) = @_; 951 my ($self, $cb, $eol) = @_;
738 952
739 $eol = qr|(\015?\012)| if @_ < 3; 953 if (@_ < 3) {
740 $eol = quotemeta $eol unless ref $eol; 954 # this is more than twice as fast as the generic code below
741 $eol = qr|^(.*?)($eol)|s;
742
743 sub { 955 sub {
744 $_[0]{rbuf} =~ s/$eol// or return; 956 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
745 957
746 $cb->($_[0], $1, $2); 958 $cb->($_[0], $1, $2);
747 1
748 }
749};
750
751# compatibility with older API
752sub push_read_line {
753 my $self = shift;
754 $self->push_read (line => @_);
755}
756
757sub unshift_read_line {
758 my $self = shift;
759 $self->unshift_read (line => @_);
760}
761
762=item netstring => $cb->($handle, $string)
763
764A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
765
766Throws an error with C<$!> set to EBADMSG on format violations.
767
768=cut
769
770register_read_type netstring => sub {
771 my ($self, $cb) = @_;
772
773 sub {
774 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
775 if ($_[0]{rbuf} =~ /[^0-9]/) {
776 $! = &Errno::EBADMSG;
777 $self->error;
778 } 959 1
779 return;
780 } 960 }
961 } else {
962 $eol = quotemeta $eol unless ref $eol;
963 $eol = qr|^(.*?)($eol)|s;
781 964
782 my $len = $1; 965 sub {
966 $_[0]{rbuf} =~ s/$eol// or return;
783 967
784 $self->unshift_read (chunk => $len, sub { 968 $cb->($_[0], $1, $2);
785 my $string = $_[1];
786 $_[0]->unshift_read (chunk => 1, sub {
787 if ($_[1] eq ",") {
788 $cb->($_[0], $string);
789 } else {
790 $! = &Errno::EBADMSG;
791 $self->error;
792 }
793 }); 969 1
794 }); 970 }
795
796 1
797 } 971 }
798}; 972};
799 973
800=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 974=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
801 975
853 return 1; 1027 return 1;
854 } 1028 }
855 1029
856 # reject 1030 # reject
857 if ($reject && $$rbuf =~ $reject) { 1031 if ($reject && $$rbuf =~ $reject) {
858 $! = &Errno::EBADMSG; 1032 $self->_error (&Errno::EBADMSG);
859 $self->error;
860 } 1033 }
861 1034
862 # skip 1035 # skip
863 if ($skip && $$rbuf =~ $skip) { 1036 if ($skip && $$rbuf =~ $skip) {
864 $data .= substr $$rbuf, 0, $+[0], ""; 1037 $data .= substr $$rbuf, 0, $+[0], "";
866 1039
867 () 1040 ()
868 } 1041 }
869}; 1042};
870 1043
1044=item netstring => $cb->($handle, $string)
1045
1046A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1047
1048Throws an error with C<$!> set to EBADMSG on format violations.
1049
1050=cut
1051
1052register_read_type netstring => sub {
1053 my ($self, $cb) = @_;
1054
1055 sub {
1056 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1057 if ($_[0]{rbuf} =~ /[^0-9]/) {
1058 $self->_error (&Errno::EBADMSG);
1059 }
1060 return;
1061 }
1062
1063 my $len = $1;
1064
1065 $self->unshift_read (chunk => $len, sub {
1066 my $string = $_[1];
1067 $_[0]->unshift_read (chunk => 1, sub {
1068 if ($_[1] eq ",") {
1069 $cb->($_[0], $string);
1070 } else {
1071 $self->_error (&Errno::EBADMSG);
1072 }
1073 });
1074 });
1075
1076 1
1077 }
1078};
1079
1080=item packstring => $format, $cb->($handle, $string)
1081
1082An octet string prefixed with an encoded length. The encoding C<$format>
1083uses the same format as a Perl C<pack> format, but must specify a single
1084integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1085optional C<!>, C<< < >> or C<< > >> modifier).
1086
1087DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1088
1089Example: read a block of data prefixed by its length in BER-encoded
1090format (very efficient).
1091
1092 $handle->push_read (packstring => "w", sub {
1093 my ($handle, $data) = @_;
1094 });
1095
1096=cut
1097
1098register_read_type packstring => sub {
1099 my ($self, $cb, $format) = @_;
1100
1101 sub {
1102 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1103 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1104 or return;
1105
1106 $format = length pack $format, $len;
1107
1108 # bypass unshift if we already have the remaining chunk
1109 if ($format + $len <= length $_[0]{rbuf}) {
1110 my $data = substr $_[0]{rbuf}, $format, $len;
1111 substr $_[0]{rbuf}, 0, $format + $len, "";
1112 $cb->($_[0], $data);
1113 } else {
1114 # remove prefix
1115 substr $_[0]{rbuf}, 0, $format, "";
1116
1117 # read remaining chunk
1118 $_[0]->unshift_read (chunk => $len, $cb);
1119 }
1120
1121 1
1122 }
1123};
1124
871=item json => $cb->($handle, $hash_or_arrayref) 1125=item json => $cb->($handle, $hash_or_arrayref)
872 1126
873Reads a JSON object or array, decodes it and passes it to the callback. 1127Reads a JSON object or array, decodes it and passes it to the callback.
874 1128
875If a C<json> object was passed to the constructor, then that will be used 1129If a C<json> object was passed to the constructor, then that will be used
885the C<json> write type description, above, for an actual example. 1139the C<json> write type description, above, for an actual example.
886 1140
887=cut 1141=cut
888 1142
889register_read_type json => sub { 1143register_read_type json => sub {
890 my ($self, $cb, $accept, $reject, $skip) = @_; 1144 my ($self, $cb) = @_;
891 1145
892 require JSON; 1146 require JSON;
893 1147
894 my $data; 1148 my $data;
895 my $rbuf = \$self->{rbuf}; 1149 my $rbuf = \$self->{rbuf};
910 () 1164 ()
911 } 1165 }
912 } 1166 }
913}; 1167};
914 1168
1169=item storable => $cb->($handle, $ref)
1170
1171Deserialises a L<Storable> frozen representation as written by the
1172C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1173data).
1174
1175Raises C<EBADMSG> error if the data could not be decoded.
1176
1177=cut
1178
1179register_read_type storable => sub {
1180 my ($self, $cb) = @_;
1181
1182 require Storable;
1183
1184 sub {
1185 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1186 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1187 or return;
1188
1189 my $format = length pack "w", $len;
1190
1191 # bypass unshift if we already have the remaining chunk
1192 if ($format + $len <= length $_[0]{rbuf}) {
1193 my $data = substr $_[0]{rbuf}, $format, $len;
1194 substr $_[0]{rbuf}, 0, $format + $len, "";
1195 $cb->($_[0], Storable::thaw ($data));
1196 } else {
1197 # remove prefix
1198 substr $_[0]{rbuf}, 0, $format, "";
1199
1200 # read remaining chunk
1201 $_[0]->unshift_read (chunk => $len, sub {
1202 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1203 $cb->($_[0], $ref);
1204 } else {
1205 $self->_error (&Errno::EBADMSG);
1206 }
1207 });
1208 }
1209
1210 1
1211 }
1212};
1213
915=back 1214=back
916 1215
917=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1216=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
918 1217
919This function (not method) lets you add your own types to C<push_read>. 1218This function (not method) lets you add your own types to C<push_read>.
937=item $handle->stop_read 1236=item $handle->stop_read
938 1237
939=item $handle->start_read 1238=item $handle->start_read
940 1239
941In rare cases you actually do not want to read anything from the 1240In rare cases you actually do not want to read anything from the
942socket. In this case you can call C<stop_read>. Neither C<on_read> no 1241socket. In this case you can call C<stop_read>. Neither C<on_read> nor
943any queued callbacks will be executed then. To start reading again, call 1242any queued callbacks will be executed then. To start reading again, call
944C<start_read>. 1243C<start_read>.
1244
1245Note that AnyEvent::Handle will automatically C<start_read> for you when
1246you change the C<on_read> callback or push/unshift a read callback, and it
1247will automatically C<stop_read> for you when neither C<on_read> is set nor
1248there are any read requests in the queue.
945 1249
946=cut 1250=cut
947 1251
948sub stop_read { 1252sub stop_read {
949 my ($self) = @_; 1253 my ($self) = @_;
960 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1264 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
961 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1265 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
962 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1266 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
963 1267
964 if ($len > 0) { 1268 if ($len > 0) {
1269 $self->{_activity} = AnyEvent->now;
1270
965 $self->{filter_r} 1271 $self->{filter_r}
966 ? $self->{filter_r}->($self, $rbuf) 1272 ? $self->{filter_r}($self, $rbuf)
967 : $self->_drain_rbuf; 1273 : $self->{_in_drain} || $self->_drain_rbuf;
968 1274
969 } elsif (defined $len) { 1275 } elsif (defined $len) {
970 delete $self->{_rw}; 1276 delete $self->{_rw};
971 $self->{_eof} = 1; 1277 $self->{_eof} = 1;
972 $self->_drain_rbuf; 1278 $self->_drain_rbuf unless $self->{_in_drain};
973 1279
974 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1280 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
975 return $self->error; 1281 return $self->_error ($!, 1);
976 } 1282 }
977 }); 1283 });
978 } 1284 }
979} 1285}
980 1286
981sub _dotls { 1287sub _dotls {
982 my ($self) = @_; 1288 my ($self) = @_;
1289
1290 my $buf;
983 1291
984 if (length $self->{_tls_wbuf}) { 1292 if (length $self->{_tls_wbuf}) {
985 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1293 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
986 substr $self->{_tls_wbuf}, 0, $len, ""; 1294 substr $self->{_tls_wbuf}, 0, $len, "";
987 } 1295 }
988 } 1296 }
989 1297
990 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1298 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
991 $self->{wbuf} .= $buf; 1299 $self->{wbuf} .= $buf;
992 $self->_drain_wbuf; 1300 $self->_drain_wbuf;
993 } 1301 }
994 1302
995 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1303 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1304 if (length $buf) {
996 $self->{rbuf} .= $buf; 1305 $self->{rbuf} .= $buf;
997 $self->_drain_rbuf; 1306 $self->_drain_rbuf unless $self->{_in_drain};
1307 } else {
1308 # let's treat SSL-eof as we treat normal EOF
1309 $self->{_eof} = 1;
1310 $self->_shutdown;
1311 return;
1312 }
998 } 1313 }
999 1314
1000 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1315 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1001 1316
1002 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1317 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1003 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1318 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1004 $self->error; 1319 return $self->_error ($!, 1);
1005 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1320 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1006 $! = &Errno::EIO; 1321 return $self->_error (&Errno::EIO, 1);
1007 $self->error;
1008 } 1322 }
1009 1323
1010 # all others are fine for our purposes 1324 # all others are fine for our purposes
1011 } 1325 }
1012} 1326}
1027call and can be used or changed to your liking. Note that the handshake 1341call and can be used or changed to your liking. Note that the handshake
1028might have already started when this function returns. 1342might have already started when this function returns.
1029 1343
1030=cut 1344=cut
1031 1345
1032# TODO: maybe document...
1033sub starttls { 1346sub starttls {
1034 my ($self, $ssl, $ctx) = @_; 1347 my ($self, $ssl, $ctx) = @_;
1035 1348
1036 $self->stoptls; 1349 $self->stoptls;
1037 1350
1090 1403
1091sub DESTROY { 1404sub DESTROY {
1092 my $self = shift; 1405 my $self = shift;
1093 1406
1094 $self->stoptls; 1407 $self->stoptls;
1408
1409 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1410
1411 if ($linger && length $self->{wbuf}) {
1412 my $fh = delete $self->{fh};
1413 my $wbuf = delete $self->{wbuf};
1414
1415 my @linger;
1416
1417 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1418 my $len = syswrite $fh, $wbuf, length $wbuf;
1419
1420 if ($len > 0) {
1421 substr $wbuf, 0, $len, "";
1422 } else {
1423 @linger = (); # end
1424 }
1425 });
1426 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1427 @linger = ();
1428 });
1429 }
1095} 1430}
1096 1431
1097=item AnyEvent::Handle::TLS_CTX 1432=item AnyEvent::Handle::TLS_CTX
1098 1433
1099This function creates and returns the Net::SSLeay::CTX object used by 1434This function creates and returns the Net::SSLeay::CTX object used by
1141=over 4 1476=over 4
1142 1477
1143=item * all constructor arguments become object members. 1478=item * all constructor arguments become object members.
1144 1479
1145At least initially, when you pass a C<tls>-argument to the constructor it 1480At least initially, when you pass a C<tls>-argument to the constructor it
1146will end up in C<< $handle->{tls} >>. Those members might be changes or 1481will end up in C<< $handle->{tls} >>. Those members might be changed or
1147mutated later on (for example C<tls> will hold the TLS connection object). 1482mutated later on (for example C<tls> will hold the TLS connection object).
1148 1483
1149=item * other object member names are prefixed with an C<_>. 1484=item * other object member names are prefixed with an C<_>.
1150 1485
1151All object members not explicitly documented (internal use) are prefixed 1486All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines