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.49 by root, Thu May 29 03:45:37 2008 UTC vs.
Revision 1.86 by root, Thu Aug 21 20:41:16 2008 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines