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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines