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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines