ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines